coordinator: extend audio/video media support alongside images

config/coordinator/coordinator.yaml

@@ -45,6 +45,9 @@ server:
   # max_request_body_size caps the request body in megabytes. The default is
   # sized for multimodal requests that inline images as data: URIs, which land
   # in the body verbatim and are not subject to replace-media-urls.max_download_size.
+  # Deployments that also accept inline audio or video (input_audio parts or
+  # data:audio/data:video URIs in audio_url/video_url slots) may need a much
+  # higher value: a 100 MB video inlines as ~135 MB of base64 in the body.
   # A text-only deployment can set a much smaller value. Megabytes; default 64.
   # env: COORDINATOR_SERVER_MAX_REQUEST_BODY_SIZE
   # max_request_body_size: 64
@@ -101,36 +104,81 @@ pipeline:
 
   steps:
     # -------------------------------------------------------------------
-    # replace-media-urls: first step. Downloads any http(s) image_url
-    # references, base64-inlines them as data: URIs, and seeds
-    # MultimodalEntries on the request context.
+    # replace-media-urls: first step. Downloads http(s) URLs from image_url,
+    # audio_url, and video_url content parts (base64-inlining them as data:
+    # URIs), validates input_audio inline items, and seeds MultimodalEntries
+    # for every recognized media part. All modalities feed into the encode
+    # step's fanout — the encoder pod must accept audio/video content-part
+    # types (see H1 in coordinator-audio-video-plan.md).
     # -------------------------------------------------------------------
     - type: replace-media-urls
       params:
-        # download_timeout caps each individual image download.
+        # download_timeout caps each individual media download.
         download_timeout: 10s
 
         # max_concurrent_downloads bounds the errgroup that fans out
-        # downloads. Useful when a single request references many images.
+        # downloads. Useful when a single request references many media items.
         max_concurrent_downloads: 10
 
-        # max_multimodal_entries rejects requests with more than N image_url
-        # parts BEFORE any download happens. Cheapest admission gate.
+        # max_multimodal_entries rejects requests with more than N media
+        # parts (image_url + audio_url + video_url + input_audio combined)
+        # BEFORE any download happens. Cheapest admission gate.
         # 0 (or omitted) = unlimited.
         # max_multimodal_entries: 8
 
-        # max_download_size caps the megabytes read per image download. A response
+        # max_download_size caps the megabytes read per media download. A response
         # whose Content-Length or body exceeds the cap is rejected as a client
         # error (4xx). Bounds peak memory together with the two limits above:
         # roughly max_concurrent_downloads x max_download_size x ~2.3 MB.
-        # data: URIs do not hit the network and are unaffected. Default 10 MB.
+        # data: URIs do not hit the network and are unaffected. This value is
+        # the global default; each modality below may override it. Default 10 MB.
         # max_download_size: 10
 
+        # max_image_download_size overrides max_download_size specifically for
+        # image_url items. Same units (MB), same overflow validation.
+        # max_image_download_size: 10
+
+        # max_audio_download_size overrides max_download_size for audio_url
+        # items and for input_audio inline payloads (checked against the
+        # decoded byte count implied by the base64 length). Audio codec
+        # decoders historically carry more CVEs than image decoders; keep
+        # this cap tight until your security team has reviewed the audio
+        # decode path end-to-end (H2 in coordinator-audio-video-plan.md).
+        # max_audio_download_size: 60
+
+        # max_video_download_size overrides max_download_size for video_url
+        # items. Videos routinely need 100+ MB per clip; leave commented out
+        # only if your deployment truly does not accept video. Same codec
+        # CVE caveat as audio applies — treat pending security review as a
+        # gate on raising this in production.
+        # max_video_download_size: 200
+
+        # allowed_image_content_types overrides the built-in image MIME
+        # allowlist for image_url data URIs. When set, ONLY these types are
+        # accepted. Default: image/jpeg, image/png, image/gif, image/webp.
+        # allowed_image_content_types:
+        #   - image/jpeg
+        #   - image/png
+
+        # allowed_audio_content_types overrides the built-in audio MIME
+        # allowlist. Applies to audio_url data URIs and to input_audio parts.
+        # Default: audio/wav, audio/x-wav, audio/mpeg, audio/mp3, audio/flac,
+        # audio/x-flac, audio/ogg, audio/opus, audio/webm.
+        # allowed_audio_content_types:
+        #   - audio/wav
+        #   - audio/mpeg
+
+        # allowed_video_content_types overrides the built-in video MIME
+        # allowlist for video_url data URIs. Default: video/mp4, video/webm,
+        # video/quicktime, video/mpeg, video/ogg.
+        # allowed_video_content_types:
+        #   - video/mp4
+
         # allow_private_networks permits downloads whose resolved IP falls in
         # an RFC1918 private range (10/8, 172.16/12, 192.168/16). Loopback,
         # link-local (including the 169.254.169.254 cloud metadata endpoint),
         # CGNAT, and unique-local addresses stay blocked regardless. Enable
