micro--go-micro/agent/a2a_stream_test.go

package agent

import (
	"bytes"
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	"go-micro.dev/v6/ai"
	"go-micro.dev/v6/gateway/a2a"
)

func TestA2AStreamUsesAgentChatPathWithTools(t *testing.T) {
	var sawTool bool
	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
		if opts.ToolHandler == nil {
			t.Fatal("model was not wired with agent tool handler")
		}
		result := opts.ToolHandler(ctx, ai.ToolCall{
			ID:    "call-1",
			Name:  "echo",
			Input: map[string]any{"value": "a2a-stream"},
		})
		if !strings.Contains(result.Content, "a2a-stream-ok") {
			t.Fatalf("tool result = %q, want marker", result.Content)
		}
		return &ai.Response{Answer: "streamed " + result.Content}, nil
	}
	defer func() { fakeGen = nil }()

	a := newTestAgent(Name("stream-agent"), WithTool("echo", "echo text", nil, func(ctx context.Context, input map[string]any) (string, error) {
		sawTool = true
		if info, ok := ai.RunInfoFrom(ctx); !ok || info.RunID == "" || info.Agent != "stream-agent" {
			t.Fatalf("RunInfo = %+v ok=%v, want stream-agent run", info, ok)
		}
		if input["value"] != "a2a-stream" {
			t.Fatalf("tool input = %+v, want a2a-stream", input)
		}
		return "a2a-stream-ok", nil
	}))
	h := a2a.NewAgentStreamHandler(
		a2a.Card("stream-agent", "http://example.invalid/stream-agent", "", nil),
		func(ctx context.Context, text string) (string, error) {
			resp, err := a.Ask(ctx, text)
			if err != nil {
				return "", err
			}
			return resp.Reply, nil
		},
		a.streamAskAI,
	)

	body := []byte(`{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"run stream tool"}],"kind":"message"}}}`)
	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
	rr := httptest.NewRecorder()
	h.ServeHTTP(rr, req)

	if !sawTool {
		t.Fatal("A2A stream did not execute the agent tool path")
	}
	if ct := rr.Result().Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
		t.Fatalf("content-type = %q, want text/event-stream", ct)
	}
	if !strings.Contains(rr.Body.String(), "a2a-stream-ok") {
		t.Fatalf("stream body missing tool marker: %s", rr.Body.String())
	}

	// The spec-shaped stream carries the answer as append artifact-update
	// deltas and closes with a completed status-update (final:true).
	var (
		text       strings.Builder
		finalState string
		sawFinal   bool
	)
	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
		line = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(line), "data: "))
		if line == "" {
			continue
		}
		var ev struct {
			Result json.RawMessage `json:"result"`
			Error  any             `json:"error"`
		}
		if err := json.Unmarshal([]byte(line), &ev); err != nil {
			t.Fatalf("decode event %q: %v", line, err)
		}
		if ev.Error != nil {
			t.Fatalf("event carried an error field: %+v", ev.Error)
		}
		var kind struct {
			Kind string `json:"kind"`
		}
		_ = json.Unmarshal(ev.Result, &kind)
		switch kind.Kind {
		case "artifact-update":
			var au struct {
				Artifact struct {
					Parts []struct {
						Text string `json:"text"`
					} `json:"parts"`
				} `json:"artifact"`
			}
			_ = json.Unmarshal(ev.Result, &au)
			for _, p := range au.Artifact.Parts {
				text.WriteString(p.Text)
			}
		case "status-update":
			var su struct {
				Status struct {
					State string `json:"state"`
				} `json:"status"`
				Final bool `json:"final"`
			}
			_ = json.Unmarshal(ev.Result, &su)
			if su.Final {
				sawFinal = true
				finalState = su.Status.State
			}
		default: // opening "task" snapshot
			var task struct {
				Artifacts []struct {
					Parts []struct {
						Text string `json:"text"`
					} `json:"parts"`
				} `json:"artifacts"`
			}
			_ = json.Unmarshal(ev.Result, &task)
			for _, a := range task.Artifacts {
				for _, p := range a.Parts {
					if p.Text != "" {
						text.WriteString(p.Text)
					}
				}
			}
		}
	}
	if !sawFinal || finalState != "completed" {
		t.Fatalf("want a completed final:true status-update; sawFinal=%v state=%q", sawFinal, finalState)
	}
	if !strings.Contains(text.String(), "a2a-stream-ok") {
		t.Fatalf("reassembled stream text missing tool marker: %q", text.String())
	}
}