feat(monetize): ServiceOffer type=dataset — CRD, catalog & x402 extras wiring (P1)

internal/embed/embed_crd_test.go

@@ -149,8 +149,9 @@ func TestServiceOfferCRD_Fields(t *testing.T) {
 	}
 
 	// Required fields in spec (aligned with x402/ERC-8004 schema). agent
-	// joins this list as part of the type=agent offer flow.
-	for _, field := range []string{"type", "agent", "model", "upstream", "payment", "path", "registration"} {
+	// joins this list as part of the type=agent offer flow; dataset joins
+	// it for the type=dataset offer flow.
+	for _, field := range []string{"type", "agent", "dataset", "model", "upstream", "payment", "path", "registration"} {
 		if _, exists := pm[field]; !exists {
 			t.Errorf("spec.properties missing field %q", field)
 		}
@@ -162,13 +163,64 @@ func TestServiceOfferCRD_Fields(t *testing.T) {
 	for _, v := range enum {
 		got[v.(string)] = true
 	}
-	for _, want := range []string{"inference", "fine-tuning", "http", "agent"} {
+	for _, want := range []string{"inference", "fine-tuning", "http", "agent", "dataset"} {
 		if !got[want] {
 			t.Errorf("spec.type.enum missing %q", want)
 		}
 	}
 }
 
+// TestServiceOfferCRD_DatasetFields pins the type=dataset schema surface:
+// the spec.dataset block carries the pinned artifact metadata (mirroring
+// spec.agent), and price.perMB enables per-megabyte pricing. Mirrors
+// TestServiceOfferCRD_Fields' navigation.
+func TestServiceOfferCRD_DatasetFields(t *testing.T) {
+	data, err := ReadInfrastructureFile("base/templates/serviceoffer-crd.yaml")
+	if err != nil {
+		t.Fatalf("ReadInfrastructureFile: %v", err)
+	}
+
+	crd := findDoc(multiDoc(data), "CustomResourceDefinition")
+	if crd == nil {
+		t.Fatal("no CRD document found")
+	}
+
+	versions, ok := nested(crd, "spec", "versions").([]any)
+	if !ok || len(versions) == 0 {
+		t.Fatal("spec.versions is empty or wrong type")
+	}
+	v0, ok := versions[0].(map[string]any)
+	if !ok {
+		t.Fatal("versions[0] is not a map")
+	}
+
+	specProps, ok := nested(v0, "schema", "openAPIV3Schema", "properties", "spec", "properties").(map[string]any)
+	if !ok {
+		t.Fatal("spec.properties is not a map")
+	}
+
+	datasetProps, ok := nested(specProps, "dataset", "properties").(map[string]any)
+	if !ok {
+		t.Fatal("spec.dataset.properties missing — type=dataset offers can't pin a version")
+	}
+	for _, field := range []string{"manifestHash", "version", "fileHash", "sizeBytes"} {
+		if _, exists := datasetProps[field]; !exists {
+			t.Errorf("spec.dataset.properties missing field %q", field)
+		}
+	}
+	if sb, ok := datasetProps["sizeBytes"].(map[string]any); ok && sb["type"] != "integer" {
+		t.Errorf("spec.dataset.sizeBytes type = %v, want integer", sb["type"])
+	}
+
+	priceProps, ok := nested(specProps, "payment", "properties", "price", "properties").(map[string]any)
+	if !ok {
+		t.Fatal("spec.payment.price.properties missing")
+	}
+	if _, exists := priceProps["perMB"]; !exists {
+		t.Error("spec.payment.price.properties missing perMB — dataset offers can't price per-MB")
+	}
+}
+
 func TestServiceOfferCRD_PrinterColumns(t *testing.T) {
 	data, err := ReadInfrastructureFile("base/templates/serviceoffer-crd.yaml")
 	if err != nil {

internal/embed/infrastructure/base/templates/serviceoffer-crd.yaml

@@ -79,6 +79,34 @@ spec:
                     - namespace
                     type: object
                 type: object
+              dataset:
+                description: |-
+                  Populated when type='dataset'. Pins the versioned dataset artifact
+                  (an export bundle) the offer sells: the content-address anchor
+                  (manifestHash), the published version, and the artifact size. The
+                  controller surfaces these in the 402 response's extra.dataset block.
+                properties:
+                  fileHash:
+                    description: |-
+                      SHA-256 of the served file, for whole-file integrity verification
+                      after download.
+                    type: string
+                  manifestHash:
+                    description: |-
+                      Content-address anchor of the artifact: the export bundle
+                      manifestHash (SHA-256 over the artifact contents).
+                    type: string
+                  sizeBytes:
+                    description: Size of the served artifact in bytes. Drives per-MB
+                      pricing.
+                    format: int64
+                    minimum: 0
+                    type: integer
+                  version:
+                    description: Monotonic published version tag of the dataset (e.g.
+                      "1", "2").
+                    type: string
+                type: object
               drainAt:
                 description: |-
                   DrainAt marks the offer as draining when non-nil. While the offer
@@ -178,6 +206,9 @@ spec:
                       perHour:
                         description: Per-compute-hour price in USDC. Fine-tuning only.
                         type: string
+                      perMB:
+                        description: Per-megabyte price in USDC. Dataset only.
+                        type: string
                       perMTok:
                         description: Per-million-tokens price in USDC. Inference only.
                         type: string
@@ -278,12 +309,14 @@ spec:
                 description: |-
                   Service type. 'inference' enables model management; 'http' for any HTTP
                   service; 'agent' references an Agent CR via spec.agent.ref and the
-                  controller derives upstream + model + skills from the agent's status.
+                  controller derives upstream + model + skills from the agent's status;
+                  'dataset' sells a versioned dataset artifact via spec.dataset.
                 enum:
                 - inference
                 - fine-tuning
                 - http
                 - agent
+                - dataset
                 type: string
               upstream:
                 description: In-cluster service that handles the actual workload.

internal/monetizeapi/types.go

@@ -101,16 +101,23 @@ type ServiceOfferList struct {
 type ServiceOfferSpec struct {
 	// Service type. 'inference' enables model management; 'http' for any HTTP
 	// service; 'agent' references an Agent CR via spec.agent.ref and the
-	// controller derives upstream + model + skills from the agent's status.
+	// controller derives upstream + model + skills from the agent's status;
+	// 'dataset' sells a versioned dataset artifact via spec.dataset.
 	// +kubebuilder:default="http"
-	// +kubebuilder:validation:Enum=inference;fine-tuning;http;agent
+	// +kubebuilder:validation:Enum=inference;fine-tuning;http;agent;dataset
 	Type string `json:"type,omitempty"`
 
 	// Required when type='agent'. The controller resolves spec.agent.ref to
 	// the referenced Agent CR, derives upstream from Agent.status.endpoint,
 	// and surfaces the agent's pinned model + skills in the 402 response.
 	Agent ServiceOfferAgent `json:"agent,omitempty"`
 
+	// Populated when type='dataset'. Pins the versioned dataset artifact
+	// (an export bundle) the offer sells: the content-address anchor
+	// (manifestHash), the published version, and the artifact size. The
+	// controller surfaces these in the 402 response's extra.dataset block.
+	Dataset ServiceOfferDataset `json:"dataset,omitempty"`
+
 	// LLM model metadata. Required when the upstream serves an LLM.
 	Model ServiceOfferModel `json:"model,omitempty"`
 
@@ -164,6 +171,24 @@ type ServiceOfferAgentRef struct {
 	Namespace string `json:"namespace"`
 }
 
+// ServiceOfferDataset is populated when Spec.Type == "dataset". It pins the
+// versioned dataset artifact (an export bundle) the offer sells. The
+// controller surfaces these fields in the 402 response's extra.dataset block
+// so buyers see exactly which content-addressed version they're paying for.
+type ServiceOfferDataset struct {
+	// Content-address anchor of the artifact: the export bundle
+	// manifestHash (SHA-256 over the artifact contents).
+	ManifestHash string `json:"manifestHash,omitempty"`
+	// Monotonic published version tag of the dataset (e.g. "1", "2").
+	Version string `json:"version,omitempty"`
+	// SHA-256 of the served file, for whole-file integrity verification
+	// after download.
+	FileHash string `json:"fileHash,omitempty"`
+	// Size of the served artifact in bytes. Drives per-MB pricing.
+	// +kubebuilder:validation:Minimum=0
+	SizeBytes int64 `json:"sizeBytes,omitempty"`
+}
+
 type ServiceOfferModel struct {
 	// Model identifier (e.g. qwen3.5:35b).
 	// +kubebuilder:validation:Required
@@ -245,6 +270,8 @@ type ServiceOfferPriceTable struct {
 	PerHour string `json:"perHour,omitempty"`
 	// Per-training-epoch price in USDC. Fine-tuning only.
 	PerEpoch string `json:"perEpoch,omitempty"`
+	// Per-megabyte price in USDC. Dataset only.
+	PerMB string `json:"perMB,omitempty"`
 }
 
 type ServiceOfferRegistration struct {
@@ -423,6 +450,13 @@ func (o *ServiceOffer) IsAgent() bool {
 	return o.Spec.Type == "agent"
 }
 
+// IsDataset reports whether the offer sells a versioned dataset artifact.
+// Type=="dataset" is the only signal; spec.dataset carries the pinned
+// version metadata surfaced in the 402 extra block.
+func (o *ServiceOffer) IsDataset() bool {
+	return o.Spec.Type == "dataset"
+}
+
 // IsDraining reports whether spec.drainAt has been set. Drained offers
 // transition through three phases: pre-drain (DrainAt nil), draining
 // (DrainAt set, now < DrainEndsAt), and drain-expired (DrainAt set,
@@ -724,8 +758,7 @@ type AgentIdentityList struct {
 	Items           []AgentIdentity `json:"items"`
 }
 
-type AgentIdentitySpec struct {
-}
+type AgentIdentitySpec struct{}
 
 type AgentIdentityStatus struct {
 	// Per-chain ERC-8004 registrations for this identity document.

internal/schemas/service-catalog.schema.json

@@ -95,7 +95,8 @@
             "inference",
             "fine-tuning",
             "http",
-            "agent"
+            "agent",
+            "dataset"
           ]
         },
         "model": {
@@ -119,7 +120,8 @@
             "perRequest",
             "perMTok",
             "perHour",
-            "perEpoch"
+            "perEpoch",
+            "perMB"
           ]
         },
         "priceAtomicUnits": {
@@ -165,6 +167,21 @@
           "type": "string",
           "format": "date-time",
           "description": "RFC3339 timestamp at which the offer's HTTPRoute will be torn down. Set only when the offer is draining. Catalog consumers should detect drain via the presence of this field."
+        },
+        "datasetManifestHash": {
+          "type": "string",
+          "minLength": 1,
+          "description": "type=dataset only: content-address anchor (export bundle manifestHash, SHA-256) of the sold artifact."
+        },
+        "datasetVersion": {
+          "type": "string",
+          "minLength": 1,
+          "description": "type=dataset only: monotonic published version tag of the dataset."
+        },
+        "datasetSizeBytes": {
+          "type": "integer",
+          "minimum": 0,
+          "description": "type=dataset only: size of the served artifact in bytes."
         }
       }
     }

internal/schemas/service_catalog.go

@@ -53,6 +53,15 @@ type ServiceCatalogEntry struct {
 	// purely additive vs. pre-drain catalogs. Buyers SHOULD migrate to
 	// alternative providers before this time.
 	DrainEndsAt string `json:"drainEndsAt,omitempty"`
+
+	// Dataset* fields are populated for type=dataset offers, mirroring how
+	// Model is populated for inference/agent. They expose the pinned,
+	// content-addressed dataset version on discovery surfaces so buyers
+	// know exactly which artifact (and version) an offer sells. Additive
+	// only — see the stable-wire-schema note above.
+	DatasetManifestHash string `json:"datasetManifestHash,omitempty"`
+	DatasetVersion      string `json:"datasetVersion,omitempty"`
+	DatasetSizeBytes    int64  `json:"datasetSizeBytes,omitempty"`
 }
 
 // ServiceCatalogAsset describes the settlement token resolved for a catalog

internal/serviceoffercontroller/render.go

@@ -1162,6 +1162,14 @@ func buildServiceCatalogJSON(offers []*monetizeapi.ServiceOffer, baseURL string)
 			}
 		}
 
+		// type=dataset offers surface the pinned, content-addressed version
+		// on discovery, mirroring how Model is surfaced for inference/agent.
+		if offer.IsDataset() {
+			svc.DatasetManifestHash = offer.Spec.Dataset.ManifestHash
+			svc.DatasetVersion = offer.Spec.Dataset.Version
+			svc.DatasetSizeBytes = offer.Spec.Dataset.SizeBytes
+		}
+
 		services = append(services, svc)
 	}
 
@@ -1173,8 +1181,8 @@ func buildServiceCatalogJSON(offers []*monetizeapi.ServiceOffer, baseURL string)
 }
 
 // offerPriceRawAndUnit returns the raw decimal price string and which slot it
-// occupies in the price table. Only one of perRequest / perMTok / perHour is
-// expected to be set on a given offer.
+// occupies in the price table. Only one of perRequest / perMTok / perHour /
+// perMB is expected to be set on a given offer.
 func offerPriceRawAndUnit(offer *monetizeapi.ServiceOffer) (string, string) {
 	switch {
 	case offer.Spec.Payment.Price.PerRequest != "":
@@ -1183,6 +1191,8 @@ func offerPriceRawAndUnit(offer *monetizeapi.ServiceOffer) (string, string) {
 		return offer.Spec.Payment.Price.PerMTok, "perMTok"
 	case offer.Spec.Payment.Price.PerHour != "":
 		return offer.Spec.Payment.Price.PerHour, "perHour"
+	case offer.Spec.Payment.Price.PerMB != "":
+		return offer.Spec.Payment.Price.PerMB, "perMB"
 	default:
 		return "", ""
 	}
@@ -1352,6 +1362,8 @@ func describeOfferPrice(offer *monetizeapi.ServiceOffer) string {
 		return offer.Spec.Payment.Price.PerMTok + " " + symbol + "/MTok"
 	case offer.Spec.Payment.Price.PerHour != "":
 		return offer.Spec.Payment.Price.PerHour + " " + symbol + "/hour"
+	case offer.Spec.Payment.Price.PerMB != "":
+		return offer.Spec.Payment.Price.PerMB + " " + symbol + "/MB"
 	default:
 		return "—"
 	}