Function PollUntilTerminal

internal/experimental/a2abridge/a2a_event_handler.go:244–332

PollUntilTerminal polls GetTask with exponential backoff until the task reaches a terminal state, the context is canceled, or an error occurs. Newly discovered artifacts and status changes are emitted to the client. client and agentID identify the SDK client and the bridge's registered agent ID for

(
	ctx context.Context,
	client *a2aclient.Client,
	agentID string,
	conversationID string,
	toolCallID string,
	taskID a2a.TaskID,
	lastSeen *a2a.Task,
	alreadyEmitted []string,
	o agent.OutputHandler,
)

// the caller. It prevents re-emission across the streaming<->polling handoff
// and across invocation boundaries.
func PollUntilTerminal(
	ctx context.Context,
	client *a2aclient.Client,
	agentID string,
	conversationID string,
	toolCallID string,
	taskID a2a.TaskID,
	lastSeen *a2a.Task,
	alreadyEmitted []string,
	o agent.OutputHandler,
) error {
	emittedArtifactIDs := make(map[a2a.ArtifactID]bool)
	if lastSeen != nil {
		for _, art := range lastSeen.Artifacts {
			emittedArtifactIDs[art.ID] = true
		}
	}
	for _, id := range alreadyEmitted {
		emittedArtifactIDs[a2a.ArtifactID(id)] = true
	}

	const initialDelay = 250 * time.Millisecond
	const maxDelay = 60 * time.Second

	delay := initialDelay
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(delay):
		}

		task, err := client.GetTask(ctx, &a2a.GetTaskRequest{ID: taskID})
		if err != nil {
			// If the server has dropped the task while we were polling,
			// the marker is stale: clear it and surface a clean error.
			if errors.Is(err, a2a.ErrTaskNotFound) {
				_ = ClearStateMarker(conversationID, o)
				return fmt.Errorf("a2a agent %s: task %s no longer exists on server (state marker cleared)", agentID, taskID)
			}
			return fmt.Errorf("a2a agent %s: poll GetTask(%s): %w", agentID, taskID, err)
		}

		// Emit newly observed artifacts. Track whether anything new
		// arrived this iteration.
		sawNewArtifact := false
		for _, art := range task.Artifacts {
			if emittedArtifactIDs[art.ID] {
				continue
			}
			emittedArtifactIDs[art.ID] = true
			sawNewArtifact = true
			msgs := a2aArtifactToMessages(art, "agent")
			if len(msgs) > 0 {
				if err := o(&proto.AgentOutputs{Messages: msgs}); err != nil {
					return err
				}
			}