M56: make RAGcore sync fail closed

scripts/check-contracts.py

@@ -109,6 +109,34 @@ def check_workflows(failures: list[str]) -> None:
             fail(f"{filename} contains a cleartext Fleet Ops callback", failures)
         if filename == "fleet-ops-vehicle-return.json":
             check_return_workflow_timing(definition, failures)
+        if filename == "fleet-ops-ragcore-procedure-sync.json":
+            check_ragcore_sync_workflow(definition, failures)
+
+
+def check_ragcore_sync_workflow(definition: dict, failures: list[str]) -> None:
+    """Keep the sync identity stable and reject HTTP error bodies as successful syncs."""
+    nodes_by_name = {node.get("name"): node for node in definition.get("nodes", [])}
+    upload = nodes_by_name.get("Upload to RAGcore", {})
+    credential = upload.get("credentials", {}).get("httpHeaderAuth", {})
+    if credential.get("name") != "RAGcore Sync Token":
+        fail("RAGcore upload must use the dedicated sync credential", failures)
+
+    parameters = upload.get("parameters", {}).get("bodyParameters", {}).get("parameters", [])
+    field_names = {
+        item.get("name") for item in parameters if isinstance(item, dict)
+    }
+    required = {"space_id", "source_id", "external_id", "locale", "file"}
+    if not required.issubset(field_names):
+        fail(
+            "RAGcore upload must preserve source_id, external_id and locale provenance",
+            failures,
+        )
+
+    summary_code = (
+        nodes_by_name.get("Summarize sync result", {}).get("parameters", {}).get("jsCode", "")
+    )
+    if not all(token in summary_code for token in ("job_id", "status_url", "failed += 1")):
+        fail("RAGcore sync summary must accept only a real AcceptedJob response", failures)
 
 
 def check_return_workflow_timing(definition: dict, failures: list[str]) -> None: