From c70583a3b9008cd38e90a8071221e0ab7b4fdb6a Mon Sep 17 00:00:00 2001 From: lda Date: Mon, 31 Aug 2026 02:15:03 +0700 Subject: [PATCH] fix: harden workflow client lifecycle identities --- .../python-workflow-client/task-6-report.md | 9 +++ src/wf_client/app.py | 2 + src/wf_client/codec.py | 7 +- src/wf_client/deployments.py | 34 ++++++++- src/wf_client/runs.py | 70 +++++++++++++++---- tests/wf_client/test_deployments.py | 68 ++++++++++++++++-- tests/wf_client/test_runs.py | 45 +++++++++++- 7 files changed, 213 insertions(+), 22 deletions(-) diff --git a/.superpowers/sdd/python-workflow-client/task-6-report.md b/.superpowers/sdd/python-workflow-client/task-6-report.md index 806638cd..64e94999 100644 --- a/.superpowers/sdd/python-workflow-client/task-6-report.md +++ b/.superpowers/sdd/python-workflow-client/task-6-report.md @@ -29,3 +29,12 @@ Implemented and committed as `112e9656` (`feat: add Python deployment and run ob and were intentionally not staged. - Trace limits follow the existing server bound of 1–100 and are validated before issuing a trace request. + +## Review fix round 1 + +Identity checks now reject mismatched deployment/artifact/run ids at every +inspect, validate, start, refresh, and resume boundary. Nested interrupt route +references are converted to `InvalidResponse` with operation context, and run +decoding accepts truthful inspect/start/resume operation names. + +Fixes committed in the follow-up review commit for this report. diff --git a/src/wf_client/app.py b/src/wf_client/app.py index f4ec4ec7..0672f57a 100644 --- a/src/wf_client/app.py +++ b/src/wf_client/app.py @@ -206,4 +206,6 @@ class App: return Run.from_payload( self._port, await self._port.inspect_run(run_id=run_id), + expected_run_id=run_id, + operation="workflow.runs.inspect", ) diff --git a/src/wf_client/codec.py b/src/wf_client/codec.py index 443b44bc..56c43ed5 100644 --- a/src/wf_client/codec.py +++ b/src/wf_client/codec.py @@ -232,9 +232,12 @@ def _decode_run_fields( ) -def decode_run_result(payload: object) -> DecodedRunResult: +def decode_run_result( + payload: object, + *, + operation: str = "workflow.runs.inspect", +) -> DecodedRunResult: """Validate and decode a start/inspect/resume run response.""" - operation = "workflow.runs.inspect" wire = _validate(payload, RunResult, operation) fields = _decode_run_fields( wire, diff --git a/src/wf_client/deployments.py b/src/wf_client/deployments.py index 7444dc70..1383d04c 100644 --- a/src/wf_client/deployments.py +++ b/src/wf_client/deployments.py @@ -80,6 +80,18 @@ class Deployment: f"match requested {self.deployment_id!r}" ), ) + if ( + result["artifact_id"] != self.artifact_id + or result["artifact_version"] != self.artifact_version + ): + raise InvalidResponse( + operation="workflow.deployments.validate", + details=( + f"validation for {self.deployment_id!r} targets artifact " + f"{result['artifact_id']!r} version {result['artifact_version']}, " + f"expected {self.artifact_id!r} version {self.artifact_version}" + ), + ) diagnostics = decode_dependency_diagnostics(result["diagnostics"]) return DeploymentValidation( deployment_id=result["deployment_id"], @@ -97,8 +109,17 @@ class Deployment: await self._port.run_deployment( deployment_id=self.deployment_id, workflow_input=dict(workflow_input), - ) + ), + operation="workflow.runs.start", ) + if decoded.deployment_id != self.deployment_id: + raise InvalidResponse( + operation="workflow.runs.start", + details=( + f"returned deployment {decoded.deployment_id!r} does not " + f"match requested {self.deployment_id!r}" + ), + ) if decoded.run_id is None or decoded.status in {"unrunnable", "rejected"}: raise DeploymentNotRunnable( deployment_id=self.deployment_id, @@ -137,6 +158,17 @@ async def run_artifact( f"match requested {deployment_id!r}" ), ) + if ( + deployment.artifact_id != artifact.artifact.id + or deployment.artifact_version != artifact.artifact.version + ): + raise InvalidResponse( + operation="workflow.deployments.inspect", + details=( + f"deployment {deployment_id!r} does not target artifact " + f"{artifact.artifact.id!r} version {artifact.artifact.version}" + ), + ) return await deployment.run(workflow_input) summaries = _decode_summaries(await artifact._port.list_deployments()) diff --git a/src/wf_client/runs.py b/src/wf_client/runs.py index 8d0bdbfb..f1da7b7b 100644 --- a/src/wf_client/runs.py +++ b/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: diff --git a/tests/wf_client/test_deployments.py b/tests/wf_client/test_deployments.py index 261fb166..c79e7c39 100644 --- a/tests/wf_client/test_deployments.py +++ b/tests/wf_client/test_deployments.py @@ -6,7 +6,7 @@ import pytest from wf_artifacts import WorkflowArtifact as ArtifactModel from wf_client import DeploymentRequired, WorkflowClientPort -from wf_client.errors import DeploymentNotRunnable +from wf_client.errors import DeploymentNotRunnable, InvalidResponse from wf_client.workflows import WorkflowArtifact from wf_core import Workflow @@ -40,6 +40,8 @@ class _FakePort: def __init__(self) -> None: self.calls: list[tuple[str, dict[str, Any]]] = [] self.list_result: dict[str, Any] = {"deployments": []} + self.inspect_artifact_id = "report" + self.inspect_artifact_version = 1 self.validation_result: dict[str, Any] = { "deployment_id": "report.production", "artifact_id": "report", @@ -86,8 +88,8 @@ class _FakePort: self.calls.append(("inspect_deployment", {"deployment_id": deployment_id})) return { "id": deployment_id, - "artifact_id": "report", - "artifact_version": 1, + "artifact_id": self.inspect_artifact_id, + "artifact_version": self.inspect_artifact_version, "bindings": [ { "logical_source": "app.default", @@ -160,5 +162,63 @@ async def test_deployment_run_rejects_missing_run_id() -> None: artifact = _artifact() deployment = await artifact.deploy("report.production") cast(_FakePort, deployment._port).run_result["run_id"] = None - with pytest.raises((DeploymentNotRunnable, AttributeError)): + with pytest.raises(DeploymentNotRunnable) as captured: await deployment.run({}) + assert captured.value.error is None + + +@pytest.mark.asyncio +async def test_explicit_artifact_run_rejects_deployment_for_another_artifact() -> None: + artifact = _artifact() + port = cast(_FakePort, artifact._port) + port.inspect_artifact_id = "other" + with pytest.raises(InvalidResponse, match="does not target artifact"): + await artifact.run({}, deployment_id="report.production") + assert not any(call[0] == "run_deployment" for call in port.calls) + + +@pytest.mark.asyncio +async def test_deployment_validation_rejects_identity_mismatch() -> None: + artifact = _artifact() + deployment = await artifact.deploy("report.production") + port = cast(_FakePort, deployment._port) + port.validation_result["artifact_version"] = 2 + with pytest.raises(InvalidResponse, match="workflow.deployments.validate"): + await deployment.validate() + + +@pytest.mark.asyncio +async def test_deployment_run_rejects_mismatched_start_deployment() -> None: + artifact = _artifact() + deployment = await artifact.deploy("report.production") + port = cast(_FakePort, deployment._port) + port.run_result["deployment_id"] = "other.deployment" + with pytest.raises(InvalidResponse, match="workflow.runs.start"): + await deployment.run({}) + + +@pytest.mark.asyncio +async def test_deployment_run_preserves_server_error_and_diagnostics() -> None: + artifact = _artifact() + deployment = await artifact.deploy("report.production") + port = cast(_FakePort, deployment._port) + port.run_result.update( + run_id=None, + outcome="rejected", + error="dependency check failed", + diagnostics=[ + { + "severity": "error", + "code": "missing_source", + "logical_ref": "app.default", + "bound_source": None, + "message": "missing source", + "repair_hint": "bind a source", + } + ], + ) + with pytest.raises(DeploymentNotRunnable) as captured: + await deployment.run({}) + assert captured.value.error == "dependency check failed" + assert captured.value.outcome == "rejected" + assert captured.value.diagnostics[0].code == "missing_source" diff --git a/tests/wf_client/test_runs.py b/tests/wf_client/test_runs.py index 68e183ff..c9479c96 100644 --- a/tests/wf_client/test_runs.py +++ b/tests/wf_client/test_runs.py @@ -4,7 +4,8 @@ from typing import Any, cast import pytest -from wf_client import Run, WorkflowClientPort +from wf_client import App, Run, WorkflowClientPort +from wf_client.errors import InvalidResponse def _payload( @@ -107,6 +108,48 @@ async def test_refresh_returns_a_new_snapshot() -> None: assert original.status == "interrupted" +@pytest.mark.asyncio +async def test_app_run_rejects_mismatched_inspection_id() -> None: + port = _Port() + port.resume_payload["run_id"] = "different-run" + app = App._from_port(cast(WorkflowClientPort, port)) + with pytest.raises(InvalidResponse, match="workflow.runs.inspect"): + await app.run("run-1") + + +@pytest.mark.asyncio +async def test_refresh_rejects_mismatched_inspection_id() -> None: + port = _Port() + port.resume_payload["run_id"] = "different-run" + run = Run.from_payload(cast(WorkflowClientPort, port), _payload()) + with pytest.raises(InvalidResponse, match="workflow.runs.inspect"): + await run.refresh() + + +@pytest.mark.asyncio +async def test_resume_rejects_mismatched_result_id() -> None: + port = _Port() + port.resume_payload["run_id"] = "different-run" + run = Run.from_payload(cast(WorkflowClientPort, port), _payload()) + with pytest.raises(InvalidResponse, match="workflow.runs.resume"): + await run.resume({"approved": True}) + + +@pytest.mark.asyncio +async def test_malformed_interrupt_route_is_invalid_response() -> None: + payload = _payload() + payload["interrupt"]["route"] = { + "frame_id": "child", + "node_id": "approve", + "scope_id": "scope", + "lineage_id": "lineage", + "parent_frame_id": "root", + "workflow_ref": {"name": "local", "artifact_id": "also-invalid", "version": 1}, + } + with pytest.raises(InvalidResponse, match="workflow.runs.inspect"): + Run.from_payload(cast(WorkflowClientPort, _Port()), payload) + + @pytest.mark.asyncio async def test_non_resumable_run_is_rejected_before_io() -> None: port = _Port()