[Bug]: Context canceled error when sending multiple streaming messages with delays

[obutora]

What happened?

---

Context canceled error when sending multiple streaming messages with delays

Summary

When implementing an agent that sends multiple messages over time using q.Write() with delays (e.g., time.Sleep()), the context gets canceled after only a few messages, resulting in "context canceled" errors.

Environment

• Library version: v0.3.2
• Transport: gRPC
• OS: macOS

Expected Behavior

An agent's Execute() function should be able to send multiple messages over time:

func (*agentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestContext, q eventqueue.Queue) error {
    for i := 0; i < 10; i++ {
        q.Write(ctx, a2a.NewMessage(a2a.MessageRoleAgent, a2a.TextPart{Text: fmt.Sprintf("message %d", i)}))
        time.Sleep(time.Second * 1)
    }
    return nil
}

All 10 messages should be successfully delivered to the client.

Actual Behavior

The context gets canceled after sending only 1-2 messages:

2025/11/28 16:11:47 Sent message 0
2025/11/28 16:11:48 Sent message 1
2025/11/28 16:11:49 Failed to write a message: context canceled

Root Cause Analysis

The issue appears to be in internal/taskexec/execution.go:

func (e *Execution) processEvents(ctx context.Context, queue eventqueue.Queue) (a2a.SendMessageResult, error) {
    // ...
    queueReadCtx, cancelCtx := context.WithCancel(ctx)
    defer cancelCtx()  // ← This cancels the queue reading context
    go readQueueToChannels(queueReadCtx, queue, eventChan, errorChan)
    // ...
}

Problem flow:

1. Execute() returns after writing messages to queue
2. processEvents() exits
3. defer cancelCtx() is called
4. readQueueToChannels() goroutine's context is canceled
5. Messages still in the queue are never sent

Investigation

