From 5a427bbd611db142e3a15bc2cd8b0ca8057e5f1e Mon Sep 17 00:00:00 2001 From: Jeremy Alvis Date: Tue, 30 Jun 2026 09:29:12 -0700 Subject: [PATCH 1/4] backfill task ids for tool messages to stop false stale-send guard Signed-off-by: Jeremy Alvis --- .../ChatInterface.sendGuard.test.tsx | 37 +++++++++++++++++++ ui/src/lib/messageHandlers.ts | 13 ++++++- 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx b/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx index b8efc1906..c309215b1 100644 --- a/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx +++ b/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx @@ -337,6 +337,43 @@ describe("ChatInterface send guard", () => { expect(mockSendMessageStream).toHaveBeenCalledTimes(1); }); + it("does not block the next send after a same-tab tool-call turn is persisted", async () => { + // Reproduces the false positive seen with tool calls: a same-tab turn that + // makes a tool call streams a ToolCallRequestEvent locally (keyed by its + // real contextId/taskId), but the backend persists agent messages with + // EMPTY contextId/taskId. extractMessagesFromTasks leaves those empty + // (`"" ?? task.contextId` is still "") and rebuilds the converted tool + // message with a fresh uuidv4 messageId, so it keys on ["message", ] + // — a different key on every extraction. The guard therefore can never match + // the persisted tool message against the local one, counts the backend as + // ahead, and falsely blocks the next send. + mockBackendTasks([completedTask("task-initial", initialTurn)]); + mockSendMessageStream + .mockResolvedValueOnce(streamOf(completedToolCallStatusEvent("session-1", "task-streamed", "shared-call"))) + .mockResolvedValueOnce(streamOf(completedStatusEvent("next answer", "session-1", "task-next"))); + + renderExistingSession(); + + expect(await screen.findByText("initial answer")).toBeInTheDocument(); + + await sendText("tool question"); + await waitFor(() => expect(mockSendMessageStream).toHaveBeenCalledTimes(1)); + + // Backend now reflects exactly that same-tab tool turn. The persisted agent + // tool message carries empty contextId/taskId, as the real backend stores it. + mockBackendTasks([ + completedTask("task-initial", initialTurn), + completedTask("task-streamed", [ + textMessage(sentMessage().messageId, "user", "tool question", "session-1", "task-streamed"), + toolCallMessage("backend-tool", "", "", "shared-call"), + ]), + ]); + await sendText("next question"); + + await waitFor(() => expect(mockSendMessageStream).toHaveBeenCalledTimes(2)); + expect(mockToastInfo).not.toHaveBeenCalledWith(staleToastMessage); + }); + it("still blocks when the backend has a cross-tab message not visible locally", async () => { mockBackendTasks([completedTask("task-initial", initialTurn)]); diff --git a/ui/src/lib/messageHandlers.ts b/ui/src/lib/messageHandlers.ts index df0be1659..f1f1e04d2 100644 --- a/ui/src/lib/messageHandlers.ts +++ b/ui/src/lib/messageHandlers.ts @@ -60,8 +60,17 @@ export function extractMessagesFromTasks(tasks: Task[]): Message[] { // Agent messages: convert function_call / function_response DataParts to // the same ToolCallRequestEvent / ToolCallExecutionEvent format produced // by the live-stream handlers so the rendering component can display them. - const msgContextId = historyItem.contextId ?? task.contextId; - const msgTaskId = historyItem.taskId ?? task.id; + // + // Backfill contextId/taskId from the task when the history item omits them. + // Persisted agent messages frequently carry empty strings here, and `??` + // would keep the empty string ("" is not nullish) — leaving the converted + // tool messages without a contextId/taskId. The send guard then can't key + // them on (contextId, taskId) and falls back to the regenerated uuid + // messageId, so the same backend tool message keys differently on every + // extraction and never matches its local counterpart (false stale-send + // block). Treat "" as absent so these messages get the task's stable ids. + const msgContextId = historyItem.contextId || task.contextId; + const msgTaskId = historyItem.taskId || task.id; const source = getSourceFromMetadata(historyItem.metadata as ADKMetadata | undefined, "assistant"); const msgStats = getMessageTokenStats(historyItem.metadata as Record); From 62ae37c2ecb573d127a2f2b64a7e545fc2cfdf52 Mon Sep 17 00:00:00 2001 From: Jeremy Alvis Date: Tue, 30 Jun 2026 09:38:40 -0700 Subject: [PATCH 2/4] stop false stale-send guard on tool-call turns Signed-off-by: Jeremy Alvis --- .../kagent/adk/converters/event_converter.py | 25 +++++++++++++++++-- .../converters/test_event_converter.py | 20 +++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/python/packages/kagent-adk/src/kagent/adk/converters/event_converter.py b/python/packages/kagent-adk/src/kagent/adk/converters/event_converter.py index 92c4b664d..fadd1e574 100644 --- a/python/packages/kagent-adk/src/kagent/adk/converters/event_converter.py +++ b/python/packages/kagent-adk/src/kagent/adk/converters/event_converter.py @@ -161,6 +161,8 @@ def convert_event_to_a2a_message( invocation_context: InvocationContext, role: Role = Role.agent, subagent_session_ids: Optional[Dict[str, str]] = None, + task_id: Optional[str] = None, + context_id: Optional[str] = None, ) -> Optional[Message]: """Converts an ADK event to an A2A message. @@ -171,6 +173,12 @@ def convert_event_to_a2a_message( subagent_session_ids: Optional mapping of tool name to pre-generated subagent session ID. When provided, function_call DataParts for matching tools will have the session ID stamped into their metadata. + task_id: Optional task ID stamped onto the message so it carries the + same identity as the enclosing task. A2A allows these to be omitted + (the task is the canonical carrier), but stamping them lets consumers + that flatten task.history into standalone messages key each message to + its task without backfilling. + context_id: Optional context ID stamped onto the message, as task_id. Returns: An A2A Message if the event has content, None otherwise. @@ -198,7 +206,14 @@ def convert_event_to_a2a_message( if a2a_parts: message_metadata = _get_context_metadata(event, invocation_context) - return Message(message_id=str(uuid.uuid4()), role=role, parts=a2a_parts, metadata=message_metadata) + return Message( + message_id=str(uuid.uuid4()), + role=role, + parts=a2a_parts, + metadata=message_metadata, + task_id=task_id, + context_id=context_id, + ) except Exception as e: logger.error("Failed to convert event to status message: %s", e) @@ -342,7 +357,13 @@ def convert_event_to_a2a_events( a2a_events.append(error_event) # Handle regular message content - message = convert_event_to_a2a_message(event, invocation_context, subagent_session_ids=subagent_session_ids) + message = convert_event_to_a2a_message( + event, + invocation_context, + subagent_session_ids=subagent_session_ids, + task_id=task_id, + context_id=context_id, + ) if message: running_event = _create_status_update_event(message, invocation_context, event, task_id, context_id) a2a_events.append(running_event) diff --git a/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py b/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py index 6040a7e04..0b3b7de3e 100644 --- a/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py +++ b/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py @@ -117,3 +117,23 @@ def test_convert_event_to_a2a_events(self): error_code_key = get_kagent_metadata_key("error_code") assert error_code_key in error_event.metadata assert error_event.metadata[error_code_key] == str(genai_types.FinishReason.MALFORMED_FUNCTION_CALL) + + def test_message_carries_task_and_context_ids(self): + """The converted message stamps task_id/context_id so consumers that + flatten task.history can key it to its task without backfilling.""" + invocation_context = _create_mock_invocation_context() + content = genai_types.Content(parts=[genai_types.Part(text="hello world")]) + event = _create_mock_event(content=content, invocation_id="test_invocation_ids") + + result = convert_event_to_a2a_events( + event, invocation_context, task_id="task-xyz", context_id="ctx-xyz" + ) + + working_events = [ + e for e in result if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.working + ] + assert len(working_events) == 1 + message = working_events[0].status.message + assert message is not None + assert message.task_id == "task-xyz" + assert message.context_id == "ctx-xyz" From a1136cee8fb3feb881767ed0c836b6b88b8b6d14 Mon Sep 17 00:00:00 2001 From: Jeremy Alvis Date: Tue, 30 Jun 2026 09:48:47 -0700 Subject: [PATCH 3/4] Fix python linting issue Signed-off-by: Jeremy Alvis --- .../tests/unittests/converters/test_event_converter.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py b/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py index 0b3b7de3e..62611648c 100644 --- a/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py +++ b/python/packages/kagent-adk/tests/unittests/converters/test_event_converter.py @@ -125,9 +125,7 @@ def test_message_carries_task_and_context_ids(self): content = genai_types.Content(parts=[genai_types.Part(text="hello world")]) event = _create_mock_event(content=content, invocation_id="test_invocation_ids") - result = convert_event_to_a2a_events( - event, invocation_context, task_id="task-xyz", context_id="ctx-xyz" - ) + result = convert_event_to_a2a_events(event, invocation_context, task_id="task-xyz", context_id="ctx-xyz") working_events = [ e for e in result if isinstance(e, TaskStatusUpdateEvent) and e.status.state == TaskState.working From cf049b58060b507caba6e192ab24e580d113729d Mon Sep 17 00:00:00 2001 From: Jeremy Alvis Date: Tue, 30 Jun 2026 10:50:53 -0700 Subject: [PATCH 4/4] carry context/task ids on all agent messages to stop false stale-send guard Signed-off-by: Jeremy Alvis --- go/adk/pkg/a2a/executor.go | 45 +++++++++---- go/adk/pkg/a2a/executor_test.go | 67 +++++++++++++++++++ .../ChatInterface.sendGuard.test.tsx | 36 ++++++++++ ui/src/lib/messageHandlers.ts | 29 ++++---- 4 files changed, 153 insertions(+), 24 deletions(-) create mode 100644 go/adk/pkg/a2a/executor_test.go diff --git a/go/adk/pkg/a2a/executor.go b/go/adk/pkg/a2a/executor.go index b9879988b..e92fc7df1 100644 --- a/go/adk/pkg/a2a/executor.go +++ b/go/adk/pkg/a2a/executor.go @@ -100,6 +100,33 @@ func (u *userIDInterceptor) Before(ctx context.Context, callCtx *a2asrv.CallCont return ctx, nil } +// newAgentMessage builds an agent message stamped with the request's context +// and task ids. A2A allows omitting these (the task is the canonical carrier), +// but stamping them lets consumers that flatten task.history into standalone +// messages key each message to its task without backfilling. Mirrors the Python +// kagent-adk event converter. +func newAgentMessage(reqCtx *a2asrv.RequestContext, parts ...a2atype.Part) *a2atype.Message { + msg := a2atype.NewMessage(a2atype.MessageRoleAgent, parts...) + msg.ContextID = reqCtx.ContextID + msg.TaskID = reqCtx.TaskID + return msg +} + +// newAgentStatusEvent builds a working TaskStatusUpdateEvent whose agent message +// carries the given parts, the given metadata, and the request's context/task +// ids (via newAgentMessage). The message and event share the same metadata map, +// matching the executor's emission paths. This is the per-event seam where a +// streamed agent message is turned into an emitted (and persisted) event, so the +// id stamping here is what the send guard relies on when it later flattens +// task.history. +func newAgentStatusEvent(reqCtx *a2asrv.RequestContext, parts a2atype.ContentParts, meta map[string]any) *a2atype.TaskStatusUpdateEvent { + msg := newAgentMessage(reqCtx, parts...) + msg.Metadata = meta + statusEv := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateWorking, msg) + statusEv.Metadata = meta + return statusEv +} + // Execute implements a2asrv.AgentExecutor. // It follows the Python _handle_request pattern: set up session, handle HITL, // convert inbound message, run the agent loop, and emit A2A events. @@ -266,7 +293,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont // Events with no content carry metadata only; still track invocationID/usage. // Check for LLM error. if adkEvent.ErrorCode != "" { - errMsg := a2atype.NewMessage(a2atype.MessageRoleAgent, + errMsg := newAgentMessage(reqCtx, a2atype.TextPart{Text: fmt.Sprintf("LLM error: %s %s", adkEvent.ErrorCode, adkEvent.ErrorMessage)}) failed := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateFailed, errMsg) failed.Final = true @@ -278,7 +305,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont // Check for LLM error (even with content present). if adkEvent.ErrorCode != "" { - errMsg := a2atype.NewMessage(a2atype.MessageRoleAgent, + errMsg := newAgentMessage(reqCtx, a2atype.TextPart{Text: fmt.Sprintf("LLM error: %s %s", adkEvent.ErrorCode, adkEvent.ErrorMessage)}) failed := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateFailed, errMsg) failed.Final = true @@ -326,10 +353,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont if len(textOnly) > 0 { mirrorMeta := maps.Clone(eventMeta) mirrorMeta[adka2a.ToA2AMetaKey("partial")] = true - msg := a2atype.NewMessage(a2atype.MessageRoleAgent, textOnly...) - msg.Metadata = mirrorMeta - statusEv := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateWorking, msg) - statusEv.Metadata = mirrorMeta + statusEv := newAgentStatusEvent(reqCtx, textOnly, mirrorMeta) if err := queue.Write(ctx, statusEv); err != nil { return fmt.Errorf("failed to write partial status event: %w", err) } @@ -338,10 +362,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont mirrorParts := a2aParts if len(hitlParts) == 0 { // Only mirror when not accumulating HITL parts (those go into input_required). - msg := a2atype.NewMessage(a2atype.MessageRoleAgent, mirrorParts...) - msg.Metadata = maps.Clone(eventMeta) - statusEv := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateWorking, msg) - statusEv.Metadata = maps.Clone(eventMeta) + statusEv := newAgentStatusEvent(reqCtx, mirrorParts, maps.Clone(eventMeta)) if err := queue.Write(ctx, statusEv); err != nil { return fmt.Errorf("failed to write mirror status event: %w", err) } @@ -362,7 +383,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont } if runErr != nil { - errMsg := a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.TextPart{Text: runErr.Error()}) + errMsg := newAgentMessage(reqCtx, a2atype.TextPart{Text: runErr.Error()}) failed := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateFailed, errMsg) failed.Final = true failed.Metadata = finalMeta @@ -371,7 +392,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestCont if len(hitlParts) > 0 { // input_required: the agent is waiting for HITL decisions. - hitlMsg := a2atype.NewMessage(a2atype.MessageRoleAgent, hitlParts...) + hitlMsg := newAgentMessage(reqCtx, hitlParts...) inputRequired := a2atype.NewStatusUpdateEvent(reqCtx, a2atype.TaskStateInputRequired, hitlMsg) inputRequired.Final = true inputRequired.Metadata = finalMeta diff --git a/go/adk/pkg/a2a/executor_test.go b/go/adk/pkg/a2a/executor_test.go new file mode 100644 index 000000000..91af7fe16 --- /dev/null +++ b/go/adk/pkg/a2a/executor_test.go @@ -0,0 +1,67 @@ +package a2a + +import ( + "testing" + + a2atype "github.com/a2aproject/a2a-go/a2a" + "github.com/a2aproject/a2a-go/a2asrv" +) + +// TestNewAgentMessage_StampsContextAndTaskID verifies agent messages carry the +// request's context and task ids. A2A allows omitting them (the task is the +// canonical carrier), but stamping them lets consumers that flatten task.history +// key each message to its task without backfilling. +func TestNewAgentMessage_StampsContextAndTaskID(t *testing.T) { + reqCtx := &a2asrv.RequestContext{ + ContextID: "ctx-xyz", + TaskID: a2atype.TaskID("task-xyz"), + } + + msg := newAgentMessage(reqCtx, a2atype.TextPart{Text: "hello"}) + + if msg.ContextID != "ctx-xyz" { + t.Errorf("ContextID = %q, want %q", msg.ContextID, "ctx-xyz") + } + if msg.TaskID != a2atype.TaskID("task-xyz") { + t.Errorf("TaskID = %q, want %q", msg.TaskID, a2atype.TaskID("task-xyz")) + } + if msg.Role != a2atype.MessageRoleAgent { + t.Errorf("Role = %q, want %q", msg.Role, a2atype.MessageRoleAgent) + } +} + +// TestNewAgentStatusEvent_MessageCarriesIDs verifies the per-event emission seam: +// the working status event the executor writes (and that is persisted into +// task.history) carries an agent message stamped with the request's context/task +// ids. This is the property the send guard depends on — without it the persisted +// message keys differently from its locally-streamed counterpart and falsely +// blocks the next send. Mirrors the Python converter test. +func TestNewAgentStatusEvent_MessageCarriesIDs(t *testing.T) { + reqCtx := &a2asrv.RequestContext{ + ContextID: "ctx-xyz", + TaskID: a2atype.TaskID("task-xyz"), + } + meta := map[string]any{"k": "v"} + + ev := newAgentStatusEvent(reqCtx, a2atype.ContentParts{a2atype.TextPart{Text: "hi"}}, meta) + + if ev.Status.State != a2atype.TaskStateWorking { + t.Errorf("State = %q, want %q", ev.Status.State, a2atype.TaskStateWorking) + } + if ev.Status.Message == nil { + t.Fatal("status message is nil") + } + if ev.Status.Message.ContextID != "ctx-xyz" { + t.Errorf("message ContextID = %q, want %q", ev.Status.Message.ContextID, "ctx-xyz") + } + if ev.Status.Message.TaskID != a2atype.TaskID("task-xyz") { + t.Errorf("message TaskID = %q, want %q", ev.Status.Message.TaskID, a2atype.TaskID("task-xyz")) + } + // The event itself also carries the ids (from reqCtx), matching the message. + if ev.ContextID != "ctx-xyz" { + t.Errorf("event ContextID = %q, want %q", ev.ContextID, "ctx-xyz") + } + if ev.TaskID != a2atype.TaskID("task-xyz") { + t.Errorf("event TaskID = %q, want %q", ev.TaskID, a2atype.TaskID("task-xyz")) + } +} diff --git a/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx b/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx index c309215b1..5a301b447 100644 --- a/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx +++ b/ui/src/components/chat/__tests__/ChatInterface.sendGuard.test.tsx @@ -374,6 +374,42 @@ describe("ChatInterface send guard", () => { expect(mockToastInfo).not.toHaveBeenCalledWith(staleToastMessage); }); + it("does not block the next send after a same-tab text turn persisted with empty agent ids", async () => { + // The general case behind "every message needs sending twice": the backend + // persists AGENT messages with empty contextId/taskId (A2A optional fields; + // the task is the canonical carrier). The locally-streamed agent message + // carries the task's real ids, so it keys on ["task", ...] while the + // backend-extracted copy — pushed as-is with empty ids — keys on + // ["message", ]. They never match, so every turn (each has an + // agent text response) counts the backend as ahead and falsely blocks. + mockBackendTasks([completedTask("task-initial", initialTurn)]); + mockSendMessageStream + .mockResolvedValueOnce(streamOf(completedStatusEvent("same tab answer", "session-1", "task-streamed"))) + .mockResolvedValueOnce(streamOf(completedStatusEvent("next answer", "session-1", "task-next"))); + + renderExistingSession(); + + expect(await screen.findByText("initial answer")).toBeInTheDocument(); + + await sendText("same tab question"); + await waitFor(() => expect(mockSendMessageStream).toHaveBeenCalledTimes(1)); + expect(await screen.findByText("same tab answer")).toBeInTheDocument(); + + // Backend reflects the turn, with the agent message persisted with empty + // contextId/taskId (the user message keeps its contextId, as the client stamps it). + mockBackendTasks([ + completedTask("task-initial", initialTurn), + completedTask("task-streamed", [ + textMessage(sentMessage().messageId, "user", "same tab question", "session-1", ""), + textMessage("same-tab-agent", "agent", "same tab answer", "", ""), + ]), + ]); + await sendText("next question"); + + await waitFor(() => expect(mockSendMessageStream).toHaveBeenCalledTimes(2)); + expect(mockToastInfo).not.toHaveBeenCalledWith(staleToastMessage); + }); + it("still blocks when the backend has a cross-tab message not visible locally", async () => { mockBackendTasks([completedTask("task-initial", initialTurn)]); diff --git a/ui/src/lib/messageHandlers.ts b/ui/src/lib/messageHandlers.ts index f1f1e04d2..30e2dee11 100644 --- a/ui/src/lib/messageHandlers.ts +++ b/ui/src/lib/messageHandlers.ts @@ -62,13 +62,14 @@ export function extractMessagesFromTasks(tasks: Task[]): Message[] { // by the live-stream handlers so the rendering component can display them. // // Backfill contextId/taskId from the task when the history item omits them. - // Persisted agent messages frequently carry empty strings here, and `??` - // would keep the empty string ("" is not nullish) — leaving the converted - // tool messages without a contextId/taskId. The send guard then can't key - // them on (contextId, taskId) and falls back to the regenerated uuid - // messageId, so the same backend tool message keys differently on every - // extraction and never matches its local counterpart (false stale-send - // block). Treat "" as absent so these messages get the task's stable ids. + // Persisted agent messages (both tool and text) frequently carry empty + // strings here, and `??` would keep the empty string ("" is not nullish). + // The locally-streamed copies of these messages carry the task's real ids, + // so the send guard keys them on (contextId, taskId); without the backfill + // the backend-extracted copies fall back to messageId and never match the + // local ones, counting the backend as ahead and falsely blocking the next + // send on every turn. Treat "" as absent so both copies get the task's + // stable ids. Applied to every extracted agent message below. const msgContextId = historyItem.contextId || task.contextId; const msgTaskId = historyItem.taskId || task.id; const source = getSourceFromMetadata(historyItem.metadata as ADKMetadata | undefined, "assistant"); @@ -158,12 +159,16 @@ export function extractMessagesFromTasks(tasks: Task[]): Message[] { } } - // Text messages (or any message without data parts): push with tokenStats. + // Text messages (or any message without data parts): push with tokenStats + // and the backfilled contextId/taskId so they key the same way the + // locally-streamed copy does. if (!hasConvertedParts) { - messages.push(msgStats - ? { ...historyItem, metadata: { ...(historyItem.metadata as object || {}), tokenStats: msgStats } } - : historyItem - ); + messages.push({ + ...historyItem, + contextId: msgContextId, + taskId: msgTaskId, + ...(msgStats ? { metadata: { ...(historyItem.metadata as object || {}), tokenStats: msgStats } } : {}), + }); } } }