save hash/snapshot of in/output schema

docs/workflow_artifacts.md

@@ -463,6 +463,13 @@ demo.personal.echo_tool + demo -> demo.echo_tool
 That helper should also populate `required_capabilities` so LLM clients do not
 have to reverse-engineer dependency metadata from formatted names.
 
+When artifact creation can observe the concrete `NodeSpecInventory` used during
+authoring, it should persist that node spec's input/output schema snapshots and
+stable hashes into the generated `RequiredCapability`. Later deployment
+validation compares the saved contract against the currently bound concrete
+source and can report `schema_changed` when a source still has the same logical
+capability name but no longer the same contract.
+
 Bindings should live outside the immutable artifact, preferably on a deployment
 or run configuration:
 

src/wf_artifacts/factory.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Mapping
 
 from wf_core import ReducerRef, Workflow
-from wf_platform import CapabilityRef
+from wf_platform import CapabilityRef, NodeSpecInventory
 
 from .models import ArtifactKind, JsonObject, RequiredCapability, WorkflowArtifact
 from .references import normalize_plan_node_refs
@@ -20,12 +20,14 @@ def create_workflow_artifact_from_plan(
     description: str | None = None,
     required_capabilities: Mapping[str, RequiredCapability] | None = None,
     source_bindings: Mapping[str, str] | None = None,
+    observed_node_specs: Mapping[str, NodeSpecInventory] | None = None,
     created_from_catalog_version: str | None = None,
 ) -> WorkflowArtifact:
     """Create an immutable artifact from a declarative workflow plan."""
     normalized_plan, node_requirements = normalize_plan_node_refs(
         plan,
         source_bindings or {},
+        observed_node_specs,
     )
     _validate_workflow_plan(normalized_plan)
     return WorkflowArtifact(

src/wf_artifacts/references.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Mapping
 from copy import deepcopy
 
-from wf_platform import CapabilityRef, SourceRef
+from wf_platform import CapabilityRef, NodeSpecInventory, SourceRef, hash_json_schema
 
 from .models import JsonObject, RequiredCapability
 
@@ -11,6 +11,7 @@ from .models import JsonObject, RequiredCapability
 def normalize_plan_node_refs(
     plan: JsonObject,
     source_bindings: Mapping[str, str],
+    observed_node_specs: Mapping[str, NodeSpecInventory] | None = None,
 ) -> tuple[JsonObject, dict[str, RequiredCapability]]:
     """Rewrite concrete node refs in a plan to deployment-bound logical refs.
 
@@ -39,10 +40,31 @@ def normalize_plan_node_refs(
             continue
         logical_ref, logical_source, capability_name, concrete_source = replacement
         node["node"] = logical_ref
+        observed = (
+            observed_node_specs.get(raw_node_ref)
+            if observed_node_specs is not None
+            else None
+        )
         requirements[logical_ref] = RequiredCapability(
             logical_source=logical_source,
             capability_name=capability_name,
             kind="node_spec",
+            input_schema_hash=(
+                hash_json_schema(observed.input_schema)
+                if observed is not None
+                else None
+            ),
+            input_schema_snapshot=(
+                observed.input_schema if observed is not None else None
+            ),
+            output_schema_hash=(
+                hash_json_schema(observed.output_schema)
+                if observed is not None
+                else None
+            ),
+            output_schema_snapshot=(
+                observed.output_schema if observed is not None else None
+            ),
             observed_concrete_source=concrete_source,
         )
 

src/wf_core/models/workflow.py

@@ -34,5 +34,5 @@ class Workflow(BaseModel):
 
     def validate_structure(self) -> "ValidationReport":
         """Return all structural validation issues for this workflow."""
-        import wf_core.validation.core as validation
+        validation = import_module("wf_core.validation.core")
         return cast("ValidationReport", validation.validate_workflow(self))

src/wf_mcp/workflow_surface/handlers.py

@@ -16,7 +16,7 @@ from wf_artifacts import (
     create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
     validate_deployment_dependencies,
 )
-from wf_platform import CapabilityRef
+from wf_platform import CapabilityRef, NodeSpecInventory, hash_json_schema
 from wf_authoring import build_async_registry
 from wf_core import RuntimeContext
 
@@ -182,6 +182,7 @@ class WorkflowSurfaceHandlers:
                 for name, capability in (required_capabilities or {}).items()
             },
             source_bindings=source_bindings,
+            observed_node_specs=_observed_node_specs(self.service),
             created_from_catalog_version=created_from_catalog_version,
         )
         self.service.artifact_store.save_artifact(workflow_artifact)
@@ -317,15 +318,20 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     """Convert broker capability sources into artifact validation snapshots."""
     sources: list[AvailableSource] = []
     for source in service.capability_sources.values():
+        node_spec_details = {
+            detail.name: detail
+            for detail in source.as_inventory().capabilities.node_spec_details
+        }
         capabilities = {
             capability_name: AvailableCapability(
                 name=capability_name,
                 kind="node_spec",
-                input_schema_hash=None,
+                input_schema_hash=hash_json_schema(detail.input_schema),
-                output_schema_hash=None,
+                output_schema_hash=hash_json_schema(detail.output_schema),
             )
             for spec in source.capabilities.node_specs.values()
             if (capability_name := _capability_name(spec.name)) is not None
+            if (detail := node_spec_details.get(spec.name)) is not None
         }
         capabilities.update(
             {
@@ -347,6 +353,20 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     return sources
 
 
+def _observed_node_specs(service: WfMcpService) -> dict[str, NodeSpecInventory]:
+    """Project current executable specs into serializable observed contracts."""
+    observed: dict[str, NodeSpecInventory] = {}
+    for source in service.capability_sources.values():
+        inventory = source.as_inventory()
+        observed.update(
+            {
+                detail.name: detail
+                for detail in inventory.capabilities.node_spec_details
+            }
+        )
+    return observed
+
+
 def _capability_name(qualified_name: str) -> str | None:
     """Return the local name of one qualified capability ref if it is valid."""
     try:

src/wf_platform/__init__.py

@@ -1,4 +1,5 @@
 from .refs import CapabilityRef, SourceRef
+from .schema_hashes import hash_json_schema
 from .sources import (
     CapabilityBuckets,
     CapabilitySource,
@@ -27,4 +28,5 @@ __all__ = [
     "SourceVisibility",
     "SourceVisibilitySnapshot",
     "SourceRef",
+    "hash_json_schema",
 ]

src/wf_platform/schema_hashes.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any
+
+
+def hash_json_schema(schema: dict[str, Any]) -> str:
+    """Return a stable hash for one JSON-compatible schema document."""
+    payload = json.dumps(
+        schema,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+    ).encode("utf-8")
+    return f"sha256:{hashlib.sha256(payload).hexdigest()}"

tests/artifacts/test_factory.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from typing import Any, cast
 
 from wf_artifacts import RequiredCapability, create_workflow_artifact_from_plan
+from wf_platform import NodeSpecInventory
 
 
 def test_create_workflow_artifact_from_plan_derives_boundary_schemas() -> None:
@@ -77,6 +78,45 @@ def test_create_workflow_artifact_from_plan_rewrites_bound_node_specs() -> None:
     assert required.observed_concrete_source == "demo.personal"
 
 
+def test_create_workflow_artifact_from_plan_snapshots_observed_node_spec() -> None:
+    plan = _plan()
+    _set_first_node_ref(plan, "demo.personal.echo_tool")
+
+    artifact = create_workflow_artifact_from_plan(
+        artifact_id="echo",
+        version=1,
+        title="Echo",
+        plan=plan,
+        outcomes=("done",),
+        source_bindings={"demo": "demo.personal"},
+        observed_node_specs={
+            "demo.personal.echo_tool": NodeSpecInventory(
+                name="demo.personal.echo_tool",
+                outcomes=("ok",),
+                input_schema={"type": "object", "properties": {"text": {"type": "string"}}},
+                output_schema={
+                    "type": "object",
+                    "properties": {"echoed": {"type": "string"}},
+                },
+                is_async=False,
+                accepts_context=False,
+            )
+        },
+    )
+
+    required = artifact.required_capabilities["demo.echo_tool"]
+    assert required.input_schema_snapshot == {
+        "type": "object",
+        "properties": {"text": {"type": "string"}},
+    }
+    assert required.output_schema_snapshot == {
+        "type": "object",
+        "properties": {"echoed": {"type": "string"}},
+    }
+    assert required.input_schema_hash is not None
+    assert required.output_schema_hash is not None
+
+
 def test_create_workflow_artifact_from_plan_keeps_explicit_capability_metadata() -> (
     None
 ):

tests/wf_mcp/test_workflow_surface.py

@@ -34,11 +34,24 @@ class AmountOutput(BaseModel):
     amount: int
 
 
+class ChangedEchoInput(BaseModel):
+    message: str
+
+
+class ChangedEchoOutput(BaseModel):
+    echoed: str
+
+
 @node()
 async def amount_tool(payload: AmountInput) -> AmountOutput:
     return AmountOutput(amount=payload.amount)
 
 
+@node(name="echo_tool")
+def changed_echo_tool(payload: ChangedEchoInput) -> ChangedEchoOutput:
+    return ChangedEchoOutput(echoed=payload.message)
+
+
 @reducer(name="custom.default.multiply")
 def multiply(current: int | None, incoming: int) -> int:
     return (current or 1) * incoming
@@ -281,6 +294,72 @@ def test_workflow_surface_runs_artifact_created_from_concrete_node_ref() -> None
     assert payload["diagnostics"] == []
 
 
+def test_workflow_surface_detects_drift_from_saved_node_spec_snapshot() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_created_drift"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_created_drift_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+
+    asyncio.run(
+        handlers.create_artifact_from_plan(
+            artifact_id="created_echo_drift",
+            version=1,
+            title="Created Echo Drift",
+            plan=_echo_artifact().plan,
+            outcomes=("completed",),
+            source_bindings={"demo": "demo.personal"},
+        )
+    )
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="created_echo_drift.personal",
+            artifact_id="created_echo_drift",
+            artifact_version=1,
+            bindings={
+                "demo": "demo.personal",
+                "wf.std": "wf.std",
+            },
+        )
+    )
+
+    required = artifact_store.get_artifact(
+        "created_echo_drift",
+        1,
+    ).required_capabilities["demo.echo_tool"]
+    assert required.input_schema_hash is not None
+
+    service.register_connection(
+        ConnectionConfig(id="demo.work", server="demo", account="work")
+    )
+    service.register_specs("demo.work", changed_echo_tool)
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="created_echo_drift.work",
+            artifact_id="created_echo_drift",
+            artifact_version=1,
+            bindings={
+                "demo": "demo.work",
+                "wf.std": "wf.std",
+            },
+        )
+    )
+
+    payload = asyncio.run(
+        handlers.validate_deployment(deployment_id="created_echo_drift.work")
+    )
+
+    assert payload["status"] == "unrunnable"
+    assert payload["diagnostics"][0]["code"] == "schema_changed"
+
+
 def test_workflow_surface_runs_deployment_with_bound_reducer_dependency() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_reducer")
     artifact_store.save_artifact(_custom_reducer_artifact())