Track parent run IDs for durable flows

Co-authored-by: Codex <codex@openai.com>

ai/model.go

@@ -113,12 +113,17 @@ const (
 // RunInfo describes the agent run a tool call belongs to. The agent
 // attaches it to the context passed to a ToolHandler, so a wrapper can
 // correlate calls within a run and across delegation without coupling to
-// the agent package. Per-call detail (tool name, id) is on the ToolCall;
-// step and attempt counts are naturally counted by the wrapper itself.
+// the agent package. Flows also attach their name and current step so
+// tools and agents called from a workflow can be tied back to the
+// services → agents → workflows lifecycle that invoked them. Per-call
+// detail (tool name, id) is on the ToolCall; attempt counts are naturally
+// counted by the wrapper itself.
 type RunInfo struct {
-	RunID    string // correlation id for this agent run (one per Ask)
+	RunID    string // correlation id for this agent or flow run
 	ParentID string // the run that delegated to this one, if any
 	Agent    string // the agent's name
+	Flow     string // the flow's name, when the call is part of a workflow
+	Step     string // the flow step currently executing, when known
 }
 
 type runInfoKey struct{}

cmd/micro/cli/new/contract_test.go

@@ -19,6 +19,45 @@ import (
 // It shells out to `micro new` (which runs `go mod tidy`) and `go build`, so
 // it needs the Go toolchain and module access; it is skipped under `-short`.
 func TestZeroToOneContract(t *testing.T) {
+	generated := generateService(t, "helloworld")
+
+	for _, rel := range []string{"go.mod", "main.go", "handler/helloworld.go", "README.md", "Makefile"} {
+		if _, err := os.Stat(filepath.Join(generated.dir, rel)); err != nil {
+			t.Fatalf("generated file %s: %v", rel, err)
+		}
+	}
+
+	generated.replaceModule(t)
+	generated.build(t)
+}
+
+// TestZeroToOneNoMCPContract keeps the MCP opt-out path honest. Some services
+// intentionally run without the local MCP listener, but that variant must still
+// satisfy the same 0→1 contract: scaffold, tidy, and build without additional
+// toolchain dependencies.
+func TestZeroToOneNoMCPContract(t *testing.T) {
+	generated := generateService(t, "worker", "--no-mcp")
+
+	main, err := os.ReadFile(filepath.Join(generated.dir, "main.go"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(main), "gateway/mcp") || strings.Contains(string(main), "WithMCP") {
+		t.Fatalf("--no-mcp generated main.go with MCP wiring:\n%s", main)
+	}
+
+	generated.replaceModule(t)
+	generated.build(t)
+}
+
+type generatedService struct {
+	dir      string
+	repoRoot string
+}
+
+func generateService(t *testing.T, name string, args ...string) generatedService {
+	t.Helper()
+
 	if testing.Short() {
 		t.Skip("contract test shells out to the Go toolchain; skipped with -short")
 	}
@@ -39,35 +78,44 @@ func TestZeroToOneContract(t *testing.T) {
 	defer os.Chdir(oldwd)
 
 	set := flag.NewFlagSet("micro-new", flag.ContinueOnError)
-	if err := set.Parse([]string{"helloworld"}); err != nil {
+	set.Bool("no-mcp", false, "")
+	set.Bool("proto", false, "")
+	set.String("template", "", "")
+	set.String("prompt", "", "")
+	set.String("provider", "", "")
+	set.String("api_key", "", "")
+	if err := set.Parse(append(args, name)); err != nil {
 		t.Fatal(err)
 	}
 	ctx := cli.NewContext(cli.NewApp(), set, nil)
 
 	if err := Run(ctx); err != nil {
-		t.Fatalf("micro new helloworld: %v", err)
+		t.Fatalf("micro new %s %s: %v", strings.Join(args, " "), name, err)
 	}
 
-	serviceDir := filepath.Join(tmp, "helloworld")
-	for _, rel := range []string{"go.mod", "main.go", "handler/helloworld.go", "README.md", "Makefile"} {
-		if _, err := os.Stat(filepath.Join(serviceDir, rel)); err != nil {
-			t.Fatalf("generated file %s: %v", rel, err)
-		}
-	}
+	return generatedService{dir: filepath.Join(tmp, name), repoRoot: repoRoot}
+}
 
-	modPath := filepath.Join(serviceDir, "go.mod")
+func (g generatedService) replaceModule(t *testing.T) {
+	t.Helper()
+
+	modPath := filepath.Join(g.dir, "go.mod")
 	mod, err := os.ReadFile(modPath)
 	if err != nil {
 		t.Fatal(err)
 	}
 	modText := strings.Replace(string(mod), "go-micro.dev/v6 latest", "go-micro.dev/v6 v6.0.0", 1)
-	modText += "\nreplace go-micro.dev/v6 => " + filepath.ToSlash(repoRoot) + "\n"
+	modText += "\nreplace go-micro.dev/v6 => " + filepath.ToSlash(g.repoRoot) + "\n"
 	if err := os.WriteFile(modPath, []byte(modText), 0644); err != nil {
 		t.Fatal(err)
 	}
+}
+
+func (g generatedService) build(t *testing.T) {
+	t.Helper()
 
 	cmd := exec.Command("go", "build", "./...")
-	cmd.Dir = serviceDir
+	cmd.Dir = g.dir
 	out, err := cmd.CombinedOutput()
 	if err != nil {
 		t.Fatalf("generated service go build ./... failed: %v\n%s", err, out)

flow/otel.go

@@ -17,6 +17,7 @@ const (
 	spanNameFlowStep = "flow.step"
 
 	AttrFlowRunID     = "flow.run.id"
+	AttrFlowParentID  = "flow.run.parent_id"
 	AttrFlowName      = "flow.name"
 	AttrFlowStepName  = "flow.step.name"
 	AttrFlowStatus    = "flow.status"
@@ -34,6 +35,7 @@ func (f *Flow) startRunSpan(ctx context.Context, run Run) (context.Context, func
 	}
 	ctx, span := f.tracer().Start(ctx, spanNameFlowRun, trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes(
 		attribute.String(AttrFlowRunID, run.ID),
+		attribute.String(AttrFlowParentID, run.ParentID),
 		attribute.String(AttrFlowName, f.name),
 		attribute.String(AttrFlowStatus, run.Status),
 	))
@@ -60,6 +62,7 @@ func (f *Flow) runStepSpan(ctx context.Context, step Step, in State) (State, int
 	info, _ := ai.RunInfoFrom(ctx)
 	ctx, span := f.tracer().Start(ctx, spanNameFlowStep, trace.WithAttributes(
 		attribute.String(AttrFlowRunID, info.RunID),
+		attribute.String(AttrFlowParentID, info.ParentID),
 		attribute.String(AttrFlowName, f.name),
 		attribute.String(AttrFlowStepName, step.Name),
 	))

flow/otel_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/store"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/trace"
@@ -19,7 +20,8 @@ func TestFlowOpenTelemetrySpans(t *testing.T) {
 		return in, nil
 	}}
 	f := New("observed", WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "observed")), TraceProvider(tp), Steps(step))
-	if err := f.Execute(context.Background(), "start"); err != nil {
+	ctx := withTestRunInfo(context.Background(), "agent-run-otel")
+	if err := f.Execute(ctx, "start"); err != nil {
 		t.Fatal(err)
 	}
 
@@ -32,12 +34,12 @@ func TestFlowOpenTelemetrySpans(t *testing.T) {
 		case spanNameFlowRun:
 			seen[spanNameFlowRun] = true
 			runID = attrs[AttrFlowRunID]
-			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStatus] != "done" {
+			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStatus] != "done" || attrs[AttrFlowParentID] != "agent-run-otel" {
 				t.Fatalf("run span attributes = %#v", attrs)
 			}
 		case spanNameFlowStep:
 			seen[spanNameFlowStep] = true
-			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStepName] != "inspect" {
+			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStepName] != "inspect" || attrs[AttrFlowParentID] != "agent-run-otel" {
 				t.Fatalf("step span attributes = %#v", attrs)
 			}
 		}
@@ -68,3 +70,7 @@ func flowSpanAttributes(attrs []attribute.KeyValue) map[string]string {
 	}
 	return out
 }
+
+func withTestRunInfo(ctx context.Context, runID string) context.Context {
+	return ai.WithRunInfo(ctx, ai.RunInfo{RunID: runID, Agent: "planner"})
+}

flow/steps.go

@@ -74,13 +74,14 @@ type StepRecord struct {
 // saves and loads. It is retained for success and failure unless the flow
 // opts into cleanup with DeleteOnSuccess.
 type Run struct {
-	ID      string       `json:"id"`
-	Flow    string       `json:"flow"`
-	State   State        `json:"state"`
-	Steps   []StepRecord `json:"steps"`
-	Status  string       `json:"status"` // running | done | failed
-	Started time.Time    `json:"started"`
-	Updated time.Time    `json:"updated"`
+	ID       string       `json:"id"`
+	ParentID string       `json:"parent_id,omitempty"`
+	Flow     string       `json:"flow"`
+	State    State        `json:"state"`
+	Steps    []StepRecord `json:"steps"`
+	Status   string       `json:"status"` // running | done | failed
+	Started  time.Time    `json:"started"`
+	Updated  time.Time    `json:"updated"`
 }
 
 // Checkpoint persists and restores flow runs so a run survives a crash
@@ -309,12 +310,17 @@ func (f *Flow) startRun(ctx context.Context, data string) (Run, error) {
 	if err := validateSteps(f.opts.Steps); err != nil {
 		return Run{}, err
 	}
+	parentID := ""
+	if info, ok := ai.RunInfoFrom(ctx); ok {
+		parentID = info.RunID
+	}
 	run := Run{
-		ID:      uuid.New().String(),
-		Flow:    f.name,
-		State:   State{Stage: f.opts.Steps[0].Name, Data: []byte(data)},
-		Status:  "running",
-		Started: time.Now(),
+		ID:       uuid.New().String(),
+		ParentID: parentID,
+		Flow:     f.name,
+		State:    State{Stage: f.opts.Steps[0].Name, Data: []byte(data)},
+		Status:   "running",
+		Started:  time.Now(),
 	}
 	for _, s := range f.opts.Steps {
 		run.Steps = append(run.Steps, StepRecord{Name: s.Name, Status: "pending"})
@@ -390,7 +396,7 @@ func (f *Flow) Pending(ctx context.Context) ([]Run, error) {
 func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
 	steps := f.opts.Steps
 	ctx = withDeps(ctx, &runDeps{client: f.client, model: f.model, tools: f.toolSet})
-	ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: run.ID, Agent: f.name})
+	ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: run.ID, ParentID: run.ParentID, Agent: f.name, Flow: f.name})
 	ctx, finishSpan := f.startRunSpan(ctx, run)
 	var spanErr error
 	defer func() { finishSpan(run, spanErr) }()
@@ -478,6 +484,10 @@ func (f *Flow) runStep(ctx context.Context, step Step, in State) (State, int, er
 		if err := ctx.Err(); err != nil {
 			return in, attempt - 1, err
 		}
+		if info, ok := ai.RunInfoFrom(ctx); ok {
+			info.Step = step.Name
+			ctx = ai.WithRunInfo(ctx, info)
+		}
 		out, err := step.Run(ctx, in)
 		if err == nil {
 			return out, attempt, nil

flow/steps_test.go

@@ -99,11 +99,13 @@ func TestFlowStepContextIncludesRunInfo(t *testing.T) {
 		return in, nil
 	}}
 
+	mem := store.NewMemoryStore()
 	f := New("correlated",
-		WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "correlated")),
+		WithCheckpoint(StoreCheckpoint(mem, "correlated")),
 		Steps(step),
 	)
-	if err := f.Execute(context.Background(), "start"); err != nil {
+	ctx := ai.WithRunInfo(context.Background(), ai.RunInfo{RunID: "agent-run-1", Agent: "planner"})
+	if err := f.Execute(ctx, "start"); err != nil {
 		t.Fatalf("Execute: %v", err)
 	}
 	if got.Agent != "correlated" {
@@ -112,6 +114,22 @@ func TestFlowStepContextIncludesRunInfo(t *testing.T) {
 	if got.RunID == "" {
 		t.Fatal("RunInfo.RunID is empty")
 	}
+	if got.Flow != "correlated" {
+		t.Fatalf("RunInfo.Flow = %q, want correlated", got.Flow)
+	}
+	if got.ParentID != "agent-run-1" {
+		t.Fatalf("RunInfo.ParentID = %q, want agent-run-1", got.ParentID)
+	}
+	if got.Step != "inspect" {
+		t.Fatalf("RunInfo.Step = %q, want inspect", got.Step)
+	}
+	runs, err := StoreCheckpoint(mem, "correlated").List(context.Background())
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	if len(runs) != 1 || runs[0].ParentID != "agent-run-1" {
+		t.Fatalf("persisted parent id = %+v, want agent-run-1", runs)
+	}
 }
 
 func TestFlowResumePendingResumesOldestRunsUntilFailure(t *testing.T) {

gateway/a2a/a2a.go

@@ -18,10 +18,11 @@
 //		BaseURL:  "https://agents.example.com",
 //	})
 //
-// Scope of this version: the synchronous JSON-RPC binding — `message/send`
-// (returns a completed Task), `tasks/get`, and Agent Card discovery.
-// Streaming (`message/stream`), multi-turn `input-required`, and push
-// notifications are advertised as unsupported and are follow-ups.
+// Scope of this version: the JSON-RPC binding — `message/send`
+// (returns a completed Task), `message/stream` (SSE with the completed
+// Task event), `tasks/get`, and Agent Card discovery. Multi-turn
+// `input-required`, `tasks/resubscribe`, and push notifications are
+// advertised as unsupported and are follow-ups.
 package a2a
 
 import (
@@ -314,7 +315,7 @@ func Card(name, url, description string, services []string) AgentCard {
 		URL:             url,
 		Version:         "1.0.0",
 		ProtocolVersion: protocolVersion,
-		Capabilities:    Capabilities{Streaming: false, PushNotifications: false},
+		Capabilities:    Capabilities{Streaming: true, PushNotifications: false},
 		// The agent converses over a single Chat endpoint; advertise that
 		// as one skill, tagged with the services it manages.
 		DefaultInputModes:  []string{"text/plain"},
@@ -416,13 +417,15 @@ func (d *dispatcher) serve(w http.ResponseWriter, r *http.Request, invoke Invoke
 	switch req.Method {
 	case "message/send":
 		d.send(requestContext(r.Context()), w, req, invoke)
+	case "message/stream":
+		d.stream(requestContext(r.Context()), w, req, invoke)
 	case "tasks/get":
 		d.get(w, req)
 	case "tasks/cancel":
 		// v1 tasks complete synchronously, so they're already terminal.
 		writeRPC(w, req.ID, nil, &rpcError{Code: errNotCancelable, Message: "task is not cancelable"})
-	case "message/stream", "tasks/resubscribe":
-		writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "streaming is not supported"})
+	case "tasks/resubscribe":
+		writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "resubscribe is not supported"})
 	default:
 		writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "method not found: " + req.Method})
 	}
@@ -433,15 +436,38 @@ type sendParams struct {
 }
 
 func (d *dispatcher) send(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke Invoke) {
-	var p sendParams
-	if err := json.Unmarshal(req.Params, &p); err != nil {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
+	task, e := d.run(ctx, req.Params, invoke)
+	if e != nil {
+		writeRPC(w, req.ID, nil, e)
 		return
 	}
+	writeRPC(w, req.ID, task, nil)
+}
+
+func (d *dispatcher) stream(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke Invoke) {
+	task, e := d.run(ctx, req.Params, invoke)
+	if e != nil {
+		writeRPC(w, req.ID, nil, e)
+		return
+	}
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.WriteHeader(http.StatusOK)
+	_ = json.NewEncoder(sseWriter{w: w}).Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
+	if f, ok := w.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+func (d *dispatcher) run(ctx context.Context, params json.RawMessage, invoke Invoke) (*Task, *rpcError) {
+	var p sendParams
+	if err := json.Unmarshal(params, &p); err != nil {
+		return nil, &rpcError{Code: errInvalidParams, Message: "invalid params"}
+	}
 	text := textOf(p.Message.Parts)
 	if text == "" {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "message has no text part"})
-		return
+		return nil, &rpcError{Code: errInvalidParams, Message: "message has no text part"}
 	}
 
 	reply, err := invoke(ctx, text)
@@ -464,7 +490,7 @@ func (d *dispatcher) send(ctx context.Context, w http.ResponseWriter, req rpcReq
 		task.Artifacts = []Artifact{textArtifact(reply)}
 	}
 	d.store(task)
-	writeRPC(w, req.ID, task, nil)
+	return task, nil
 }
 
 type getParams struct {
@@ -565,6 +591,24 @@ func requestContext(parent context.Context) context.Context {
 	return ctx
 }
 
+type sseWriter struct {
+	w http.ResponseWriter
+}
+
+func (s sseWriter) Write(p []byte) (int, error) {
+	if _, err := s.w.Write([]byte("data: ")); err != nil {
+		return 0, err
+	}
+	n, err := s.w.Write(p)
+	if err != nil {
+		return n, err
+	}
+	if _, err := s.w.Write([]byte("\n")); err != nil {
+		return n, err
+	}
+	return n, nil
+}
+
 func writeJSON(w http.ResponseWriter, status int, v any) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)

gateway/a2a/a2a_test.go

@@ -4,8 +4,10 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"
 
@@ -44,6 +46,7 @@ func newGatewayWithAgent(t *testing.T) (*httptest.Server, func()) {
 
 	srv := server.NewServer(
 		server.Name("echo"),
+		server.Address("127.0.0.1:0"),
 		server.Registry(reg),
 		server.Metadata(map[string]string{"type": "agent", "services": ""}),
 	)
@@ -145,6 +148,42 @@ func TestMessageSendUsesRequestContext(t *testing.T) {
 	}
 }
 
+func TestMessageStream(t *testing.T) {
+	ts, cleanup := newGatewayWithAgent(t)
+	defer cleanup()
+
+	body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
+	resp, err := http.Post(ts.URL+"/agents/echo", "application/json", bytes.NewBufferString(body))
+	if err != nil {
+		t.Fatalf("post: %v", err)
+	}
+	defer resp.Body.Close()
+	if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
+		t.Fatalf("content-type = %q, want text/event-stream", ct)
+	}
+
+	var line string
+	if _, err := fmt.Fscan(resp.Body, &line); err != nil {
+		t.Fatalf("read event prefix: %v", err)
+	}
+	if line != "data:" {
+		t.Fatalf("event prefix = %q, want data:", line)
+	}
+	var out struct {
+		Result Task      `json:"result"`
+		Error  *rpcError `json:"error"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		t.Fatalf("decode event: %v", err)
+	}
+	if out.Error != nil {
+		t.Fatalf("rpc error: %+v", out.Error)
+	}
+	if out.Result.Status.State != stateCompleted || len(out.Result.Artifacts) != 1 || textOf(out.Result.Artifacts[0].Parts) != "pong" {
+		t.Fatalf("streamed task = %+v", out.Result)
+	}
+}
+
 func TestUnknownMethod(t *testing.T) {
 	ts, cleanup := newGatewayWithAgent(t)
 	defer cleanup()
@@ -152,9 +191,9 @@ func TestUnknownMethod(t *testing.T) {
 	var resp struct {
 		Error *rpcError `json:"error"`
 	}
-	rpc(t, ts.URL+"/agents/echo", `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{}}`, &resp)
+	rpc(t, ts.URL+"/agents/echo", `{"jsonrpc":"2.0","id":1,"method":"tasks/resubscribe","params":{}}`, &resp)
 	if resp.Error == nil || resp.Error.Code != errMethodNotFound {
-		t.Errorf("expected method-not-found for streaming, got %+v", resp.Error)
+		t.Errorf("expected method-not-found for resubscribe, got %+v", resp.Error)
 	}
 }
 

internal/website/docs/guides/a2a-protocol.md

@@ -132,9 +132,10 @@ yet terminal, it polls `tasks/get` until it completes.
 
 ## Scope
 
-This is the synchronous JSON-RPC binding:
+This is the JSON-RPC binding for completed-task execution:
 
 - **`message/send`** runs the agent and returns a completed `Task`.
+- **`message/stream`** streams the completed `Task` as an SSE `data:` event, giving A2A clients a streaming-compatible path while the underlying agent call remains synchronous.
 - **`tasks/get`** returns a recent task by id.
 - **Agent Card** discovery, generated from the registry.
 
@@ -142,11 +143,11 @@ Both directions work: the gateway exposes your agents, and `a2a.Client` (via `fl
 
 Not yet supported (advertised as such on the card, so clients negotiate correctly):
 
-- **`message/stream`** (SSE streaming) and `tasks/resubscribe`.
+- **`tasks/resubscribe`** for reconnecting to a live stream.
 - Multi-turn `input-required` tasks.
 - Push notifications.
 
-These are the natural follow-ups; the synchronous binding is what makes a Go Micro agent both reachable from, and able to reach, the A2A ecosystem today.
+These are the natural follow-ups; the completed-task binding is what makes a Go Micro agent both reachable from, and able to reach, the A2A ecosystem today.
 
 ## See also