feat: skip bindings for platform sources

src/wf_api/deployments.py

@@ -175,6 +175,7 @@ def _available_sources(
             AvailableSource(
                 id=source.id,
                 enabled=source.enabled,
+                platform=source.policy.platform,
                 capabilities=capabilities,
             )
         )

src/wf_artifacts/models.py

@@ -96,6 +96,7 @@ class AvailableSource(BaseModel):
 
     id: str
     enabled: bool = True
+    platform: bool = False
     capabilities: dict[str, AvailableCapability] = Field(default_factory=dict)
 
 

src/wf_artifacts/validation.py

@@ -23,23 +23,30 @@ def validate_deployment_dependencies(
     diagnostics: list[DependencyDiagnostic] = []
 
     for logical_ref, required in artifact.required_capability_map().items():
-        bound_source_id = bindings.get(required.logical_source)
+        platform_source = sources_by_id.get(required.logical_source)
-        if bound_source_id is None:
+        if platform_source is not None and platform_source.platform:
-            diagnostics.append(
+            bound_source_id = required.logical_source
-                _diagnostic(
+        else:
-                    code="binding_missing",
+            bound_source_id = bindings.get(required.logical_source)
-                    logical_ref=logical_ref,
+            if bound_source_id is None:
-                    required=required,
+                if not sources_by_id:
-                    message=(
+                    bound_source_id = required.logical_source
-                        f"No binding exists for logical source "
+                else:
-                        f"{required.logical_source!r}."
+                    diagnostics.append(
-                    ),
+                        _diagnostic(
-                    repair_hint=(
+                            code="binding_missing",
-                        "Bind the logical source to a compatible concrete source."
+                            logical_ref=logical_ref,
-                    ),
+                            required=required,
-                )
+                            message=(
-            )
+                                f"No binding exists for logical source "
-            continue
+                                f"{required.logical_source!r}."
+                            ),
+                            repair_hint=(
+                                "Bind the logical source to a compatible concrete source."
+                            ),
+                        )
+                    )
+                    continue
 
         source = sources_by_id.get(bound_source_id)
         if source is None:

tests/artifacts/test_validation.py

@@ -204,3 +204,52 @@ def test_validate_deployment_accepts_reducer_capability() -> None:
     )
 
     assert diagnostics == []
+
+
+def test_platform_source_requirement_does_not_need_binding() -> None:
+    artifact = artifact_with(
+        required_capability(logical_source="wf.std", capability_name="replace")
+    )
+    deployment = WorkflowDeployment(
+        id="demo.default",
+        artifact_id=artifact.id,
+        artifact_version=artifact.version,
+        bindings={},
+    )
+
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact,
+        deployment=deployment,
+        sources=[
+            AvailableSource(
+                id="wf.std",
+                platform=True,
+                capabilities={
+                    "replace": AvailableCapability(name="replace", kind="node_spec")
+                },
+            )
+        ],
+    )
+
+    assert diagnostics == []
+
+
+def test_missing_platform_source_still_reports_source_missing() -> None:
+    artifact = artifact_with(
+        required_capability(logical_source="wf.std", capability_name="replace")
+    )
+    deployment = WorkflowDeployment(
+        id="demo.default",
+        artifact_id=artifact.id,
+        artifact_version=artifact.version,
+        bindings={},
+    )
+
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact,
+        deployment=deployment,
+        sources=[],
+    )
+
+    assert [diagnostic.code for diagnostic in diagnostics] == ["source_missing"]
+    assert diagnostics[0].bound_source == "wf.std"