The Last Cleanup, and thats most old duplicated code removed

src/wf_api/capabilities.py

@@ -88,7 +88,7 @@ class WorkflowCapabilityApi:
                 ),
             }
             for source in sorted(
-                self.context.capability_sources.values(),
+                self.context.specs.capability_sources.values(),
                 key=lambda source: source.id,
             )
             if source.enabled and source.visibility.planner
@@ -113,7 +113,7 @@ class WorkflowCapabilityApi:
 
     async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
         """Return one planner-visible workflow capability contract."""
-        for source in self.context.capability_sources.values():
+        for source in self.context.specs.capability_sources.values():
             if not source.enabled or not source.visibility.planner:
                 continue
             for detail in source.as_inventory().capabilities.node_spec_details:
@@ -150,7 +150,7 @@ class WorkflowCapabilityApi:
         spec = self.context.specs.get_qualified_spec(qualified_name)
         handler = build_async_registry(spec)[spec.name]
         source_id = _source_id_for_capability(
-            self.context.capability_sources,
+            self.context.specs.capability_sources,
             spec.name,
         )
         try:

src/wf_api/capability_requirements.py

@@ -27,7 +27,7 @@ def observed_node_specs(
 ) -> dict[str, NodeSpecInventory]:
     """Project current executable specs into serializable observed contracts."""
     observed: dict[str, NodeSpecInventory] = {}
-    for source in context.capability_sources.values():
+    for source in context.specs.capability_sources.values():
         inventory = source.as_inventory()
         observed.update(
             {detail.name: detail for detail in inventory.capabilities.node_spec_details}

src/wf_api/deployments.py

@@ -120,7 +120,7 @@ class WorkflowDeploymentApi:
             deployment.artifact_id,
             deployment.artifact_version,
         )
-        available_sources = _available_sources(self.context.capability_sources)
+        available_sources = _available_sources(self.context.specs.capability_sources)
         diagnostics = validate_deployment_dependencies(
             artifact=artifact,
             deployment=deployment,

src/wf_api/operation_context.py

@@ -105,7 +105,6 @@ class WorkflowOperationContext:
     artifact_store: WorkflowArtifactStore | None
     draft_workspace_store: DraftWorkspaceStore | None
     run_store: RunStore | None
-    capability_sources: Mapping[str, CapabilitySource]
     events: WorkflowEventRecorder
     specs: WorkflowSpecProvider
     runtime: WorkflowRuntimeRunner

src/wf_api/runs.py

@@ -116,7 +116,7 @@ class WorkflowRunApi:
         environment = record.environment
         diagnostics = validate_pinned_resume_environment(
             record=record,
-            sources=_available_sources(self.context.capability_sources),
+            sources=_available_sources(self.context.specs.capability_sources),
         )
         if has_blocking_diagnostics(diagnostics):
             blocked = mark_resume_blocked(