Method runStreamLoop

pkg/runtime/loop.go:236–536

runStreamLoop is the body of RunStream. Pulled out of the anonymous goroutine so it has a real name in stack traces and is easier to navigate in editors.

(ctx context.Context, sess *session.Session, events chan Event)

// goroutine so it has a real name in stack traces and is easier to navigate
// in editors.
func (r *LocalRuntime) runStreamLoop(ctx context.Context, sess *session.Session, events chan Event) {
	sink := &channelSink{ch: events}

	// Seed the cagent session ID at the run-loop boundary so any
	// gateway-bound HTTP call originating from this loop can correlate
	// back to the originating session. Plumbing happens in
	// pkg/httpclient/userAgentTransport, gated on `X-Cagent-Forward`.
	ctx = httpclient.ContextWithSessionID(ctx, sess.ID)
	r.telemetry.RecordSessionStart(ctx, r.currentAgentName(), sess.ID)

	// Seed `gen_ai.conversation.id` into baggage at the session
	// boundary. Every span the runtime, providers, MCP client, RAG,
	// sandbox, evaluation, hooks, and (downstream) any subprocess
	// or remote service create from here on will pick it up
	// automatically without per-helper plumbing — and the value
	// rides over W3C `baggage` so it crosses MCP / sandbox /
	// HTTP boundaries too.
	ctx = genai.WithConversationID(ctx, sess.ID)

	// A non-interactive session (background_agents via runCollecting, MCP
	// serve, A2A, evals) has no live UI that can answer an OAuth elicitation,
	// and runCollecting drops events rather than forwarding them. If a remote
	// MCP toolset needs first-time OAuth, blocking on an elicitation that
	// nobody can answer hangs the sub-agent forever (issue #3200): the
	// connection context is detached with context.WithoutCancel, so not even
	// cancellation can unblock it. Mark the context so toolset Start() fails
	// fast with an AuthorizationRequiredError — the same fast-fail the startup
	// tool probe uses (see EmitStartupInfo) — instead of eliciting. A real user
	// authorizes such servers from an interactive turn (or transfer_task, which
	// keeps NonInteractive=false and forwards the dialog to the TUI).
	if sess.NonInteractive {
		ctx = mcptools.WithoutInteractivePrompts(ctx)
	}

	// runtime.session is the root span for one stream. gen_ai.* keys
	// are emitted alongside the legacy `agent` / `session.id` keys
	// so existing dashboards keep matching while spec-aware tooling
	// can filter by `gen_ai.conversation.id` and
	// `cagent.agent.name`. Legacy keys drop out under
	// OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental.
	sessionAttrs := []attribute.KeyValue{
		attribute.String(genai.AttrConversationID, sess.ID),
		attribute.String(genai.AttrAgentNameRuntime, r.currentAgentName()),
	}
	if genai.EmitLegacyAttributes() {
		sessionAttrs = append(sessionAttrs,
			attribute.String("agent", r.currentAgentName()),
			attribute.String("session.id", sess.ID),
		)
	}
	ctx, sessionSpan := r.startSpan(ctx, "runtime.session", trace.WithAttributes(sessionAttrs...))
	defer sessionSpan.End()

	// Swap in this stream's events channel for elicitation and save the
	// previous one so it can be restored on teardown. This allows nested
	// RunStream calls to temporarily own elicitation without losing the
	// parent's channel.
	prevElicitationCh := r.elicitation.swap(events)