Expose read-only workflow and run identities

docs/reference/workflow.md

@@ -1,5 +1,18 @@
 # Workflow
 
+## Workflow and run identity
+
+Workflow code reads its task-bound identity through the read-only
+`ctx.workflow_id` and `ctx.run_id` properties. The values remain the same when
+that run suspends and replays. After continue-as-new, the successor retains the
+workflow ID and receives a new run ID; passing an older ID as workflow input
+does not change the context identity.
+
+Use these values to correlate an activity or log entry with its workflow and
+run. To inspect status from outside workflow code, use
+[`Client.describe_workflow()`](client.md) or a workflow query instead of an
+activity that repeatedly polls its own run.
+
 ## Signal, update and query handlers
 
 Handlers decorated with `@workflow.signal` or `@workflow.update` must be

src/durable_workflow/workflow.py

@@ -2056,6 +2056,16 @@ def __init__(
         self._workflow_stream_command_counter = 0
         self.logger = _ReplayLogger(_REPLAY_LOGGER)
 
+    @property
+    def workflow_id(self) -> str:
+        """Original workflow identity, retained across continue-as-new runs."""
+        return self._workflow_id
+
+    @property
+    def run_id(self) -> str:
+        """Current run identity, stable on replay and new after continue-as-new."""
+        return self._run_id
+
     def cancellation_scope(
         self, body: Callable[[], Any], *, shield_parent: bool = False,
     ) -> Generator[Any, Any, Any]:

tests/fixtures/replay_regressions/workflow-identity-cold-replay.json

@@ -0,0 +1,31 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "workflow-identity-cold-replay",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {
+    "type": "identity-across-replay",
+    "workflow_id": "017",
+    "run_id": "original-run",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "sequence": 1,
+      "event_type": "ActivityCompleted",
+      "payload": {
+        "result": {"codec": "avro", "blob": "wwHioz3/VYAiNwTGAQ=="}
+      }
+    }
+  ],
+  "expected": {
+    "type": "complete_workflow",
+    "result": {
+      "before": ["017", "original-run"],
+      "after": ["017", "original-run"],
+      "value": 99
+    }
+  }
+}

tests/fixtures/replay_regressions/workflow-identity-continue-successor.json

@@ -0,0 +1,30 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "workflow-identity-continue-successor",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {
+    "type": "identity-continue-as-new",
+    "workflow_id": "chain",
+    "run_id": "run-two",
+    "input": [false, "chain", "run-one"],
+    "payload_codec": "avro"
+  },
+  "command_sequence": [
+    {
+      "type": "complete_workflow",
+      "result": {
+        "current": ["chain", "run-two"],
+        "prior": ["chain", "run-one"]
+      }
+    }
+  ],
+  "expected": {
+    "type": "complete_workflow",
+    "result": {
+      "current": ["chain", "run-two"],
+      "prior": ["chain", "run-one"]
+    }
+  }
+}

tests/test_replay_regression_corpus.py

@@ -46,6 +46,7 @@
     UpdateSignalConditionTimerWorkflow,
 )
 from tests.test_worker import compatible_cluster_info
+from tests.test_workflow_identity import IdentityAcrossReplay, IdentityContinueAsNew
 
 FIXTURE_SCHEMA = "durable-workflow.replay-regression/v1"
 FIXTURE_DIR = Path(__file__).parent / "fixtures" / "replay_regressions"
@@ -340,6 +341,8 @@ def run(self, ctx: WorkflowContext, cleanup_seconds: int = 0):  # type: ignore[n
 
 
 WORKFLOWS = [
+    IdentityAcrossReplay,
+    IdentityContinueAsNew,
     PersistedClockWorkflow,
     NexusCooperativeDeliveryWorkflow,
     NexusTerminalDeliveryWorkflow,
@@ -514,6 +517,8 @@ def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]:
             history,
             start_input,
             workflow_type=workflow_type,
+            workflow_id=workflow.get("workflow_id"),
+            run_id=workflow.get("run_id", ""),
             payload_codec=("json" if payload_codec is None else payload_codec),
         )
         commands = _command_documents(outcome.commands)

tests/test_workflow_identity.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import pytest
+
+from durable_workflow import serializer, workflow
+from durable_workflow.workflow import (
+    CompleteWorkflow,
+    ContinueAsNew,
+    ScheduleActivity,
+    WorkflowContext,
+    replay,
+)
+
+
+@workflow.defn(name="identity-across-replay")
+class IdentityAcrossReplay:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        before = [ctx.workflow_id, ctx.run_id]
+        value = yield ctx.schedule_activity("identity", before)
+        return {"before": before, "after": [ctx.workflow_id, ctx.run_id], "value": value}
+
+
+@pytest.mark.parametrize("workflow_id", ["workflow-identity", "0", "17", "017"])
+def test_identity_survives_activity_suspension_and_cold_replay(workflow_id: str) -> None:
+    run_id = "original-run"
+    first = replay(IdentityAcrossReplay, [], [], workflow_id=workflow_id, run_id=run_id)
+    assert len(first.commands) == 1
+    assert isinstance(first.commands[0], ScheduleActivity)
+    assert first.commands[0].arguments == [workflow_id, run_id]
+
+    history = [{"event_type": "ActivityCompleted", "payload": {
+        "result": serializer.envelope("done", codec=serializer.AVRO_CODEC),
+    }}]
+    resumed = replay(IdentityAcrossReplay, history, [], workflow_id=workflow_id, run_id=run_id)
+    assert len(resumed.commands) == 1
+    assert isinstance(resumed.commands[0], CompleteWorkflow)
+    assert resumed.commands[0].result == {
+        "before": [workflow_id, run_id], "after": [workflow_id, run_id], "value": "done",
+    }
+
+
+@workflow.defn(name="identity-continue-as-new")
+class IdentityContinueAsNew:
+    def run(  # type: ignore[no-untyped-def]
+        self, ctx: WorkflowContext, again: bool,
+        prior_workflow_id: str | None = None, prior_run_id: str | None = None,
+    ):
+        if again:
+            return ctx.continue_as_new(False, ctx.workflow_id, ctx.run_id)
+        return {"current": [ctx.workflow_id, ctx.run_id], "prior": [prior_workflow_id, prior_run_id]}
+
+
+def test_successor_context_has_same_workflow_and_new_run_identity() -> None:
+    original = replay(IdentityContinueAsNew, [], [True], workflow_id="chain", run_id="run-one")
+    assert len(original.commands) == 1
+    command = original.commands[0]
+    assert isinstance(command, ContinueAsNew)
+    assert command.arguments == [False, "chain", "run-one"]
+
+    successor = replay(
+        IdentityContinueAsNew, [], command.arguments, workflow_id="chain", run_id="run-two",
+    )
+    assert len(successor.commands) == 1
+    assert isinstance(successor.commands[0], CompleteWorkflow)
+    assert successor.commands[0].result == {"current": ["chain", "run-two"], "prior": ["chain", "run-one"]}
+
+
+@pytest.mark.parametrize("attribute", ["workflow_id", "run_id"])
+def test_workflow_identity_is_read_only(attribute: str) -> None:
+    ctx = WorkflowContext(workflow_id="workflow", run_id="run")
+    with pytest.raises(AttributeError):
+        setattr(ctx, attribute, "replacement")