-        # only when image origins are inside the cluster network.
+        # only when media origins are inside the cluster network.
         # Default false.
         # allow_private_networks: false
 

pkg/coordinator/pipeline/context.go

@@ -93,11 +93,14 @@ type RequestContext struct {
 
 // MultimodalEntry describes one downloaded multimodal item (e.g. an image) and
 // where it sits in the tokenized prompt. Index is its position in the request's
-// multimodal list. Base64Data and ContentType come from the media download;
-// Hash and KwargsData are filled in by the render step; Placeholder marks the
-// span of placeholder tokens the encode step replaces.
+// multimodal list. Modality identifies the OpenAI content-part type this entry
+// originated from — one of the steps.Modality* constants ("image", "audio",
+// "video"). Base64Data and ContentType come from the media download; Hash and
+// KwargsData are filled in by the render step; Placeholder marks the span of
+// placeholder tokens the encode step replaces.
 type MultimodalEntry struct {
 	Index       int
+	Modality    string
 	Hash        string
 	Base64Data  string
 	ContentType string

pkg/coordinator/steps/decode.go

@@ -134,13 +134,32 @@ func (s *DecodeStep) injectTokensField(reqCtx *pipeline.RequestContext) {
 	reqCtx.Body["tokens"] = tokens
 }
 
+// injectUUIDs tags each media content part with the uuid the decode
+// backend uses for prefix-cache keying. Each part is paired with the
+// MultimodalEntry that shares its modality and local index — the same
+// invariant the encode fanout uses (see localIndexOf in utils.go).
+// Non-media parts (text, tool_use, unknown types) are skipped.
 func (s *DecodeStep) injectUUIDs(reqCtx *pipeline.RequestContext) {
 	messages, ok := reqCtx.Body["messages"].([]any)
 	if !ok {
 		return
 	}
 
-	hashIdx := 0
+	// Build a (modality, localIndex) → Hash lookup once so per-part
+	// tagging stays O(1) even on requests with many media items.
+	hashByModLocalIdx := make(map[string]map[int]string)
+	for i, entry := range reqCtx.MultimodalEntries {
+		mod := entryModality(entry)
+		localIdx := localIndexOf(reqCtx.MultimodalEntries, i)
+		if _, ok := hashByModLocalIdx[mod]; !ok {
+			hashByModLocalIdx[mod] = make(map[int]string)
+		}
+		hashByModLocalIdx[mod][localIdx] = entry.Hash
+	}
+
+	// Walk parts, incrementing a per-modality counter so each media part
+	// gets the hash of the entry at the matching (modality, localIndex).
+	modCounter := make(map[string]int)
 	for _, msg := range messages {
 		msgMap, ok := msg.(map[string]any)
 		if !ok {
@@ -155,12 +174,15 @@ func (s *DecodeStep) injectUUIDs(reqCtx *pipeline.RequestContext) {
 			if !ok {
 				continue
 			}
-			if partMap["type"] != "image_url" {
+			partType, _ := partMap["type"].(string)
+			modality, isMedia := partTypeModality[partType]
+			if !isMedia {
 				continue
 			}
-			if hashIdx < len(reqCtx.MultimodalEntries) {
-				partMap["uuid"] = reqCtx.MultimodalEntries[hashIdx].Hash
-				hashIdx++
+			localIdx := modCounter[modality]
+			modCounter[modality]++
+			if h, ok := hashByModLocalIdx[modality][localIdx]; ok {
+				partMap["uuid"] = h
 			}
 		}
 	}

pkg/coordinator/steps/decode_test.go

@@ -121,7 +121,7 @@ func TestDecodeStep_NonStreaming(t *testing.T) {
 		Stream:       false,
 		TokenIDs:     []int{1, 32000, 32000, 32000, 2345},
 		MultimodalEntries: []pipeline.MultimodalEntry{
-			{Index: 0, Hash: "hash-a", Placeholder: pipeline.PlaceholderRange{Offset: 1, Length: 3}},
+			{Index: 0, Modality: ModalityImage, Hash: "hash-a", Placeholder: pipeline.PlaceholderRange{Offset: 1, Length: 3}},
 		},
 		KVTransferParams: map[string]any{"block_id": "xyz", "peer_host": "10.0.0.5", "peer_port": 7777},
 		Body: map[string]any{
@@ -300,7 +300,7 @@ func TestDecodeStep_Streaming(t *testing.T) {
 		Model:        "test",
 		Stream:       true,
 		MultimodalEntries: []pipeline.MultimodalEntry{
-			{Index: 0, Hash: "h1"},
+			{Index: 0, Modality: ModalityImage, Hash: "h1"},
 		},
 		KVTransferParams: map[string]any{},
 		Body:             map[string]any{"model": "test", "stream": true},
@@ -345,7 +345,7 @@ func TestDecodeStep_GatewayError(t *testing.T) {
 		Model:        "test",
 		Stream:       false,
 		MultimodalEntries: []pipeline.MultimodalEntry{
-			{Index: 0, Hash: "h1"},
+			{Index: 0, Modality: ModalityImage, Hash: "h1"},
 		},
 		KVTransferParams: map[string]any{},
 		Body:             map[string]any{"model": "test", "stream": false},
@@ -446,3 +446,79 @@ func TestDecodeStep_TransportError(t *testing.T) {
 		t.Fatalf("expected ErrorHandler-written 502, got %d", result.StatusCode)
 	}
 }
+
+// ---- Section 5: injectUUIDs widened to audio/video --------------------------
+
+// TestInjectUUIDs_TagsAllMediaParts asserts every recognized media
+// content-part type receives a uuid tag matching its (modality, local index)
+// entry in MultimodalEntries. Non-media parts (text, unknown types) are
+// left alone.
+func TestInjectUUIDs_TagsAllMediaParts(t *testing.T) {
+	step := &DecodeStep{}
+	imagePart := map[string]any{"type": "image_url", "image_url": map[string]any{"url": "u-img"}}
+	audioURLPart := map[string]any{"type": "audio_url", "audio_url": map[string]any{"url": "u-aud"}}
+	inputAudioPart := map[string]any{"type": "input_audio", "input_audio": map[string]any{"data": "d", "format": "wav"}}
+	videoPart := map[string]any{"type": "video_url", "video_url": map[string]any{"url": "u-vid"}}
+	textPart := map[string]any{"type": "text", "text": "hi"}
+
+	reqCtx := &pipeline.RequestContext{
+		Body: map[string]any{
+			"messages": []any{
+				map[string]any{
+					"role":    "user",
+					"content": []any{textPart, imagePart, audioURLPart, videoPart, inputAudioPart},
+				},
+			},
+		},
+		MultimodalEntries: []pipeline.MultimodalEntry{
+			{Index: 0, Modality: ModalityImage, Hash: "H-img"},
+			{Index: 1, Modality: ModalityAudio, Hash: "H-aud-url"},
+			{Index: 2, Modality: ModalityVideo, Hash: "H-vid"},
+			{Index: 3, Modality: ModalityAudio, Hash: "H-input-audio"},
+		},
+	}
+	step.injectUUIDs(reqCtx)
+
+	if got := imagePart["uuid"]; got != "H-img" {
+		t.Errorf("image uuid = %v, want H-img", got)
+	}
+	if got := audioURLPart["uuid"]; got != "H-aud-url" {
+		t.Errorf("audio_url uuid = %v, want H-aud-url", got)
+	}
+	if got := inputAudioPart["uuid"]; got != "H-input-audio" {
+		t.Errorf("input_audio uuid = %v, want H-input-audio", got)
+	}
+	if got := videoPart["uuid"]; got != "H-vid" {
+		t.Errorf("video uuid = %v, want H-vid", got)
+	}
+	if _, ok := textPart["uuid"]; ok {
+		t.Errorf("text part must not be tagged: %+v", textPart)
+	}
+}
+
+// TestInjectUUIDs_TagsRepeatedModalityInOrder asserts that two audio parts
+// in the same request receive the hashes of the two audio entries, in
+// walker order.
+func TestInjectUUIDs_TagsRepeatedModalityInOrder(t *testing.T) {
+	step := &DecodeStep{}
+	aud0 := map[string]any{"type": "audio_url", "audio_url": map[string]any{"url": "u0"}}
+	aud1 := map[string]any{"type": "audio_url", "audio_url": map[string]any{"url": "u1"}}
+	reqCtx := &pipeline.RequestContext{
+		Body: map[string]any{
+			"messages": []any{
+				map[string]any{"role": "user", "content": []any{aud0, aud1}},
+			},
+		},
+		MultimodalEntries: []pipeline.MultimodalEntry{
+			{Index: 0, Modality: ModalityAudio, Hash: "H0"},
+			{Index: 1, Modality: ModalityAudio, Hash: "H1"},
+		},
+	}
+	step.injectUUIDs(reqCtx)
+	if got := aud0["uuid"]; got != "H0" {
+		t.Errorf("audio[0] uuid = %v, want H0", got)
+	}
+	if got := aud1["uuid"]; got != "H1" {
+		t.Errorf("audio[1] uuid = %v, want H1", got)
+	}
+}