feat: add gcs_uri multipart RequestTransform plugin (#256) (#260)

* feat: add gcs_uri multipart RequestTransform plugin

First concrete consumer of the request body-transform framework (#256,
building on #257). Multi-modal endpoints (Whisper transcription, OCR)
expect multipart/form-data with a `url` field pointing at a signed object
URL, not the OpenAI-style JSON the pipeline produces. Producers can't put
raw media on the broker, so the queued payload carries a signed URL in a
`gcs_uri` field; this plugin rewrites the body at dispatch time.

The gcs_uri_multipart plugin (pkg/asyncworker/transform/gcsmultipart):

- Activates only when metadata.provider matches a configured provider AND
  the payload has a non-empty gcs_uri; otherwise the default JSON path is
  preserved unchanged.
- Transform: writes gcs_uri as a `url` form field (a plain field, not a
  file upload), passes remaining fields through, drops gcs_uri, and
  rejects a non-empty file_base64 as fatal/non-retryable.
- Validate: parses the signed URL expiry (V4 X-Goog-Date/X-Goog-Expires
  or V2 Expires) and fails fatally before dispatch if it expires at or
  before the request deadline, so the broker doesn't retry a doomed
  request. Expiry parsing is best-effort and needs no new dependency.

Registered via init()/MustRegister and a blank import in cmd/main.go.
Configured under requestTransforms with a `providers` parameter. Adds a
README section documenting the flag, config shape, and plugin behavior.

Signed-off-by: Shimi Bandiel <shimib@google.com>

* feat(chart): wire request body-transform config

Address review feedback on #260: the chart had no way to configure the new
body-transform plugins. Add ap.transformConfig — rendered to transform-config.json
in the config ConfigMap and passed via --transform-config-file, mirroring how
worker-pools.json and pubsub-topics.json are mounted. Empty (the default) leaves
behavior unchanged.

Document the Helm path in the README and add helm-unittest coverage for the
set and empty cases.

Signed-off-by: Shimi Bandiel <shimib@google.com>

---------

Signed-off-by: Shimi Bandiel <shimib@google.com>

Author: shimib

README.md

@@ -38,6 +38,8 @@ The architecture adheres to the following core principles:
     - [Per-Queue Dispatch Gates](#per-queue-dispatch-gates)
   - [Request Messages and Consumption](#request-messages-and-consumption)
     - [Request Merge Policy](#request-merge-policy)
+  - [Request Body Transforms](#request-body-transforms)
+    - [`gcs_uri_multipart` plugin](#gcs_uri_multipart-plugin)
   - [Retries](#retries)
   - [Observability](#observability)
     - [OpenTelemetry Tracing](#opentelemetry-tracing)
@@ -347,6 +349,46 @@ This per-pool topology provides complete backpressure and queue-level isolation:
 
 Currently the only policy supported is `Random Robin Policy` which randomly picks messages from all queues configured for a given pool.
 
+## Request Body Transforms
+
+By default the worker dispatches the OpenAI-style JSON marshalled from a request's `payload`. Some providers need a different body shape at dispatch time — for example multi-modal endpoints (Whisper transcription, OCR) that expect `multipart/form-data` with a `url` field rather than JSON. Request body-transform plugins handle this without special-casing the worker: they rewrite the outgoing body and `Content-Type` based on per-message `metadata`, and the default JSON path is preserved byte-for-byte when no plugin applies.
+
+Transforms are configured with `--transform-config-file`, pointing at a JSON object that groups plugins by direction:
+
+```json
+{
+  "requestTransforms": [
+    {
+      "name": "whisper-multipart",
+      "type": "gcs_uri_multipart",
+      "parameters": { "providers": ["whisper"] }
+    }
+  ]
+}
+```
+
+Each entry has a unique `name`, a registered plugin `type`, and opaque `parameters`. Unknown top-level fields are rejected. When the flag is empty, no transforms are loaded and behavior is unchanged.
+
+With the Helm chart, set `ap.transformConfig` to this same object; the chart renders it to a config file and wires `--transform-config-file` automatically:
+
+```yaml
+ap:
+  transformConfig:
+    requestTransforms:
+      - name: "whisper-multipart"
+        type: "gcs_uri_multipart"
+        parameters:
+          providers: ["whisper"]
+```
+
+### `gcs_uri_multipart` plugin
+
+Rewrites a JSON body into `multipart/form-data` for endpoints that take a signed object URL. Because producers can't put raw media bytes on the broker, the queued `payload` carries a signed URL (e.g. a GCS V4 signed URL) in a `gcs_uri` field.
+
+- **Activation:** the message's `metadata.provider` must match one of the configured `providers`, and the `payload` must contain a non-empty `gcs_uri`. Otherwise the default JSON path is used unchanged.
+- **Transform:** writes the `gcs_uri` value as a `url` form field (a plain field, not a file upload), passes the remaining payload fields through as form fields, and drops `gcs_uri`. A non-empty `file_base64` is rejected as a fatal, non-retryable error (inline media is not supported on this path).
+- **Preflight:** parses the signed URL's expiry (V4 `X-Goog-Date` + `X-Goog-Expires`, or V2 `Expires`); if it expires at or before the message deadline, the request fails fatally before dispatch so the broker doesn't retry a request that cannot succeed.
+
 ## Retries
 
 When a message processing has failed, either shedded or due to a server-side error, it will be scheduled for a retry (assuming the deadline has not passed).

charts/async-processor/templates/ap-configmap.yaml

@@ -1,4 +1,4 @@
-{{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig }}
+{{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig .Values.ap.transformConfig }}
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -13,4 +13,7 @@ data:
   {{- if .Values.ap.gcpPubSub.topicsConfig }}
   pubsub-topics.json: {{ .Values.ap.gcpPubSub.topicsConfig | toJson | quote }}
   {{- end }}
+  {{- if .Values.ap.transformConfig }}
+  transform-config.json: {{ .Values.ap.transformConfig | toJson | quote }}
+  {{- end }}
 {{- end }}

charts/async-processor/templates/ap-deployments.yaml

@@ -92,6 +92,9 @@ spec:
           {{- if .Values.ap.workerPools }}
           - --pool-config-file=/etc/async-processor/config/worker-pools.json
           {{- end }}
+          {{- if .Values.ap.transformConfig }}
+          - --transform-config-file=/etc/async-processor/config/transform-config.json
+          {{- end }}
           - --concurrency={{ .Values.ap.concurrency | default 8 }}
           - --drain-timeout={{ .Values.ap.drainTimeout | default "2m" }}
           - --health-port={{ .Values.ap.health.port | default 8081 }}
@@ -184,9 +187,9 @@ spec:
         resources:
           {{- toYaml . | nindent 12 }}
         {{- end }}
-        {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig .Values.ap.tls.secretName }}
+        {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig .Values.ap.transformConfig .Values.ap.tls.secretName }}
         volumeMounts:
-          {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig }}
+          {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig .Values.ap.transformConfig }}
           - name: ap-config
             mountPath: /etc/async-processor/config
             readOnly: true
@@ -199,9 +202,9 @@ spec:
         {{- end }}
       serviceAccountName: {{ include "async-processor.fullname" . }}
       terminationGracePeriodSeconds: 130
-      {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig .Values.ap.tls.secretName }}
+      {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig .Values.ap.transformConfig .Values.ap.tls.secretName }}
       volumes:
-        {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig }}
+        {{- if or .Values.ap.workerPools .Values.ap.gcpPubSub.topicsConfig .Values.ap.transformConfig }}
         - name: ap-config
           configMap:
             name: {{ include "async-processor.fullname" . }}-config

charts/async-processor/tests/deployment_test.yaml

@@ -308,3 +308,28 @@ tests:
       - contains:
           path: spec.template.spec.containers[0].args
           content: --message-queue-impl=gcp-pubsub-gated
+
+  - it: should pass transform-config-file and mount the config when transformConfig is set
+    set:
+      ap.transformConfig:
+        requestTransforms:
+          - name: "gcs-whisper"
+            type: "gcs_uri_multipart"
+            parameters:
+              providers: ["whisper"]
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].args
+          content: --transform-config-file=/etc/async-processor/config/transform-config.json
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: ap-config
+            mountPath: /etc/async-processor/config
+            readOnly: true
+
+  - it: should not pass transform-config-file when transformConfig is empty
+    asserts:
+      - notContains:
+          path: spec.template.spec.containers[0].args
+          content: --transform-config-file=/etc/async-processor/config/transform-config.json

charts/async-processor/values.yaml

@@ -29,6 +29,17 @@ ap:
   #   - id: "premium"
   #     workers: 16
   workerPools: []
+  # Request body-transform plugins. When set, rendered to a config file and passed
+  # via --transform-config-file. Object with a requestTransforms array; each entry
+  # has a unique name, a registered plugin type, and opaque parameters. Empty
+  # disables transforms (default JSON dispatch is unchanged).
+  # transformConfig:
+  #   requestTransforms:
+  #     - name: "gcs-whisper"
+  #       type: "gcs_uri_multipart"
+  #       parameters:
+  #         providers: ["whisper"]
+  transformConfig: {}
   otel:
     # OTLP gRPC endpoint for trace collection. Examples:
     #   Dev (Jaeger all-in-one):     "http://jaeger.default.svc.cluster.local:4317"

cmd/main.go

@@ -20,6 +20,8 @@ import (
 	"github.com/llm-d-incubation/llm-d-async/pkg/async/inference/flowcontrol"
 	"github.com/llm-d-incubation/llm-d-async/pkg/asyncworker"
 	"github.com/llm-d-incubation/llm-d-async/pkg/asyncworker/transform"
+	// Register the built-in request body-transform plugins.
+	_ "github.com/llm-d-incubation/llm-d-async/pkg/asyncworker/transform/gcsmultipart"
 	"github.com/llm-d-incubation/llm-d-async/pkg/metrics"
 	"github.com/llm-d-incubation/llm-d-async/pkg/plugins"
 	"github.com/llm-d-incubation/llm-d-async/pkg/pubsub"