feat: add lightweight workflow run steps

src/onepin/_cli/_spec.py

@@ -114,6 +114,17 @@ def _list_opts(*extra: Opt) -> list[Opt]:
 # Workflow run status filter / terminal states (SDK exposes run status as raw str; no enum).
 _RUN_STATUS = ("draft", "running", "completed", "failed", "paused", "cancelled", "pending")
 
+_NODE_TYPES = (
+    "source_script",
+    "operator_translator",
+    "operator_normalizer",
+    "operator_generator",
+    "sink_preview",
+    "validator_error_rate",
+    "validator_naturalness",
+    "validator_noise",
+)
+
 # Column presets keyed by output model.
 _COLS_WORKFLOW = ["id", "name", "runs_count", "last_run_status", "updated_at"]
 _COLS_VOICE = ["id", "name", "provider", "gender", "category"]
@@ -294,7 +305,12 @@ def _list_opts(*extra: Opt) -> list[Opt]:
         "List the steps of a run.",
         subgroup="runs",
         args=[("workflow_id", "Workflow UUID."), ("run_id", "Run UUID.")],
-        options=[_JSON],
+        options=[
+            Opt("--include-result", "bool", False, help="Include each step's full result payload."),
+            Opt("--node-type", _NODE_TYPES, None, help="Filter by node type."),
+            Opt("--node-id", "str", None, help="Filter by node ID."),
+            _JSON,
+        ],
         unwrap="list",
         columns=[],
         fallback_methods=("workflows.get_run_steps",),

tests/build/test_sdk_contract.py

@@ -14,12 +14,14 @@
 from __future__ import annotations
 
 import inspect
+from typing import get_args
 
 import pytest
 
 from onepin._cli._dispatch import _resolve_method
 from onepin._cli._spec import TABLE, Cmd
 from onepin.client import OnePinClient
+from onepin.types import NodeType
 
 # Options whose dest is consumed by the CLI itself, not forwarded to the SDK.
 _LOCAL_DESTS = {"json_output_local", "reveal", "yes"}
@@ -31,6 +33,16 @@ def _resolve_cmd_method(cmd: Cmd):
     return _resolve_method(client, cmd.method_paths)
 
 
+def test_run_steps_node_type_choices_match_public_type() -> None:
+    cmd = next(cmd for cmd in TABLE if cmd.path == ("workflows", "runs", "steps"))
+    option = next(option for option in cmd.options if option.flag == "--node-type")
+    public_values = tuple(
+        value for branch in get_args(NodeType) for value in get_args(branch) if isinstance(value, str)
+    )
+
+    assert option.type == public_values
+
+
 @pytest.mark.parametrize("cmd", TABLE, ids=lambda c: ".".join(c.path))
 def test_method_resolves(cmd: Cmd) -> None:
     method = _resolve_cmd_method(cmd)

tests/build/test_workflow_run_steps_contract.py