1. All official examples (examples/helloworld/server/*) only send a single message
2. There are no examples demonstrating multiple messages being sent with time delays
3. The library appears to be designed for immediate message batching rather than time-delayed streaming

Questions

1. Is sending multiple messages with time delays a supported use case?
2. If so, what is the recommended pattern for implementing this?
3. Should processEvents() wait for Execute() to complete and then drain the queue before canceling the context?
4. Is there a better approach to implementing streaming functionality that I might be missing? If there's a more idiomatic or recommended pattern for this use case, I'd appreciate any guidance.

Current Workaround

The only working pattern is to write all messages immediately without delays:

func (*agentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestContext, q eventqueue.Queue) error {
    for i := 0; i < 10; i++ {
        q.Write(ctx, a2a.NewMessage(...))
    }
    return nil
}

However, this doesn't support time-delayed streaming scenarios.

Relevant log output

2025/11/28 16:11:47 Sent message 0
2025/11/28 16:11:48 Sent message 1
2025/11/28 16:11:49 Failed to write a message: context canceled

Code of Conduct

• I agree to follow this project's Code of Conduct

[yarolegovich]

Hi, that's unfortunately a common misconception about Message vs Task in A2A protocol.
Message is supposed to be used for short responses which immediately terminate the event stream.
This is described in detail in AgentExecutor documentation and was discussed in the main spec repository as well.
Please refer to the referenced docs to see how responses should be streamed using Artifact.

[obutora]

Thank you for the clarification. I now understand that my understanding of Message vs Task in the A2A protocol was incorrect, and I appreciate you pointing me in the right direction.

I also want to express my gratitude for opening a separate issue to discuss this matter further.

That said, I believe my confusion partly stems from the limited examples available in this Go implementation. While I found that the Python implementation provides more comprehensive examples showing the complete flow, I wasn't able to find similar end-to-end examples in the Go implementation—for instance, how to wire up the Executor with the server's request handling, or how the RequestContext and Queue are instantiated and passed in.

If such examples already exist for the Go implementation, I would appreciate it if you could point me to them. If not, adding similar examples to those available in the Python implementation might be helpful for developers like myself and could benefit the community.

[yarolegovich]

Fair point, we're currently mostly focused on #115 and preparatory work for 1.0.
After that I think the effort will shift towards examples and conformance testing.
Can you please share some links to example you found the most useful / insightful?

[obutora]

Thank you for the update. I fully understand the prioritization, and I think focusing on #115 and the preparatory work for 1.0 is the right approach.

As for what helped me the most: it was actually writing my own implementation while reading through the documentation you provided. The process of implementing the stream handling myself gave me a much clearer understanding of how things work.

On the other hand, I did look at the existing Go sample, but it didn't quite help me understand the streaming flow, as it focuses more on the overall server setup rather than demonstrating the streaming behavior with Artifacts.

I think if a simple example like example/stream (separate from AgentExecutor) were available, showing both the client and server sides of streaming, it would have made understanding the protocol much smoother.

Here's the code I ended up with, in case it's helpful as a reference:

Client:

t := time.Now()
for event, err := range client.SendStreamingMessage(ctx, &a2a.MessageSendParams{Message: msg}) {
    if err != nil {
        log.Fatalf("Stream error: %v", err)
    }

    switch e := event.(type) {
    case *a2a.TaskArtifactUpdateEvent:
        if e.Artifact != nil {
            for _, part := range e.Artifact.Parts {
                if textPart, ok := part.(a2a.TextPart); ok {
                    if info, ok := textPart.Metadata["info"]; ok {
                        fmt.Printf("\n[Artifact info: %s]", info)
                    }
                    since := time.Since(t)
                    fmt.Printf(" %vms %s", since.Milliseconds(), textPart.Text+"\n")
                }
            }
        }
    case *a2a.TaskStatusUpdateEvent:
        fmt.Println()
        log.Printf("Status update: %v", e.Status.State)
    }
}

Server:

type agentExecutor struct{}

func (*agentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestContext, q eventqueue.Queue) error {
    log.Printf("Executing a request. TaskID: %s", reqCtx.TaskID)
    if reqCtx.Message != nil {
        for _, part := range reqCtx.Message.Parts {
            if textPart, ok := part.(a2a.TextPart); ok {
                log.Printf("Input: %s", textPart.Text)
            }
        }
    }

    if reqCtx.StoredTask == nil {
        event := a2a.NewStatusUpdateEvent(reqCtx, a2a.TaskStateSubmitted, nil)
        if err := q.Write(ctx, event); err != nil {
            return fmt.Errorf("failed to write state submitted: %w", err)
        }
    }

    workEvent := a2a.NewStatusUpdateEvent(reqCtx, a2a.TaskStateWorking, nil)
    if err := q.Write(ctx, workEvent); err != nil {
        return fmt.Errorf("failed to write state working: %w", err)
    }

    chunks := []string{"Hello", ", ", "world", "!"}
    var artifactID a2a.ArtifactID

    for _, chunk := range chunks {
        time.Sleep(1000 * time.Millisecond)

        parts := []a2a.Part{a2a.TextPart{Text: chunk, Metadata: map[string]any{
            "info": "working",
        }}}
        var event *a2a.TaskArtifactUpdateEvent

        if artifactID == "" {
            event = a2a.NewArtifactEvent(reqCtx, parts...)
            artifactID = event.Artifact.ID
        } else {
            event = a2a.NewArtifactUpdateEvent(reqCtx, artifactID, parts...)
        }

        if err := q.Write(ctx, event); err != nil {
            return fmt.Errorf("failed to write artifact update: %w", err)
        }
    }

    compEvent := a2a.NewStatusUpdateEvent(reqCtx, a2a.TaskStateCompleted, nil)
    compEvent.Final = true
    if err := q.Write(ctx, compEvent); err != nil {
        return fmt.Errorf("failed to write state completed: %w", err)
    }

    log.Printf("All messages written to queue")
    return nil
}

Feel free to use or adapt this if it's useful for future examples.