Overview

Add ExecAggStageResultsAndMetadataSource, a new SDK base class for extension execution stages that produce two streams. Provides helper methods that wrap user BSON in the {$_streamType, payload} envelope expected by the host's Exchange routing.

Background

The host's Exchange routes documents from the extension source to two consumers using a kKeyRange policy on a _streamType field. Rather than requiring each extension to manually construct this envelope, the SDK provides advanced() helpers that wrap user BSON as {_streamType: N, payload: }}}. The host then unwraps via {{$replaceRoot($payload) downstream of each consumer.

When the host calls skip_stream(kMetadataResult) during optimization, the extension sets _multiStreamEnabled = false and the advanced() helper omits the envelope, since no Exchange is present in the elided path.

Scope of Work

Add ExecAggStageResultsAndMetadataSource to src/mongo/db/extension/sdk/aggregation_stage.h:

• Extends ExecAggStageSource
• enum class StreamType
• advanced(BSONObjBuilder& bob, StreamType streamType) — wraps as {_streamType: N, payload: bob.done()} when _multiStreamEnabled, otherwise returns plain doc
• advanced(BSONObjBuilder& bob, StreamType streamType, BSONObjBuilder& metaBob) — overload for stages that also produce per-document metadata (e.g. $sortKey, $searchScore)
• void setMultiStreamEnabled(bool enabled) / bool _multiStreamEnabled = true

Acceptance Criteria

• advanced() wraps BSON in envelope when _multiStreamEnabled = true
• advanced() passes BSON through unwrapped when _multiStreamEnabled = false
• Existing ExecAggStageSource subclasses are unaffected
• SDK unit tests cover both modes and both advanced() overloads