From 1b6c32cbe982bb3b9a1de264c5cddef21aae4073 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 05:41:07 +0000 Subject: [PATCH 1/3] test: cover workflow identity across replay and continuation (#193) --- tests/test_workflow_identity.py | 72 +++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 tests/test_workflow_identity.py diff --git a/tests/test_workflow_identity.py b/tests/test_workflow_identity.py new file mode 100644 index 0000000..8abfda8 --- /dev/null +++ b/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") From 43971b07ab660de7a5d3a2ecf716595036cc8528 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 05:43:54 +0000 Subject: [PATCH 2/3] feat: expose read-only workflow and run identities (#193) --- docs/reference/workflow.md | 13 +++++++++++++ src/durable_workflow/workflow.py | 10 ++++++++++ 2 files changed, 23 insertions(+) diff --git a/docs/reference/workflow.md b/docs/reference/workflow.md index 841496b..d6d181b 100644 --- a/docs/reference/workflow.md +++ b/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 diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index d47e0e9..3c1aea9 100644 --- a/src/durable_workflow/workflow.py +++ b/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]: From 005a49321f45050464f55ec5a40bc3b14352f7e1 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 07:23:25 +0000 Subject: [PATCH 3/3] test: retain workflow identity replay corpus evidence --- .../workflow-identity-cold-replay.json | 31 +++++++++++++++++++ .../workflow-identity-continue-successor.json | 30 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 5 +++ 3 files changed, 66 insertions(+) create mode 100644 tests/fixtures/replay_regressions/workflow-identity-cold-replay.json create mode 100644 tests/fixtures/replay_regressions/workflow-identity-continue-successor.json diff --git a/tests/fixtures/replay_regressions/workflow-identity-cold-replay.json b/tests/fixtures/replay_regressions/workflow-identity-cold-replay.json new file mode 100644 index 0000000..6917e2b --- /dev/null +++ b/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 + } + } +} diff --git a/tests/fixtures/replay_regressions/workflow-identity-continue-successor.json b/tests/fixtures/replay_regressions/workflow-identity-continue-successor.json new file mode 100644 index 0000000..6e07269 --- /dev/null +++ b/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"] + } + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index f694f47..e2be47c 100644 --- a/tests/test_replay_regression_corpus.py +++ b/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)