feat(compilers/openapi): populate the operation streaming fields

compilers/openapi/conformance_test.go

@@ -187,6 +187,8 @@ func conformanceCases() []conformanceCase {
 		{"multipart-encoding", assertMultipartEncoding},
 		{"file-body", assertFileBody},
 		{"sequential-media", assertSequentialMedia},
+		{"streaming-media-30", assertStreamingMedia30},
+		{"streaming-media-31", assertStreamingMedia31},
 		{"per-status-errors", assertPerStatusErrors},
 		{"response-links", assertResponseLinks},
 		{"webhooks", assertWebhooks},
@@ -1733,6 +1735,11 @@ func assertPartHeaders(t *testing.T, doc *ir.Document, headers []ir.Property) {
 // Content.ItemEncoding, while a positional prefixEncoding — which a single
 // every-item encoding has no ordinals for — takes itself and the tail encoding
 // beside it into Unmodeled instead.
+//
+// It also pins what itemSchema says about the operation, which for a long while
+// was nothing: the keyword states that the body is a sequence of items, so the
+// operation streams, and it says so itself rather than being guessed at from
+// the media type — multipart/mixed is in no streaming media-type list.
 func assertSequentialMedia(t *testing.T, doc *ir.Document, _ []ir.Diagnostic) {
 	events, ok := opByName(doc, "streamEvents")
 	require.True(t, ok)
@@ -1743,6 +1750,13 @@ func assertSequentialMedia(t *testing.T, doc *ir.Document, _ []ir.Diagnostic) {
 	assert.True(t, c.ItemEncoding.Multi, "the construct describes a repeated tail")
 	assert.Empty(t, c.Unmodeled, "nothing is left over once it lowers")
 
+	assert.Equal(t, ir.StreamingServer, events.Streaming)
+	require.NotNil(t, events.ResponseStream)
+	require.NotNil(t, events.ResponseStream.Events)
+	assert.Empty(t, cmp.Diff(*c.Item, *events.ResponseStream.Events),
+		"the declared item schema is the stream element type")
+	assert.Empty(t, events.Provenance.Inferred, "a declared sequence is not a heuristic")
+
 	parts, ok := opByName(doc, "streamParts")
 	require.True(t, ok)
 	pc := firstContent(t, parts)

compilers/openapi/helpers_test.go

@@ -155,7 +155,7 @@ func newLowerer(doc *load.Document, opts Options) *lowerer {
 func newRawLowerer(doc *soa.OpenAPI) *lowerer {
 	rawTypes := compile.NewTypes(0)
 	l := &lowerer{
-		ctx:          lowering.New(0, doc, ir.SourceInfo{}, "", overlay.Origin{}),
+		ctx:          lowering.New(0, doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}),
 		out:          &ir.Document{Types: rawTypes.Registry()},
 		types:        rawTypes,
 		operationIDs: make(map[string]string),

compilers/openapi/internal/lowering/lowering.go

@@ -44,9 +44,9 @@ type Ctx struct {
 	// SrcIndex is this source's index within the compile, stamped into every
 	// Provenance.
 	SrcIndex int
-	// Grouping selects how operations are grouped into OperationGroups. It is the
-	// only policy the context carries: everything else here is a fact about the
-	// document, and this is a fact about the caller.
+	// Grouping selects how operations are grouped into OperationGroups. It is one
+	// of the two caller policies the context carries; everything else here is a
+	// fact about the document.
 	//
 	// It arrives as the caller wrote it, normalized or not — the compiler's
 	// Options fills an unset one in before building a context, but nothing here
@@ -55,6 +55,14 @@ type Ctx struct {
 	// than a second spelling of the default to keep in step.
 	Grouping GroupingStrategy
 
+	// streaming is the media-type streaming policy, normalized into the set
+	// MediaTypeStreams answers from, and nil when the caller disabled it.
+	//
+	// It is the second caller policy, and it is unexported where Grouping is not
+	// because it holds a map: a struct copy would share it, which is the one
+	// thing keeping the other maps here unexported is for.
+	streaming map[string]bool
+
 	// schemas is the set of component-schema names the document declares.
 	//
 	// It is unexported and read through DeclaresSchema because a struct copy
@@ -99,19 +107,26 @@ type Ctx struct {
 // document as a valid target. It stays nil for a document that declares no
 // components, which reads the same as an empty set.
 //
+// The streaming policy is normalized into its lookup set here for a related
+// reason: normalizing at each reader would be as many places for the comparison
+// to differ as there are readers, and a media type that matched at one of them
+// and not another would classify one direction of an operation and not the
+// other.
+//
 // The $dynamicAnchor index is deliberately not derived here, though GitHub #172
 // asked for it. Building it emits a diagnostic when the walk hits its bounds, so
 // building it is a lowering action rather than context: done at entry, that
 // warning would reach documents that never write $dynamicRef, changing what the
 // compiler reports about them. It stays where it is, built on first use.
-func New(srcIndex int, doc *soa.OpenAPI, src ir.SourceInfo, grouping GroupingStrategy, origin overlay.Origin) Ctx {
+func New(srcIndex int, doc *soa.OpenAPI, src ir.SourceInfo, grouping GroupingStrategy, streaming StreamingMedia, origin overlay.Origin) Ctx {
 	return Ctx{
-		Doc:      doc,
-		Source:   src,
-		SrcIndex: srcIndex,
-		Grouping: grouping,
-		schemas:  declaredSchemaNames(doc),
-		overlay:  origin,
+		Doc:       doc,
+		Source:    src,
+		SrcIndex:  srcIndex,
+		Grouping:  grouping,
+		schemas:   declaredSchemaNames(doc),
+		streaming: streamingSet(streaming),
+		overlay:   origin,
 	}
 }
 

compilers/openapi/internal/lowering/lowering_test.go

@@ -94,7 +94,7 @@ func TestNew_DerivesTheDeclaredSchemaNames(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
-			c := lowering.New(0, tc.doc, ir.SourceInfo{}, "", overlay.Origin{})
+			c := lowering.New(0, tc.doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{})
 			for _, n := range tc.declares {
 				assert.True(t, c.DeclaresSchema(n), "%q is declared", n)
 			}
@@ -123,7 +123,7 @@ func TestNew_KeepsTheDocumentItWasGiven(t *testing.T) {
 	doc := docDeclaring("User")
 	src := ir.SourceInfo{Format: "openapi@3.1", Path: "spec.yaml", Hash: "abc"}
 
-	c := lowering.New(7, doc, src, lowering.GroupByPathPrefix, overlay.Origin{})
+	c := lowering.New(7, doc, src, lowering.GroupByPathPrefix, lowering.StreamingMedia{}, overlay.Origin{})
 
 	assert.Same(t, doc, c.Doc, "the document is referenced, never copied")
 	assert.Equal(t, src, c.Source)
@@ -140,7 +140,7 @@ func TestWithAuth_ExtendsACopy(t *testing.T) {
 	t.Parallel()
 	doc := docDeclaring("User")
 	src := ir.SourceInfo{Format: "openapi@3.1", Path: "spec.yaml", Hash: "abc"}
-	before := lowering.New(7, doc, src, lowering.GroupByPathPrefix, overlay.Origin{})
+	before := lowering.New(7, doc, src, lowering.GroupByPathPrefix, lowering.StreamingMedia{}, overlay.Origin{})
 	schemes := map[ir.AuthID]ir.AuthScheme{"a/apiKey": {ID: "a/apiKey"}}
 
 	after := before.WithAuth(schemes)
@@ -203,7 +203,7 @@ func TestExclusiveBoundIsBoolean_FollowsTheDialect(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.version, func(t *testing.T) {
 			t.Parallel()
-			c := lowering.New(0, &soa.OpenAPI{OpenAPI: tc.version}, ir.SourceInfo{}, "", overlay.Origin{})
+			c := lowering.New(0, &soa.OpenAPI{OpenAPI: tc.version}, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{})
 			assert.Equal(t, tc.want, c.ExclusiveBoundIsBoolean())
 		})
 	}
@@ -215,7 +215,7 @@ func TestExclusiveBoundIsBoolean_FollowsTheDialect(t *testing.T) {
 // decides whether an internal pointer names anything.
 func TestRefScope_IsTheContextSeenAsAScope(t *testing.T) {
 	t.Parallel()
-	c := lowering.New(0, docDeclaring("User"), ir.SourceInfo{Path: "spec.yaml"}, "", overlay.Origin{})
+	c := lowering.New(0, docDeclaring("User"), ir.SourceInfo{Path: "spec.yaml"}, "", lowering.StreamingMedia{}, overlay.Origin{})
 
 	scope := c.RefScope()
 
@@ -281,7 +281,7 @@ func TestSources_ListsTheOverlayAfterTheSourceItPatched(t *testing.T) {
 		"overlay: 1.0.0\ninfo: {title: O, version: \"1\"}\nactions:\n"+
 			"  - target: $.info\n    update: {description: d}\n")
 
-	c := lowering.New(0, docDeclaring(), src, "", origin)
+	c := lowering.New(0, docDeclaring(), src, "", lowering.StreamingMedia{}, origin)
 
 	require.Len(t, c.Sources(), 2)
 	assert.Equal(t, src, c.Sources()[0], "the source being lowered comes first")
@@ -296,7 +296,7 @@ func TestSources_ListsOnlyTheSourceWhenNoOverlayApplied(t *testing.T) {
 	t.Parallel()
 	src := ir.SourceInfo{Format: "openapi@3.1", Path: "spec.yaml"}
 
-	c := lowering.New(0, docDeclaring(), src, "", overlay.Origin{})
+	c := lowering.New(0, docDeclaring(), src, "", lowering.StreamingMedia{}, overlay.Origin{})
 
 	assert.Equal(t, []ir.SourceInfo{src}, c.Sources())
 }
@@ -312,7 +312,7 @@ func TestProvenanceAt_NamesTheOverlayForThePositionsItIntroduced(t *testing.T) {
 		"overlay: 1.0.0\ninfo: {title: O, version: \"1\"}\nactions:\n"+
 			"  - target: $.info\n    update: {description: d}\n")
 
-	c := lowering.New(0, docDeclaring(), ir.SourceInfo{}, "", origin)
+	c := lowering.New(0, docDeclaring(), ir.SourceInfo{}, "", lowering.StreamingMedia{}, origin)
 
 	assert.Equal(t, ir.Provenance{Source: 1, Pointer: "/info/description"},
 		c.ProvenanceAt("/info/description"), "the overlay introduced this position")

compilers/openapi/internal/lowering/streaming.go

@@ -0,0 +1,80 @@
+package lowering
+
+import "strings"
+
+// StreamingMediaTypeHeuristic is the name Provenance.Inferred carries on an
+// operation whose streaming was read out of a media type rather than declared.
+// It is a constant because the marker is what an auditor greps for, and a
+// spelling written at the producing site and again at a reading test can drift.
+const StreamingMediaTypeHeuristic = "streaming-media-type"
+
+// StreamingMedia is the media-type streaming policy: which media types mean
+// "this body is a sequence of frames" in a document that declares nothing
+// saying so.
+//
+// It is a policy rather than a table in the lowering because the reading is a
+// guess (architecture principle 6). OpenAPI below 3.2 has no keyword for a
+// sequential body at all, so an SSE or NDJSON API says what it does only by
+// naming a media type — and a media type is a content encoding, not a promise
+// about framing. The vocabulary is declared here, below both walks, and
+// re-exported by the compiler's public options for the reason GroupingStrategy
+// is: one declaration cannot drift from itself.
+type StreamingMedia struct {
+	// Disabled turns the inference off. Off means off: an operation then carries
+	// the streaming fields a 3.2 itemSchema declares and nothing else, which is
+	// what a caller who does not want guesses in their IR asked for.
+	Disabled bool `json:"disabled,omitempty"`
+	// MediaTypes replaces the default list rather than extending it, so a caller
+	// who states a list gets exactly that list. Empty means the default.
+	//
+	// Entries are matched against the media type alone: the comparison is
+	// case-insensitive and ignores parameters, because `text/event-stream` and
+	// `text/event-stream; charset=utf-8` name one type.
+	MediaTypes []string `json:"mediaTypes,omitempty"`
+}
+
+// DefaultStreamingMediaTypes is the list the policy uses when the caller states
+// none. It is a default and not a standard: `text/event-stream` is the only
+// registered one of the three, and the two JSON-lines spellings are conventions
+// that happen to be what generators in this space already look for. A document
+// using a fourth spelling is not wrong — it names its own list.
+func DefaultStreamingMediaTypes() []string {
+	return []string{"application/jsonl", "application/x-ndjson", "text/event-stream"}
+}
+
+// MediaTypeStreams reports whether the policy classifies mediaType as a stream
+// of frames. It is a predicate rather than a getter for the reason
+// DeclaresSchema is: handing back the set would make it writable through a copy
+// of the context.
+func (c Ctx) MediaTypeStreams(mediaType string) bool {
+	return c.streaming[normalizeMediaType(mediaType)]
+}
+
+// streamingSet normalizes a policy into the set MediaTypeStreams answers from,
+// or nil when the inference is off — which reads the same as an empty set, so
+// no lowering has to ask whether the policy was disabled or merely empty.
+func streamingSet(p StreamingMedia) map[string]bool {
+	if p.Disabled {
+		return nil
+	}
+	types := p.MediaTypes
+	if len(types) == 0 {
+		types = DefaultStreamingMediaTypes()
+	}
+	set := make(map[string]bool, len(types))
+	for _, mt := range types {
+		if normalized := normalizeMediaType(mt); normalized != "" {
+			set[normalized] = true
+		}
+	}
+	return set
+}
+
+// normalizeMediaType reduces a media type to the form the policy compares:
+// lowercased, with any parameters dropped.
+func normalizeMediaType(mediaType string) string {
+	if i := strings.IndexByte(mediaType, ';'); i >= 0 {
+		mediaType = mediaType[:i]
+	}
+	return strings.ToLower(strings.TrimSpace(mediaType))
+}

compilers/openapi/internal/lowering/streaming_test.go

@@ -0,0 +1,74 @@
+package lowering_test
+
+import (
+	"testing"
+
+	soa "github.com/speakeasy-api/openapi/openapi"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/dexpace/morphic/compilers/openapi/internal/lowering"
+	"github.com/dexpace/morphic/compilers/openapi/internal/overlay"
+	"github.com/dexpace/morphic/ir"
+)
+
+// streamingCtx builds a context carrying nothing but the streaming policy.
+func streamingCtx(policy lowering.StreamingMedia) lowering.Ctx {
+	return lowering.New(0, &soa.OpenAPI{}, ir.SourceInfo{}, "", policy, overlay.Origin{})
+}
+
+// TestMediaTypeStreams_AnswersFromThePolicy pins every answer the policy gives,
+// because each is a different decision: the default list, a caller's list
+// replacing it rather than extending it, the off switch, and the normalization
+// that makes one media type written two ways match once.
+func TestMediaTypeStreams_AnswersFromThePolicy(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		policy    lowering.StreamingMedia
+		mediaType string
+		want      bool
+	}{
+		{"default list", lowering.StreamingMedia{}, "text/event-stream", true},
+		{"default list, ordinary type", lowering.StreamingMedia{}, "application/json", false},
+		{"parameters ignored", lowering.StreamingMedia{}, "text/event-stream; charset=utf-8", true},
+		{"case ignored", lowering.StreamingMedia{}, "TEXT/Event-Stream", true},
+		{"surrounding space ignored", lowering.StreamingMedia{}, "  text/event-stream  ", true},
+		{"disabled", lowering.StreamingMedia{Disabled: true}, "text/event-stream", false},
+		{
+			"caller's list replaces the default",
+			lowering.StreamingMedia{MediaTypes: []string{"application/vnd.acme.frames"}},
+			"text/event-stream", false,
+		},
+		{
+			"caller's list is honoured",
+			lowering.StreamingMedia{MediaTypes: []string{"Application/VND.acme.frames"}},
+			"application/vnd.acme.frames", true,
+		},
+		{
+			"a blank entry names no media type",
+			lowering.StreamingMedia{MediaTypes: []string{"  ", "text/event-stream"}},
+			"", false,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, streamingCtx(tc.policy).MediaTypeStreams(tc.mediaType))
+		})
+	}
+}
+
+// TestDefaultStreamingMediaTypes_AreAllRecognized holds the exported default
+// list to the set the policy actually applies. Two transcriptions of one set is
+// one of them going stale unnoticed, and the exported one is what a caller
+// extending the list starts from.
+func TestDefaultStreamingMediaTypes_AreAllRecognized(t *testing.T) {
+	t.Parallel()
+	defaults := lowering.DefaultStreamingMediaTypes()
+	require.NotEmpty(t, defaults, "an empty default list would make every case below vacuous")
+	c := streamingCtx(lowering.StreamingMedia{})
+	for _, mediaType := range defaults {
+		assert.True(t, c.MediaTypeStreams(mediaType), "default media type %q is not recognized", mediaType)
+	}
+}

compilers/openapi/internal/operation/helpers_internal_test.go

@@ -44,7 +44,7 @@ func loweredFor(t *testing.T, src string) (*lowerer, []ir.Diagnostic) {
 	require.NotNil(t, loadedDoc, "load returned no document: %+v", diags)
 	types := compile.NewTypes(0)
 	return &lowerer{
-		ctx:          lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, overlay.Origin{}),
+		ctx:          lowering.New(0, loadedDoc.Doc, loadedDoc.Source, lowering.GroupByTags, lowering.StreamingMedia{}, overlay.Origin{}),
 		out:          &ir.Document{Types: types.Registry()},
 		types:        types,
 		operationIDs: make(map[string]string),
@@ -56,7 +56,7 @@ func loweredFor(t *testing.T, src string) (*lowerer, []ir.Diagnostic) {
 func newRawLowerer(doc *soa.OpenAPI) *lowerer {
 	types := compile.NewTypes(0)
 	return &lowerer{
-		ctx:          lowering.New(0, doc, ir.SourceInfo{}, "", overlay.Origin{}),
+		ctx:          lowering.New(0, doc, ir.SourceInfo{}, "", lowering.StreamingMedia{}, overlay.Origin{}),
 		out:          &ir.Document{Types: types.Registry()},
 		types:        types,
 		operationIDs: make(map[string]string),

compilers/openapi/internal/operation/helpers_test.go

@@ -200,7 +200,7 @@ func serviceWithGrouping(t *testing.T, src string, grouping lowering.GroupingStr
 	require.NotNil(t, loadedDoc)
 
 	types := compile.NewTypes(0)
-	c := lowering.New(0, loadedDoc.Doc, loadedDoc.Source, grouping, overlay.Origin{})
+	c := lowering.New(0, loadedDoc.Doc, loadedDoc.Source, grouping, lowering.StreamingMedia{}, overlay.Origin{})
 	var anchors schema.AnchorIndex
 	var acc compile.Diags
 	acc.AppendAll(schema.LowerComponentSchemas(c, types, &anchors))

compilers/openapi/internal/operation/operations.go

@@ -246,10 +246,11 @@ type opContext struct {
 // registered alongside it in the same group (ir-design §7.2, §8.1).
 func lowerOperation(c lowering.Ctx, ts *compile.Types, anchors *schema.AnchorIndex, operationIDs map[string]string, src *soa.Operation, opCtx opContext) (ir.Operation, []ir.Operation, []ir.Diagnostic) {
 	mount, decl := opCtx.ptrs.mount, opCtx.ptrs.decl
-	// Built through the context so the source index is spelled in one place, then
-	// marked inferred — the one provenance in this compiler that is.
+	// Built through the context so the source index is spelled in one place. Its
+	// heuristic marker is filled in below, once every lowering that can add one
+	// has run: the grouping choice is known here, but the streaming reading is
+	// not known until the payloads are lowered.
 	opProv := c.ProvenanceAt(decl)
-	opProv.Inferred = opCtx.inferred
 	opAuth, diags := auth.LowerSecurityRequirements(c, src.Security, decl)
 	op := ir.Operation{
 		ID:   ids.Op(mount),
@@ -280,6 +281,10 @@ func lowerOperation(c lowering.Ctx, ts *compile.Types, anchors *schema.AnchorInd
 		ParamBindings: bindings,
 	}
 	diags = append(diags, lowerRequestBody(c, ts, anchors, &op, &hb, src, decl)...)
+	// After both payload lowerings, which are what it reads.
+	streaming, streamDiags := applyStreaming(c, &op, decl)
+	diags = append(diags, streamDiags...)
+	op.Provenance.Inferred = joinInferred(opCtx.inferred, streaming)
 	var extra []ir.Operation
 	if opCtx.withCallbacks {
 		var cbDiags []ir.Diagnostic