@@ -0,0 +1,120 @@
+"""Generated SDK contract coverage for lightweight workflow run steps."""
+
+from __future__ import annotations
+
+import inspect
+
+import httpx
+import pytest
+import respx
+
+from onepin.client import AsyncOnePinClient, OnePinClient
+from onepin.types import WorkflowRunStepOut
+from onepin.workflows.runs.client import AsyncRunsClient, RunsClient
+from onepin.workflows.runs.raw_client import AsyncRawRunsClient, RawRunsClient
+
+_BASE = "https://api.onepin.ai"
+_STEPS_URL = f"{_BASE}/api/v1/workflows/wf-1/runs/run-1/steps"
+_META = {"request_id": "req-1", "timestamp": "2025-01-01T00:00:00Z"}
+_STEP = {
+    "id": "step-1",
+    "node_id": "node-1",
+    "node_type": "validator_error_rate",
+    "node_display_name": "Error Rate",
+    "status": "completed",
+    "iteration": 0,
+    "started_at": "2025-01-01T00:00:00Z",
+    "completed_at": "2025-01-01T00:00:01Z",
+    "result": None,
+    "has_result": True,
+    "active_ports": ["output"],
+    "error": None,
+}
+_FILTER_CASES = [
+    ({}, {}),
+    ({"include_result": True}, {"include_result": "true"}),
+    ({"node_type": "validator_error_rate"}, {"node_type": "validator_error_rate"}),
+    ({"node_id": "node-1"}, {"node_id": "node-1"}),
+    (
+        {"include_result": True, "node_type": "validator_error_rate", "node_id": "node-1"},
+        {"include_result": "true", "node_type": "validator_error_rate", "node_id": "node-1"},
+    ),
+]
+
+
+@pytest.mark.parametrize("client_type", [RunsClient, AsyncRunsClient, RawRunsClient, AsyncRawRunsClient])
+def test_all_generated_surfaces_expose_optional_step_filters(client_type: type[object]) -> None:
+    signature = inspect.signature(client_type.steps)
+
+    assert tuple(signature.parameters) == (
+        "self",
+        "workflow_id",
+        "run_id",
+        "include_result",
+        "node_type",
+        "node_id",
+        "workspace_id",
+        "request_options",
+    )
+    for name in ("include_result", "node_type", "node_id"):
+        assert signature.parameters[name].default is None
+
+
+def test_generated_step_model_exposes_lightweight_result_metadata() -> None:
+    step = WorkflowRunStepOut.model_validate(_STEP)
+
+    assert step.result is None
+    assert step.has_result is True
+    assert step.active_ports == ["output"]
+
+
+def test_generated_step_model_accepts_older_backend_payload() -> None:
+    legacy_payload = {key: value for key, value in _STEP.items() if key not in {"has_result", "active_ports"}}
+    legacy_payload["result"] = {"output": "legacy full result"}
+
+    step = WorkflowRunStepOut.model_validate(legacy_payload)
+
+    assert step.result == {"output": "legacy full result"}
+    assert step.has_result is None
+    assert step.active_ports is None
+
+
+@pytest.mark.parametrize("raw", [False, True], ids=["standard", "raw"])
+@pytest.mark.parametrize(("kwargs", "expected_query"), _FILTER_CASES)
+@respx.mock
+def test_generated_sync_clients_serialize_step_filters(
+    raw: bool, kwargs: dict[str, object], expected_query: dict[str, str]
+) -> None:
+    route = respx.get(_STEPS_URL).mock(return_value=httpx.Response(200, json={"data": [_STEP], "meta": _META}))
+    http_client = httpx.Client()
+    client = OnePinClient(token="op_live_test", httpx_client=http_client)
+    try:
+        runs = client.workflows.runs.with_raw_response if raw else client.workflows.runs
+        response = runs.steps("wf-1", "run-1", **kwargs)
+    finally:
+        http_client.close()
+
+    data = response.data.data if raw else response.data
+    assert data[0].has_result is True
+    assert dict(route.calls[0].request.url.params) == expected_query
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("raw", [False, True], ids=["standard", "raw"])
+@pytest.mark.parametrize(("kwargs", "expected_query"), _FILTER_CASES)
+@respx.mock
+async def test_generated_async_clients_serialize_step_filters(
+    raw: bool, kwargs: dict[str, object], expected_query: dict[str, str]
+) -> None:
+    route = respx.get(_STEPS_URL).mock(return_value=httpx.Response(200, json={"data": [_STEP], "meta": _META}))
+    http_client = httpx.AsyncClient()
+    client = AsyncOnePinClient(token="op_live_test", httpx_client=http_client)
+    try:
+        runs = client.workflows.runs.with_raw_response if raw else client.workflows.runs
+        response = await runs.steps("wf-1", "run-1", **kwargs)
+    finally:
+        await http_client.aclose()
+
+    data = response.data.data if raw else response.data
+    assert data[0].active_ports == ["output"]
+    assert dict(route.calls[0].request.url.params) == expected_query

tests/cli/_manifest_snapshot.json

@@ -1488,6 +1488,27 @@
       "group": "workflows",
       "name": "steps",
       "options": [
+        {
+          "default": false,
+          "flag": "--include-result",
+          "help": "Include each step's full result payload.",
+          "required": false,
+          "type": "bool"
+        },
+        {
+          "default": null,
+          "flag": "--node-type",
+          "help": "Filter by node type.",
+          "required": false,
+          "type": "choice[source_script,operator_translator,operator_normalizer,operator_generator,sink_preview,validator_error_rate,validator_naturalness,validator_noise]"
+        },
+        {
+          "default": null,
+          "flag": "--node-id",
+          "help": "Filter by node ID.",
+          "required": false,
+          "type": "text"
+        },
         {
           "default": false,
           "flag": "--json",

tests/cli/test_cli_dispatch.py

@@ -74,11 +74,18 @@ def _meta():
 class FakeRuns:
     raise_404 = False
 
+    def __init__(self) -> None:
+        self.step_calls: list[dict[str, object]] = []
+
     def cancel(self, workflow_id, run_id, **kw):
         if self.raise_404:
             raise NotFoundError(body={"error": {"code": "NOT_FOUND", "message": "gone"}})
         return _dict_response(cancelled=True, id=run_id)
 
+    def steps(self, workflow_id, run_id, **kw):
+        self.step_calls.append({"workflow_id": workflow_id, "run_id": run_id, **kw})
+        return [{"id": "step-1", "node_id": "node-1", "status": "completed"}]
+
 
 class FakeWorkflows:
     raise_404 = False
@@ -176,6 +183,71 @@ def test_cancel_404_under_yes_surfaces_error(self, fake_client: FakeClient, tmp_
         assert "cancelled" not in result.output.lower()
 
 
+class TestWorkflowRunSteps:
+    def test_omits_optional_filters_by_default(self, fake_client: FakeClient, tmp_home) -> None:
+        result = _invoke(["workflows", "runs", "steps", "wf-1", "run-1", "--json"])
+
+        assert result.exit_code == 0, result.output
+        assert fake_client.workflows.runs.step_calls == [{"workflow_id": "wf-1", "run_id": "run-1"}]
+        assert '"id": "step-1"' in result.output
+
+    @pytest.mark.parametrize(
+        ("flag", "value", "expected"),
+        [
+            ("--include-result", None, {"include_result": True}),
+            ("--node-type", "operator_generator", {"node_type": "operator_generator"}),
+            ("--node-id", "node-7", {"node_id": "node-7"}),
+        ],
+    )
+    def test_forwards_each_optional_filter(
+        self,
+        fake_client: FakeClient,
+        tmp_home,
+        flag: str,
+        value: str | None,
+        expected: dict[str, object],
+    ) -> None:
+        option = [flag] if value is None else [flag, value]
+        result = _invoke(["workflows", "runs", "steps", "wf-1", "run-1", *option, "--json"])
+
+        assert result.exit_code == 0, result.output
+        assert fake_client.workflows.runs.step_calls == [{"workflow_id": "wf-1", "run_id": "run-1", **expected}]
+
+    def test_forwards_combined_filters(self, fake_client: FakeClient, tmp_home) -> None:
+        result = _invoke(
+            [
+                "workflows",
+                "runs",
+                "steps",
+                "wf-1",
+                "run-1",
+                "--include-result",
+                "--node-type",
+                "validator_error_rate",
+                "--node-id",
+                "node-7",
+                "--json",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+        assert fake_client.workflows.runs.step_calls == [
+            {
+                "workflow_id": "wf-1",
+                "run_id": "run-1",
+                "include_result": True,
+                "node_type": "validator_error_rate",
+                "node_id": "node-7",
+            }
+        ]
+
+    def test_node_type_rejects_unknown_value(self, fake_client: FakeClient, tmp_home) -> None:
+        result = _invoke(["workflows", "runs", "steps", "wf-1", "run-1", "--node-type", "not-a-node", "--json"])
+
+        assert result.exit_code == 2
+        assert fake_client.workflows.runs.step_calls == []
+
+
 class TestNotAuthenticated:
     def test_no_creds_exit_1(self, tmp_home) -> None:
         # No flag, no env, no file -> get_client raises OnePinAuthError.