diff --git a/CHANGELOG.md b/CHANGELOG.md index 0216060..911923e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Unreleased typed run-activity visibility through `Client.get_run_activities` and `WorkflowHandle.get_activities`, with activity/attempt metadata, exact run/namespace binding, explicit unavailable fields and preserved diagnostic authority. Standalone activity listing is unchanged. Shared delivery remains tracked in durable-workflow/.github#235. + - Unreleased `ActivityInfo.activity_execution_id` exposes the received logical activity identity without task/attempt/session fallbacks. Prepared local callbacks receive their validated admission ID; inline local callbacks and older claims without an ID expose `None`. These IDs are public idempotency metadata, not live-claim or external-effect authority, and replace no credentials, heartbeat/current-attempt checks, per-effect suffixes or outcome reconciliation. Shared identity acceptance remains tracked in durable-workflow/.github#211. - Lazy `Client.iter_workflows`, `Client.iter_history` and `WorkflowHandle.iter_history` follow opaque page tokens while retaining only the current page and bounded cursor bookkeeping. They preserve run/filter/namespace identity, follow empty continued pages, stop on early close, and surface HTTP, cancellation and invalid/cyclic cursor errors. Existing one-page methods keep their contracts. (durable-workflow/.github#247) diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index ab053d0..15e35e6 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -131,6 +131,48 @@ For a fuller deployable example, see [`examples/order_processing`](https://github.com/durable-workflow/sdk-python/tree/main/examples/order_processing), which runs a multi-activity order workflow against a local server with Docker Compose. +## Inspect a run's activities (unreleased) + +Read a run's activities alongside its description, without parsing raw history: + +```python +handle = client.get_workflow_handle(workflow_id, run_id=run_id) +description = await handle.describe_run() +activities = await handle.get_activities() +print(description.status) + +for item in activities.activities: + print(item.id, item.activity_type, item.status, item.history_authority) + if item.attempts is None: + print("attempt metadata unavailable") + continue + for attempt in item.attempts: + print(attempt.attempt_number, attempt.status, attempt.last_heartbeat_at) +``` + +`Client.get_run_activities(workflow_id, run_id)` returns the same +`WorkflowActivityList`. A handle needs a stored run ID or an explicit +`get_activities(run_id)` argument. The request binds to that exact workflow/run +and the Client's namespace. `list_activities()` remains the separate standalone +job API. + +`WorkflowActivity` and `WorkflowActivityAttempt` preserve status, attempt, +heartbeat, retry, local/parallel and history-authority metadata. The response's +`type` and `class` fields are exposed as `activity_type` and `activity_class`. +`raw` retains the complete received activity or attempt, including future +fields. Missing optional metadata is `None`; it does not mean zero, healthy or +no cancellation. An explicitly empty list is distinct from missing metadata. +Unknown status strings and Server's `diagnostic_only`, `history_authority` and +`history_unsupported_reason` values remain visible. These visibility fields do +not grant permission to continue an activity attempt. + +Malformed collections or known field types and a response for another run +raise rather than manufacturing an empty view. HTTP errors propagate normally. +The wrapper performs no payload fetch or decoding. Published-package and actual +Server acceptance remain tracked in +[the shared run-activity issue](https://github.com/durable-workflow/.github/issues/235). +This source addition is outside the current typed-value release candidate. + ## Schedule visibility and paging `list_schedules()` returns one typed `ScheduleList` page. Status and workflow diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 3515e77..e2d9ed8 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -52,6 +52,9 @@ TaskQueueList, TaskQueueQueryAdmission, TaskQueueTaskAdmission, + WorkflowActivity, + WorkflowActivityAttempt, + WorkflowActivityList, WorkflowCommandResult, WorkflowExecution, WorkflowHandle, @@ -331,6 +334,9 @@ "WorkflowEntryArgumentsMismatchError", "WorkflowRun", "WorkflowRunList", + "WorkflowActivity", + "WorkflowActivityAttempt", + "WorkflowActivityList", "WorkflowStreamAppendItem", "WorkflowStreamAppendResult", "WorkflowStreamDescription", diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index cf009d1..2f38d91 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -726,6 +726,133 @@ class WorkflowRunList: runs: list[WorkflowRun] +@dataclass +class WorkflowActivityAttempt: + """Received attempt visibility. Missing metadata stays unavailable.""" + + id: str | None = None + worker_attempt_id: str | None = None + attempt_number: int | None = None + status: str | None = None + task_id: str | None = None + lease_owner: str | None = None + lease_expires_at: str | None = None + started_at: str | None = None + last_heartbeat_at: str | None = None + last_heartbeat_progress: dict[str, Any] | list[Any] | None = None + closed_at: str | None = None + can_continue: bool | None = None + cancel_requested: bool | None = None + stop_reason: str | None = None + failure_reason: str | None = None + raw: dict[str, Any] | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> WorkflowActivityAttempt: + values = {name: _run_activity_field(data, name, str) for name in ( + "id", "worker_attempt_id", "status", "task_id", "lease_owner", "lease_expires_at", "started_at", + "last_heartbeat_at", "closed_at", "stop_reason", "failure_reason", + )} + return cls(**values, attempt_number=_run_activity_field(data, "attempt_number", int), + can_continue=_run_activity_field(data, "can_continue", bool), + cancel_requested=_run_activity_field(data, "cancel_requested", bool), + last_heartbeat_progress=_run_activity_progress(data), raw=copy.deepcopy(data)) + + +@dataclass +class WorkflowActivity: + """Activity metadata from one run, retaining Server diagnostic authority.""" + + id: str | None = None + idempotency_key: str | None = None + sequence: int | None = None + activity_type: str | None = None + activity_class: str | None = None + execution_mode: str | None = None + local_activity: bool | None = None + history_authority: str | None = None + diagnostic_only: bool | None = None + history_event_types: list[str] | None = None + history_unsupported_reason: str | None = None + reused_from_run_id: str | None = None + reused_activity_execution_id: str | None = None + reused_recorded_at: str | None = None + row_status: str | None = None + parallel_group_kind: str | None = None + parallel_group_id: str | None = None + parallel_group_base_sequence: int | None = None + parallel_group_size: int | None = None + parallel_group_index: int | None = None + parallel_group_path: list[dict[str, Any]] | None = None + attempt_id: str | None = None + worker_attempt_id: str | None = None + status: str | None = None + attempt_count: int | None = None + retry_policy: Any = None + connection: str | None = None + queue: str | None = None + last_heartbeat_at: str | None = None + last_heartbeat_progress: dict[str, Any] | list[Any] | None = None + created_at: str | None = None + started_at: str | None = None + closed_at: str | None = None + attempts: list[WorkflowActivityAttempt] | None = None + raw: dict[str, Any] | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> WorkflowActivity: + values = {name: _run_activity_field(data, name, str) for name in ( + "id", "idempotency_key", "execution_mode", "history_authority", "history_unsupported_reason", + "reused_from_run_id", "reused_activity_execution_id", "reused_recorded_at", "row_status", + "parallel_group_kind", "parallel_group_id", "attempt_id", "worker_attempt_id", "status", + "connection", "queue", "last_heartbeat_at", "created_at", "started_at", "closed_at", + )} + for name in ("sequence", "parallel_group_base_sequence", "parallel_group_size", "parallel_group_index", + "attempt_count"): + values[name] = _run_activity_field(data, name, int) + for name in ("local_activity", "diagnostic_only"): + values[name] = _run_activity_field(data, name, bool) + attempts = _run_activity_items(data, "attempts", dict) + return cls(**values, activity_type=_run_activity_field(data, "type", str), + activity_class=_run_activity_field(data, "class", str), + history_event_types=_run_activity_items(data, "history_event_types", str), + parallel_group_path=_run_activity_items(data, "parallel_group_path", dict), + attempts=[WorkflowActivityAttempt.from_dict(item) for item in attempts] + if attempts is not None else None, + last_heartbeat_progress=_run_activity_progress(data), + retry_policy=copy.deepcopy(data.get("retry_policy")), raw=copy.deepcopy(data)) + + +@dataclass +class WorkflowActivityList: + """One explicit run's activity visibility, distinct from standalone jobs.""" + + workflow_id: str + run_id: str + activities: list[WorkflowActivity] + + +def _run_activity_field(data: dict[str, Any], name: str, kind: type[Any]) -> Any: + value = data.get(name) + if value is not None and type(value) is not kind: + raise RuntimeError(f"invalid run activity field: {name}") + return value + + +def _run_activity_items(data: dict[str, Any], name: str, kind: type[Any]) -> Any: + items = _run_activity_field(data, name, list) + if items is not None and any(type(item) is not kind for item in items): + raise RuntimeError(f"invalid run activity items: {name}") + return items + + +def _run_activity_progress(data: dict[str, Any]) -> Any: + value = data.get("last_heartbeat_progress") + if value is not None and not isinstance(value, (dict, list)): + raise RuntimeError("invalid run activity field: last_heartbeat_progress") + return copy.deepcopy(value) + + @dataclass class StandaloneActivityExecution: """Server view of one standalone activity execution. @@ -1425,6 +1552,13 @@ async def describe_run(self, run_id: str | None = None) -> WorkflowRun: raise ValueError("run_id is required to describe a workflow run from a handle") return await self._client.describe_workflow_run(self.workflow_id, selected_run_id) + async def get_activities(self, run_id: str | None = None) -> WorkflowActivityList: + """Read activities for an explicit run. See :meth:`Client.get_run_activities`.""" + selected_run_id = self.run_id if run_id is None else run_id + if selected_run_id is None: + raise ValueError("run_id is required to fetch workflow activities from a handle") + return await self._client.get_run_activities(self.workflow_id, selected_run_id) + async def signal( self, signal_name: str, @@ -4016,6 +4150,25 @@ async def list_workflow_runs(self, workflow_id: str) -> WorkflowRunList: runs=runs, ) + async def get_run_activities(self, workflow_id: str, run_id: str) -> WorkflowActivityList: + """Read typed activity/attempt metadata for this namespace and exact run. + + Missing optional metadata stays None. Diagnostic/unsupported states and + unknown future status strings are preserved; this visibility snapshot + does not grant attempt authority. Payloads are not fetched or decoded. + """ + data = await self._request( + "GET", f"/workflows/{quote(workflow_id, safe='._:-')}/runs/{quote(run_id, safe='._:-')}/activities", + context=workflow_id, + ) + if (not isinstance(data, dict) or data.get("workflow_id") != workflow_id + or data.get("run_id") != run_id): + raise RuntimeError("run activity response changed workflow/run identity") + items = _run_activity_items(data, "activities", dict) + if items is None: + raise RuntimeError("run activity response omitted activities") + return WorkflowActivityList(workflow_id, run_id, [WorkflowActivity.from_dict(item) for item in items]) + async def describe_workflow_run(self, workflow_id: str, run_id: str) -> WorkflowRun: """Return detailed status, payload, and actionability for one specific workflow run.""" data = await self._request("GET", f"/workflows/{workflow_id}/runs/{run_id}", context=workflow_id) diff --git a/tests/test_run_activities.py b/tests/test_run_activities.py new file mode 100644 index 0000000..8fca134 --- /dev/null +++ b/tests/test_run_activities.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +from typing import Any + +import httpx +import pytest + +import durable_workflow +from durable_workflow.client import Client +from durable_workflow.errors import ServerError, WorkflowNotFound +from durable_workflow.retry_policy import TransportRetryPolicy + + +def response(activities: Any = None) -> dict[str, Any]: + return {"workflow_id": "workflow/one", "run_id": "run?one", "activities": [] if activities is None else activities} + + +def client_for(payload: dict[str, Any], requests: list[httpx.Request], *, status: int = 200) -> Client: + def reply(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(status, json=payload) + + client = Client("https://runtime.example", control_token="synthetic-control", worker_token="synthetic-worker", + namespace="tenant-a", retry_policy=TransportRetryPolicy(max_attempts=1)) + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(reply)) + return client + + +async def test_actual_run_route_preserves_typed_activity_attempt_and_authority() -> None: + attempt = {"id": "attempt-2", "worker_attempt_id": "worker-attempt", "attempt_number": 2, + "status": "running", "task_id": "task-2", "lease_owner": "worker-1", + "lease_expires_at": "2026-11-04T12:00:30Z", "started_at": "2026-11-04T12:00:00Z", + "last_heartbeat_at": "2026-11-04T12:00:05Z", "last_heartbeat_progress": {"details": {"part": 3}}, + "closed_at": None, "can_continue": True, "cancel_requested": False, "stop_reason": None, + "failure_reason": None} + activity = {"id": "execution-1", "idempotency_key": "business-key", "sequence": 4, "type": "invoice", + "class": "InvoiceActivity", "execution_mode": "remote", "local_activity": False, + "history_authority": "typed_history", "diagnostic_only": False, + "history_event_types": ["ActivityScheduled", "ActivityStarted", "ActivityHeartbeatRecorded"], + "history_unsupported_reason": None, "row_status": "running", "attempt_id": "attempt-2", + "worker_attempt_id": "worker-attempt", "status": "running", "attempt_count": 2, + "retry_policy": {"snapshot_version": 1, "backoff_seconds": [0]}, "connection": "default", + "queue": "invoices", "last_heartbeat_at": attempt["last_heartbeat_at"], + "last_heartbeat_progress": attempt["last_heartbeat_progress"], "created_at": "2026-11-04T11:59:00Z", + "started_at": attempt["started_at"], "closed_at": None, "parallel_group_kind": "all", + "parallel_group_id": "group-1", "parallel_group_base_sequence": 3, "parallel_group_size": 2, + "parallel_group_index": 1, "parallel_group_path": [{"group_id": "group-1", "index": 1}], + "reused_from_run_id": "old-run", "reused_activity_execution_id": "old-execution", + "reused_recorded_at": "2026-11-03T12:00:00Z", "attempts": [attempt], + "future_metadata": {"preserve": [True, None]}} + requests: list[httpx.Request] = [] + async with client_for(response([activity]), requests) as client: + view = await client.get_run_activities("workflow/one", "run?one") + assert isinstance(view, durable_workflow.WorkflowActivityList) + assert (view.workflow_id, view.run_id) == ("workflow/one", "run?one") + item = view.activities[0] + assert isinstance(item, durable_workflow.WorkflowActivity) + assert item.activity_type == "invoice" + assert item.activity_class == "InvoiceActivity" + for key, value in activity.items(): + if key not in {"type", "class", "attempts", "future_metadata"}: + assert getattr(item, key) == value + assert item.raw == activity + assert item.attempts is not None + assert isinstance(item.attempts[0], durable_workflow.WorkflowActivityAttempt) + for key, value in attempt.items(): + assert getattr(item.attempts[0], key) == value + assert len(requests) == 1 + request = requests[0] + assert request.method == "GET" + assert request.url.raw_path == b"/api/workflows/workflow%2Fone/runs/run%3Fone/activities" + assert request.headers["X-Namespace"] == "tenant-a" + assert request.headers["Authorization"] == "Bearer synthetic-control" + + +async def test_empty_list_and_unavailable_optional_metadata_are_distinct() -> None: + async with client_for(response(), []) as client: + assert (await client.get_run_activities("workflow/one", "run?one")).activities == [] + async with client_for(response([{"id": "execution", "attempts": [{}]}, {"attempts": []}]), []) as client: + values = (await client.get_run_activities("workflow/one", "run?one")).activities + assert values[0].status is None + assert values[0].attempt_count is None + assert values[0].last_heartbeat_at is None + assert values[0].diagnostic_only is None + assert values[0].attempts is not None + assert values[0].attempts[0].can_continue is None + assert values[0].attempts[0].cancel_requested is None + assert values[1].attempts == [] + async with client_for(response([{}]), []) as client: + assert (await client.get_run_activities("workflow/one", "run?one")).activities[0].attempts is None + + +async def test_diagnostic_and_future_status_are_not_reinterpreted() -> None: + items = [{"status": "unsupported", "history_authority": "unsupported_terminal_without_history", + "diagnostic_only": True, "history_unsupported_reason": "terminal_activity_row_without_typed_history"}, + {"status": "future-paused", "attempt_count": 0, "local_activity": True}] + async with client_for(response(items), []) as client: + values = (await client.get_run_activities("workflow/one", "run?one")).activities + assert values[0].status == "unsupported" + assert values[0].diagnostic_only is True + assert values[0].history_unsupported_reason == "terminal_activity_row_without_typed_history" + assert values[1].status == "future-paused" + assert values[1].attempt_count == 0 + + +@pytest.mark.parametrize("change", [ + {"workflow_id": "other"}, {"run_id": "other"}, {"workflow_id": None}, {"run_id": None}, + {"activities": None}, {"activities": {}}, {"activities": ["private-marker"]}, +]) +async def test_changed_identity_or_invalid_collection_refuses(change: dict[str, Any]) -> None: + async with client_for({**response(), **change}, []) as client: + with pytest.raises(RuntimeError) as error: + await client.get_run_activities("workflow/one", "run?one") + assert "private-marker" not in str(error.value) + + +@pytest.mark.parametrize("missing", ["workflow_id", "run_id", "activities"]) +async def test_required_response_fields_cannot_fabricate_an_empty_view(missing: str) -> None: + payload = response() + del payload[missing] + async with client_for(payload, []) as client: + with pytest.raises(RuntimeError): + await client.get_run_activities("workflow/one", "run?one") + + +@pytest.mark.parametrize("item", [ + {"status": 1}, {"attempt_count": True}, {"local_activity": "false"}, {"diagnostic_only": 0}, + {"attempts": {}}, {"attempts": ["private-marker"]}, {"attempts": [{"can_continue": "true"}]}, + {"history_event_types": [1]}, {"parallel_group_path": [False]}, {"last_heartbeat_progress": "private-marker"}, +]) +async def test_malformed_known_metadata_is_not_coerced(item: dict[str, Any]) -> None: + async with client_for(response([item]), []) as client: + with pytest.raises(RuntimeError) as error: + await client.get_run_activities("workflow/one", "run?one") + assert "private-marker" not in str(error.value) + + +@pytest.mark.parametrize(("status", "error"), [(404, WorkflowNotFound), (503, ServerError)]) +async def test_http_error_propagates(status: int, error: type[Exception]) -> None: + requests: list[httpx.Request] = [] + async with client_for({"error": "not_found"}, requests, status=status) as client: + with pytest.raises(error): + await client.get_run_activities("workflow/one", "run?one") + assert len(requests) == 1 + + +@pytest.mark.parametrize("use_explicit_run", [False, True]) +async def test_handle_binds_its_selected_run(use_explicit_run: bool) -> None: + async with client_for(response(), []) as client: + handle = client.get_workflow_handle("workflow/one", run_id=None if use_explicit_run else "run?one") + view = await handle.get_activities("run?one") if use_explicit_run else await handle.get_activities() + assert view.run_id == "run?one" + + +async def test_handle_without_run_refuses_before_request() -> None: + requests: list[httpx.Request] = [] + async with client_for(response(), requests) as client: + handle = client.get_workflow_handle("workflow/one") + with pytest.raises(ValueError, match="run_id"): + await handle.get_activities() + assert not requests