[cross-repo from server#288] Conformance finding: replay workflow remains waiting on current artifacts (#102)

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -3009,6 +3009,9 @@ async def register_worker(
         sdk_version: str | None = None,
         build_id: str | None = None,
         capabilities: list[str] | None = None,
+        task_slots: dict[str, int] | None = None,
+        process_metrics: dict[str, Any] | None = None,
+        heartbeat_interval_seconds: int | None = None,
     ) -> Any:
         """Register this process with the server as a worker for ``task_queue``.
 
@@ -3041,6 +3044,12 @@ async def register_worker(
             body["max_concurrent_workflow_tasks"] = max_concurrent_workflow_tasks
         if max_concurrent_activity_tasks is not None:
             body["max_concurrent_activity_tasks"] = max_concurrent_activity_tasks
+        if task_slots is not None:
+            body["task_slots"] = task_slots
+        if process_metrics is not None:
+            body["process_metrics"] = process_metrics
+        if heartbeat_interval_seconds is not None:
+            body["heartbeat_interval_seconds"] = heartbeat_interval_seconds
         return await self._request("POST", "/worker/register", worker=True, json=body)
 
     async def heartbeat_worker(
@@ -3067,8 +3076,9 @@ async def heartbeat_worker(
 
         ``process_metrics`` is an optional dict with any subset of
         ``cpu_percent``, ``memory_bytes``, ``process_uptime_seconds``,
-        ``process_id``, and ``host`` — the SDK reports only what it has
-        cheap access to, and the server records exactly what was reported.
+        ``process_id``, ``process_started_at``, and ``host`` — the SDK
+        reports only what it has cheap access to, and the server records
+        exactly what was reported.
 
         Returns the server acknowledgement, which includes the advertised
         ``heartbeat_interval_seconds`` and ``stale_after_seconds`` so the

src/durable_workflow/worker.py

@@ -26,6 +26,7 @@
 import traceback
 import uuid
 from collections.abc import Awaitable, Callable, Iterable, Mapping
+from datetime import datetime, timezone
 from types import FunctionType
 from typing import Any
 
@@ -526,6 +527,11 @@ def __init__(
         self._activity_inflight = 0
         self._heartbeat_interval = float(heartbeat_interval)
         self._process_started_at = time.time()
+        self._process_started_at_iso = (
+            datetime.fromtimestamp(self._process_started_at, timezone.utc)
+            .isoformat(timespec="milliseconds")
+            .replace("+00:00", "Z")
+        )
         # CPU sampling baseline. The heartbeat reports an *instantaneous*
         # cpu_percent — CPU time burned in the interval since the previous
         # heartbeat, divided by that interval — rather than the lifetime
@@ -643,6 +649,8 @@ async def _register(self) -> None:
             max_concurrent_activity_tasks=self.max_concurrent_activity_tasks,
             build_id=self.build_id,
             capabilities=[QUERY_TASKS_CAPABILITY] if self._query_tasks_supported else None,
+            task_slots=self._current_task_slots(),
+            process_metrics=self._current_process_metrics(),
         )
         # Adapt to the server-advertised cadence when present so a cluster
         # can pin the worker fleet's heartbeat beat without each worker
@@ -1635,6 +1643,7 @@ def _current_process_metrics(self) -> dict[str, Any]:
         metrics: dict[str, Any] = {
             "process_uptime_seconds": int(now - self._process_started_at),
             "process_id": os.getpid(),
+            "process_started_at": self._process_started_at_iso,
         }
 
         # ``memory_bytes`` is the *current* resident set size, not the

tests/test_client.py

@@ -2539,6 +2539,28 @@ async def test_register_sends_worker_capacity_when_configured(self, client: Clie
             assert body["max_concurrent_workflow_tasks"] == 3
             assert body["max_concurrent_activity_tasks"] == 7
 
+    @pytest.mark.asyncio
+    async def test_register_sends_worker_process_state_when_configured(self, client: Client) -> None:
+        resp = _mock_response(201, {"worker_id": "w1", "registered": True})
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            await client.register_worker(
+                worker_id="w1",
+                task_queue="q1",
+                task_slots={"workflow_available": 2, "activity_available": 5},
+                process_metrics={
+                    "host": "worker-host",
+                    "process_id": 1234,
+                    "process_started_at": "2026-05-18T21:00:00Z",
+                },
+            )
+            body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
+            assert body["task_slots"] == {"workflow_available": 2, "activity_available": 5}
+            assert body["process_metrics"] == {
+                "host": "worker-host",
+                "process_id": 1234,
+                "process_started_at": "2026-05-18T21:00:00Z",
+            }
+
     @pytest.mark.asyncio
     async def test_register_rejects_non_positive_worker_capacity(self, client: Client) -> None:
         with pytest.raises(ValueError, match="max_concurrent_workflow_tasks"):

tests/test_worker.py

@@ -227,6 +227,13 @@ async def test_register(self, mock_client: AsyncMock) -> None:
         assert call_kwargs["max_concurrent_workflow_tasks"] == 10
         assert call_kwargs["max_concurrent_activity_tasks"] == 10
         assert call_kwargs["capabilities"] == ["query_tasks"]
+        assert call_kwargs["task_slots"] == {
+            "workflow_available": 10,
+            "activity_available": 10,
+        }
+        process_metrics = call_kwargs["process_metrics"]
+        assert process_metrics["process_id"] > 0
+        assert "process_started_at" in process_metrics
 
     @pytest.mark.asyncio
     async def test_register_omits_query_task_capability_when_server_does_not_support_it(
@@ -1959,6 +1966,7 @@ async def test_run_drives_periodic_heartbeats_with_slot_state(
         assert "process_id" in process_metrics
         assert process_metrics["process_id"] > 0
         assert "process_uptime_seconds" in process_metrics
+        assert "process_started_at" in process_metrics
 
     @pytest.mark.asyncio
     async def test_register_adopts_server_advertised_heartbeat_cadence(