fix: harden workflow client lifecycle identities

src/wf_client/runs.py

@@ -6,12 +6,14 @@ from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import Any
 
+from pydantic import ValidationError
+
 from wf_api import TraceRange
 from wf_artifacts import DependencyDiagnostic
 from wf_core import InterruptRequest, InterruptRoute, TraceEntry, WorkflowRef
 
 from .codec import DecodedRunResult, decode_run_result, decode_trace_result
-from .errors import DeploymentNotRunnable
+from .errors import DeploymentNotRunnable, InvalidResponse
 from .protocols import WorkflowClientPort
 
 
@@ -26,7 +28,11 @@ class TracePage:
     trace_count: int
 
 
-def _interrupt(payload: Mapping[str, Any] | None) -> InterruptRequest | None:
+def _interrupt(
+    payload: Mapping[str, Any] | None,
+    *,
+    operation: str,
+) -> InterruptRequest | None:
     if payload is None:
         return None
     data = dict(payload)
@@ -35,21 +41,41 @@ def _interrupt(payload: Mapping[str, Any] | None) -> InterruptRequest | None:
     if route_data is not None:
         route_values = dict(route_data)
         workflow_ref = route_values.get("workflow_ref")
-        route = InterruptRoute(
-            frame_id=route_values["frame_id"],
-            node_id=route_values["node_id"],
-            scope_id=route_values["scope_id"],
-            lineage_id=route_values["lineage_id"],
-            parent_frame_id=route_values["parent_frame_id"],
-            workflow_ref=WorkflowRef.model_validate(workflow_ref),
-        )
+        try:
+            route = InterruptRoute(
+                frame_id=route_values["frame_id"],
+                node_id=route_values["node_id"],
+                scope_id=route_values["scope_id"],
+                lineage_id=route_values["lineage_id"],
+                parent_frame_id=route_values["parent_frame_id"],
+                workflow_ref=WorkflowRef.model_validate(workflow_ref),
+            )
+        except (KeyError, TypeError, ValueError, ValidationError) as exc:
+            raise InvalidResponse(
+                operation=operation,
+                details=f"invalid interrupt route: {exc}",
+            ) from exc
     data["route"] = route
     # ``InterruptPayload`` is intentionally consumed at this boundary; public
     # clients receive the core runtime request instead of a wire TypedDict.
     return InterruptRequest(**data)
 
 
-def _run_from_decoded(port: WorkflowClientPort, decoded: DecodedRunResult) -> Run:
+def _run_from_decoded(
+    port: WorkflowClientPort,
+    decoded: DecodedRunResult,
+    *,
+    expected_run_id: str | None = None,
+    operation: str = "workflow.runs.inspect",
+) -> Run:
+    if expected_run_id is not None and decoded.run_id != expected_run_id:
+        raise InvalidResponse(
+            operation=operation,
+            details=(
+                f"returned run {decoded.run_id!r} does not match requested "
+                f"{expected_run_id!r}"
+            ),
+        )
     if decoded.run_id is None:
         raise DeploymentNotRunnable(
             deployment_id=decoded.deployment_id,
@@ -63,7 +89,7 @@ def _run_from_decoded(port: WorkflowClientPort, decoded: DecodedRunResult) -> Ru
         status=decoded.status,
         outcome=decoded.outcome,
         output=decoded.output,
-        interrupt=_interrupt(decoded.interrupt),
+        interrupt=_interrupt(decoded.interrupt, operation=operation),
         diagnostics=decoded.diagnostics,
         trace_count=decoded.trace_count,
     )
@@ -84,15 +110,29 @@ class Run:
     trace_count: int
 
     @classmethod
-    def from_payload(cls, port: WorkflowClientPort, payload: object) -> Run:
+    def from_payload(
+        cls,
+        port: WorkflowClientPort,
+        payload: object,
+        *,
+        expected_run_id: str | None = None,
+        operation: str = "workflow.runs.inspect",
+    ) -> Run:
         """Validate one run response and reconstruct its immutable snapshot."""
-        return _run_from_decoded(port, decode_run_result(payload))
+        return _run_from_decoded(
+            port,
+            decode_run_result(payload, operation=operation),
+            expected_run_id=expected_run_id,
+            operation=operation,
+        )
 
     async def refresh(self) -> Run:
         """Read the current server snapshot without mutating this run."""
         return self.from_payload(
             self._port,
             await self._port.inspect_run(run_id=self.run_id),
+            expected_run_id=self.run_id,
+            operation="workflow.runs.inspect",
         )
 
     async def resume(
@@ -113,6 +153,8 @@ class Run:
                 resume_payload=dict(response),
                 resume_outcome=outcome,
             ),
+            expected_run_id=self.run_id,
+            operation="workflow.runs.resume",
         )
 
     async def trace(self, *, start: int = 0, limit: int = 25) -> TracePage: