feat(import): add deterministic OpenAPI normalized IR

apps/admin-api/src/service/imports.rs

@@ -6,7 +6,8 @@ use crank_core::{
     WorkspaceId,
 };
 use crank_import::rest::{
-    ImportFinding, ImportFindingSeverity, ImportOperationCandidate, operation_draft_from_candidate,
+    ImportFinding, ImportFindingSeverity, ImportOperationCandidate, NORMALIZER_VERSION,
+    PROJECTION_VERSION, operation_draft_from_candidate,
 };
 use crank_registry::{
     ApplyImportJobRequest, ArtifactSourceId, ArtifactSourceSensitivity,
@@ -94,14 +95,22 @@ impl AdminService {
             detach_guard.detach_now().await;
             return Err(ApiError::openapi_upload(locale, "source_integrity"));
         }
-        let preview = match parse_verified_preview(verified.bytes, locale).await {
+        let parsed = match parse_verified_preview(
+            verified.bytes,
+            artifact.artifact_ref().digest_hex().to_owned(),
+            locale,
+            false,
+        )
+        .await
+        {
             Ok(preview) => preview,
             Err(error) => {
                 detach_guard.detach_now().await;
                 return Err(error);
             }
         };
-        if preview
+        if parsed
+            .preview
             .groups
             .iter()
             .all(|group| group.operations.is_empty())
@@ -113,7 +122,7 @@ impl AdminService {
             source_id,
             digest: artifact.artifact_ref().clone(),
         };
-        let preview_value = serde_json::to_value(&preview)
+        let preview_value = serde_json::to_value(&parsed.preview)
             .map_err(|error| ApiError::internal(error.to_string()))?;
         let preview_digest = preview_digest(&preview_value)?;
         let preview_payload = json!({
@@ -123,6 +132,11 @@ impl AdminService {
             },
             "preview": preview_value,
             "preview_digest": preview_digest,
+            "normalization": {
+                "normalizer_version": NORMALIZER_VERSION,
+                "projection_version": PROJECTION_VERSION,
+                "ir_fingerprint": parsed.ir_fingerprint,
+            },
         });
 
         if let Err(error) = self
@@ -131,8 +145,8 @@ impl AdminService {
                 id: &job_id,
                 workspace_id,
                 kind: ImportJobKind::OpenApi,
-                source_format: &preview.source.format,
-                source_version: preview.source.version.as_deref(),
+                source_format: &parsed.preview.source.format,
+                source_version: parsed.preview.source.version.as_deref(),
                 status: ImportJobStatus::Pending,
                 source: &source_envelope,
                 preview_payload: &preview_payload,
@@ -151,7 +165,7 @@ impl AdminService {
             expires_at: expires_at
                 .format(&Rfc3339)
                 .map_err(|error| ApiError::internal(error.to_string()))?,
-            preview,
+            preview: parsed.preview,
         })
     }
 
@@ -277,10 +291,20 @@ impl AdminService {
         if verified.source.blob.artifact_ref != source.digest {
             return Err(ApiError::openapi_upload(locale, "source_integrity"));
         }
-        let preview = parse_verified_preview(verified.bytes, locale).await?;
-        verify_preview_contract(&job.preview_payload, &preview, locale)?;
+        let legacy_v1 = job.preview_payload.get("normalization").is_none();
+        let parsed = parse_verified_preview(
+            verified.bytes,
+            source.digest.digest_hex().to_owned(),
+            locale,
+            legacy_v1,
+        )
+        .await?;
+        verify_preview_contract(&job.preview_payload, &parsed, locale)?;
+        if preview_has_blocker(&parsed.preview) {
+            return Err(ApiError::openapi_upload(locale, "invalid_document"));
+        }
         let mut candidates = BTreeMap::new();
-        for group in &preview.groups {
+        for group in &parsed.preview.groups {
             for operation in &group.operations {
                 candidates.insert(operation.key.clone(), operation);
             }
@@ -441,10 +465,17 @@ fn validate_openapi_upload(upload: &OpenApiUpload) -> Result<(), ApiError> {
     Ok(())
 }
 
+struct ParsedOpenApiPreview {
+    preview: crank_import::rest::ImportPreview,
+    ir_fingerprint: String,
+}
+
 async fn parse_verified_preview(
     bytes: Vec<u8>,
+    digest: String,
     locale: OpenApiUploadLocale,
-) -> Result<crank_import::rest::ImportPreview, ApiError> {
+    legacy_v1: bool,
+) -> Result<ParsedOpenApiPreview, ApiError> {
     let started = tokio::time::Instant::now();
     let permit = tokio::time::timeout(OPENAPI_PARSE_DEADLINE, OPENAPI_PARSE_SLOTS.acquire())
         .await
@@ -457,8 +488,32 @@ async fn parse_verified_preview(
         let _permit = permit;
         let document = std::str::from_utf8(&bytes)
             .map_err(|_| ApiError::openapi_upload(locale, "invalid_utf8"))?;
-        crank_import::rest::preview_document(document)
-            .map_err(|_| ApiError::openapi_upload(locale, "invalid_document"))
+        if legacy_v1 {
+            return crank_import::rest::preview_document_legacy_v1(document)
+                .map(|preview| ParsedOpenApiPreview {
+                    preview,
+                    ir_fingerprint: String::new(),
+                })
+                .map_err(|_| ApiError::openapi_upload(locale, "invalid_document"));
+        }
+        let digest = crank_import::rest::SourceDigest::parse(digest)
+            .map_err(|_| ApiError::openapi_upload(locale, "source_integrity"))?;
+        let ir =
+            crank_import::rest::normalize_verified_document(document, digest, &Default::default())
+                .map_err(|error| match error {
+                    crank_import::rest::ImportParseError::NoMethods => {
+                        ApiError::openapi_upload(locale, "no_methods")
+                    }
+                    _ => ApiError::openapi_upload(locale, "invalid_document"),
+                })?;
+        let ir_fingerprint = preview_digest(
+            &serde_json::to_value(&ir)
+                .map_err(|_| ApiError::openapi_upload(locale, "invalid_document"))?,
+        )?;
+        Ok::<_, ApiError>(ParsedOpenApiPreview {
+            preview: crank_import::rest::preview_from_ir(&ir),
+            ir_fingerprint,
+        })
     });
     let remaining = OPENAPI_PARSE_DEADLINE
         .checked_sub(started.elapsed())
@@ -499,28 +554,63 @@ fn preview_digest(preview: &serde_json::Value) -> Result<String, ApiError> {
 
 fn verify_preview_contract(
     payload: &serde_json::Value,
-    preview: &crank_import::rest::ImportPreview,
+    parsed: &ParsedOpenApiPreview,
     locale: OpenApiUploadLocale,
 ) -> Result<(), ApiError> {
     // Pre-fingerprint jobs are legacy rolling-upgrade records. They retain the
     // old reparse behavior; new jobs fail closed if parser output drifts or a
     // persisted preview has been changed.
-    let Some(expected) = payload
+    let expected = payload
         .get("preview_digest")
-        .and_then(serde_json::Value::as_str)
-    else {
+        .and_then(serde_json::Value::as_str);
+    if payload.get("normalization").is_some() && expected.is_none() {
+        return Err(ApiError::openapi_upload(locale, "source_integrity"));
+    }
+    let Some(expected) = expected else {
         return Ok(());
     };
     let actual = preview_digest(
-        &serde_json::to_value(preview).map_err(|error| ApiError::internal(error.to_string()))?,
+        &serde_json::to_value(&parsed.preview)
+            .map_err(|error| ApiError::internal(error.to_string()))?,
     )?;
     if actual == expected {
+        if let Some(normalization) = payload.get("normalization") {
+            let normalizer = normalization
+                .get("normalizer_version")
+                .and_then(serde_json::Value::as_str);
+            let projection = normalization
+                .get("projection_version")
+                .and_then(serde_json::Value::as_str);
+            let fingerprint = normalization
+                .get("ir_fingerprint")
+                .and_then(serde_json::Value::as_str);
+            if normalizer != Some(NORMALIZER_VERSION)
+                || projection != Some(PROJECTION_VERSION)
+                || fingerprint != Some(parsed.ir_fingerprint.as_str())
+            {
+                return Err(ApiError::openapi_upload(locale, "source_integrity"));
+            }
+        }
         Ok(())
     } else {
         Err(ApiError::openapi_upload(locale, "source_integrity"))
     }
 }
 
+fn preview_has_blocker(preview: &crank_import::rest::ImportPreview) -> bool {
+    preview
+        .findings
+        .iter()
+        .chain(
+            preview
+                .groups
+                .iter()
+                .flat_map(|group| group.operations.iter())
+                .flat_map(|operation| operation.findings.iter()),
+        )
+        .any(|finding| finding.severity == ImportFindingSeverity::Error)
+}
+
 fn import_job_source(
     payload: &serde_json::Value,
     locale: OpenApiUploadLocale,

apps/admin-api/tests/integration/openapi_import.rs

@@ -60,12 +60,26 @@ async fn previews_openapi_and_creates_draft_operations() {
         preview.preview.groups[0].operations[0].suggested_name,
         "latest_rates"
     );
+    let public_response = serde_json::to_string(&preview).unwrap();
+    assert!(!public_response.contains("normalizer_version"));
+    assert!(!public_response.contains("projection_version"));
+    assert!(!public_response.contains("ir_fingerprint"));
+    assert!(!public_response.contains("source_identity"));
     let preview_job = registry
         .get_import_job(&workspace_id, &preview.job_id.as_str().into())
         .await
         .unwrap()
         .unwrap();
     assert_eq!(preview_job.status, ImportJobStatus::Pending);
+    assert_eq!(
+        preview_job.preview_payload["normalization"]["normalizer_version"],
+        "normalized-ir-v2"
+    );
+    assert_eq!(
+        preview_job.preview_payload["normalization"]["projection_version"],
+        "preview-v2"
+    );
+    assert!(preview_job.preview_payload["normalization"]["ir_fingerprint"].is_string());
 
     let created = service
         .create_openapi_import(
@@ -321,6 +335,251 @@ async fn apply_fails_closed_when_the_preview_parser_contract_drifts() {
     );
 }
 
+#[tokio::test]
+#[serial]
+async fn apply_fails_closed_for_each_normalization_contract_field_and_digest_shape() {
+    let registry = test_registry().await;
+    let service = test_service(
+        registry.clone(),
+        test_storage_root("openapi_import_contract_fields"),
+        test_auth_settings(),
+        test_secret_crypto(),
+    );
+    let workspace_id = WorkspaceId::new("ws_default");
+    for case in [
+        "normalizer_version",
+        "projection_version",
+        "ir_fingerprint",
+        "missing_preview_digest",
+        "non_string_preview_digest",
+    ] {
+        let preview = service
+            .preview_openapi_import(&workspace_id, openapi_upload())
+            .await
+            .unwrap();
+        let job_id: crank_registry::ImportJobId = preview.job_id.as_str().into();
+        let query = match case {
+            "normalizer_version" => sqlx::query(
+                "update import_jobs set preview_payload = jsonb_set(preview_payload, '{normalization,normalizer_version}', to_jsonb('future'::text)) where id = $1",
+            ),
+            "projection_version" => sqlx::query(
+                "update import_jobs set preview_payload = jsonb_set(preview_payload, '{normalization,projection_version}', to_jsonb('future'::text)) where id = $1",
+            ),
+            "ir_fingerprint" => sqlx::query(
+                "update import_jobs set preview_payload = jsonb_set(preview_payload, '{normalization,ir_fingerprint}', to_jsonb('bad'::text)) where id = $1",
+            ),
+            "missing_preview_digest" => sqlx::query(
+                "update import_jobs set preview_payload = preview_payload - 'preview_digest' where id = $1",
+            ),
+            "non_string_preview_digest" => sqlx::query(
+                "update import_jobs set preview_payload = jsonb_set(preview_payload, '{preview_digest}', '42'::jsonb) where id = $1",
+            ),
+            _ => unreachable!(),
+        };
+        query
+            .bind(job_id.as_str())
+            .execute(registry.pool())
+            .await
+            .unwrap();
+        assert!(
+            service
+                .create_openapi_import(
+                    &workspace_id,
+                    &job_id,
+                    OpenApiImportCreatePayload {
+                        selected_operation_keys: vec!["GET /v2/latest".to_owned()],
+                        server_url: Some("https://api.frankfurter.dev".to_owned()),
+                        conflict_mode: "rename".to_owned(),
+                    }
+                )
+                .await
+                .is_err()
+        );
+        assert!(
+            service
+                .list_operations(&workspace_id)
+                .await
+                .unwrap()
+                .is_empty()
+        );
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn legacy_job_without_normalization_contract_replays_until_expiry() {
+    let registry = test_registry().await;
+    let service = test_service(
+        registry.clone(),
+        test_storage_root("openapi_import_legacy_replay"),
+        test_auth_settings(),
+        test_secret_crypto(),
+    );
+    let workspace_id = WorkspaceId::new("ws_default");
+    let preview = service
+        .preview_openapi_import(&workspace_id, openapi_upload())
+        .await
+        .unwrap();
+    let job_id: crank_registry::ImportJobId = preview.job_id.as_str().into();
+    sqlx::query(
+        "update import_jobs set preview_payload = preview_payload - 'normalization' where id = $1",
+    )
+    .bind(job_id.as_str())
+    .execute(registry.pool())
+    .await
+    .unwrap();
+
+    let result = service
+        .create_openapi_import(
+            &workspace_id,
+            &job_id,
+            OpenApiImportCreatePayload {
+                selected_operation_keys: vec!["GET /v2/latest".to_owned()],
+                server_url: Some("https://api.frankfurter.dev".to_owned()),
+                conflict_mode: "skip".to_owned(),
+            },
+        )
+        .await
+        .unwrap();
+    assert_eq!(result.created.len(), 1);
+}
+
+#[tokio::test]
+#[serial]
+async fn blocker_finding_keeps_valid_preview_but_prevents_draft_mutation() {
+    let registry = test_registry().await;
+    let service = test_service(
+        registry,
+        test_storage_root("openapi_import_blocker"),
+        test_auth_settings(),
+        test_secret_crypto(),
+    );
+    let workspace_id = WorkspaceId::new("ws_default");
+    let upload = OpenApiUpload {
+        bytes: br#"
+openapi: 3.0.3
+info: { title: Partial }
+paths:
+  /valid:
+    get:
+      responses: { '200': { description: ok } }
+  /broken: not-a-path-item
+"#
+        .to_vec(),
+        mime_type: "application/yaml".to_owned(),
+        locale: OpenApiUploadLocale::En,
+    };
+    let preview = service
+        .preview_openapi_import(&workspace_id, upload)
+        .await
+        .unwrap();
+    assert_eq!(preview.preview.groups[0].operations.len(), 1);
+    assert!(
+        preview
+            .preview
+            .findings
+            .iter()
+            .any(|finding| finding.severity == crank_import::rest::ImportFindingSeverity::Error)
+    );
+
+    assert!(
+        service
+            .create_openapi_import(
+                &workspace_id,
+                &preview.job_id.as_str().into(),
+                OpenApiImportCreatePayload {
+                    selected_operation_keys: vec!["GET /valid".to_owned()],
+                    server_url: None,
+                    conflict_mode: "skip".to_owned(),
+                },
+            )
+            .await
+            .is_err()
+    );
+    assert!(
+        service
+            .list_operations(&workspace_id)
+            .await
+            .unwrap()
+            .is_empty()
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn operation_level_reference_blocker_prevents_draft_mutation() {
+    let registry = test_registry().await;
+    let service = test_service(
+        registry,
+        test_storage_root("openapi_import_operation_blocker"),
+        test_auth_settings(),
+        test_secret_crypto(),
+    );
+    let workspace_id = WorkspaceId::new("ws_default");
+    let upload = OpenApiUpload {
+        bytes: br#"
+openapi: 3.0.3
+info: { title: References }
+paths:
+  /referenced:
+    get:
+      operationId: referenced
+      responses:
+        '200':
+          description: ok
+          content:
+            application/json:
+              schema: { $ref: '#/components/schemas/Result' }
+components:
+  schemas:
+    Result: { type: object, properties: { id: { type: string } } }
+"#
+        .to_vec(),
+        mime_type: "application/yaml".to_owned(),
+        locale: OpenApiUploadLocale::En,
+    };
+    let preview = service
+        .preview_openapi_import(&workspace_id, upload)
+        .await
+        .unwrap();
+    assert!(
+        preview.preview.findings.iter().all(|finding| {
+            finding.severity != crank_import::rest::ImportFindingSeverity::Error
+        })
+    );
+    assert!(
+        preview.preview.groups[0].operations[0]
+            .findings
+            .iter()
+            .any(|finding| {
+                finding.code == "unresolved_reference"
+                    && finding.severity == crank_import::rest::ImportFindingSeverity::Error
+            })
+    );
+
+    assert!(
+        service
+            .create_openapi_import(
+                &workspace_id,
+                &preview.job_id.as_str().into(),
+                OpenApiImportCreatePayload {
+                    selected_operation_keys: vec!["GET /referenced".to_owned()],
+                    server_url: None,
+                    conflict_mode: "skip".to_owned(),
+                },
+            )
+            .await
+            .is_err()
+    );
+    assert!(
+        service
+            .list_operations(&workspace_id)
+            .await
+            .unwrap()
+            .is_empty()
+    );
+}
+
 fn openapi_upload() -> OpenApiUpload {
     OpenApiUpload {
         bytes: OPENAPI3.as_bytes().to_vec(),