From 0ccb9b747087048a1035bfe901a20e00a6a954f9 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sun, 11 Oct 2026 09:20:32 +0000 Subject: [PATCH 1/2] Address Nexus catalogs with an explicit durable target namespace --- src/durable_workflow/client.py | 30 ++++- src/durable_workflow/nexus.py | 11 ++ src/durable_workflow/worker.py | 1 + src/durable_workflow/workflow.py | 15 +++ .../nexus-target-namespace-changed.json | 19 +++ tests/test_nexus_namespace.py | 108 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 11 ++ tests/test_worker.py | 10 +- 8 files changed, 201 insertions(+), 4 deletions(-) create mode 100644 tests/fixtures/replay_regressions/nexus-target-namespace-changed.json create mode 100644 tests/test_nexus_namespace.py diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index d20b751..522b77e 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -19,6 +19,7 @@ from __future__ import annotations import asyncio +import copy import hashlib import json as json_module import math @@ -2672,6 +2673,26 @@ def get_workflow_handle( """ return WorkflowHandle(self, workflow_id=workflow_id, run_id=run_id, workflow_type=workflow_type) + def _nexus_target_client(self, target_namespace: str | None) -> Client: + if target_namespace is not None and not target_namespace.strip(): + raise ValueError("target_namespace must be nonempty") + if target_namespace is None or target_namespace == self.namespace: + return self + self._auth_token(worker=False) + scoped = copy.copy(self) + scoped.namespace = target_namespace + # Share the HTTP pool, never the caller's namespace discovery or + # provider credentials. The worker keeps using the original client. + scoped.worker_token = None + scoped._cluster_info = None + scoped._cluster_info_lock = asyncio.Lock() + scoped._runtime_external_payload_transport_cache = None + scoped._runtime_external_payload_transport_resolved = False + scoped.external_storage = None + scoped.external_storage_threshold_bytes = None + scoped.external_storage_cache = ExternalPayloadCache() + return scoped + async def execute_nexus_operation( self, endpoint_name: str, @@ -2685,6 +2706,7 @@ async def execute_nexus_operation( idempotency_key: str | None = None, payload_codec: str | None = None, caller_namespace: str | None = None, + target_namespace: str | None = None, caller_workflow_instance_id: str | None = None, caller_workflow_run_id: str | None = None, service_sdk_language: str | None = None, @@ -2708,7 +2730,11 @@ async def execute_nexus_operation( the worker uses this client method to make the single durable execute request and records the returned service-call surface into workflow history. + + ``target_namespace`` selects the service catalog and its payload + storage. The caller identity and worker namespace are preserved. """ + execution_client = self._nexus_target_client(target_namespace) request_body = nexus_request_payload( arguments=list(arguments) if arguments is not None else [], payload_codec=payload_codec, @@ -2737,7 +2763,7 @@ async def execute_nexus_operation( context = f"{endpoint_name}/{service_name}/{operation_name}" try: - data = await self._request("POST", path, json=request_body, context=context) + data = await execution_client._request("POST", path, json=request_body, context=context) except ServerError as exc: if exc.status != 409 or not isinstance(exc.body, dict): raise @@ -2747,6 +2773,7 @@ async def execute_nexus_operation( service_name=service_name, operation_name=operation_name, request_payload=request_body, + target_namespace=target_namespace, service_sdk_language=service_sdk_language, artifact_tuple=artifact_tuple, published_artifact_worker_execution=published_artifact_worker_execution, @@ -2770,6 +2797,7 @@ async def execute_nexus_operation( service_name=service_name, operation_name=operation_name, request_payload=request_body, + target_namespace=target_namespace, service_sdk_language=service_sdk_language, artifact_tuple=artifact_tuple, published_artifact_worker_execution=published_artifact_worker_execution, diff --git a/src/durable_workflow/nexus.py b/src/durable_workflow/nexus.py index 941e54d..9e220b4 100644 --- a/src/durable_workflow/nexus.py +++ b/src/durable_workflow/nexus.py @@ -49,6 +49,7 @@ def stable_nexus_idempotency_key( service_name: str, operation_name: str, arguments: Sequence[Any], + target_namespace: str | None = None, ) -> str: """Return a deterministic idempotency key for one workflow-side Nexus call.""" source = { @@ -61,6 +62,8 @@ def stable_nexus_idempotency_key( "operation_name": operation_name, "arguments": list(arguments), } + if target_namespace is not None: + source["target_namespace"] = target_namespace encoded = json.dumps(source, sort_keys=True, separators=(",", ":"), default=_json_default) return f"dw-py-nexus-{hashlib.sha256(encoded.encode()).hexdigest()}" @@ -134,6 +137,7 @@ class NexusOperationResult: caller_observed_error_type: str | None = None typed_error_message: str | None = None result: Any = None + target_namespace: str | None = None @classmethod def from_response( @@ -147,6 +151,7 @@ def from_response( service_sdk_language: str | None = None, artifact_tuple: Mapping[str, Any] | None = None, published_artifact_worker_execution: bool | None = None, + target_namespace: str | None = None, ) -> NexusOperationResult: call = data.get("service_call") source = call if isinstance(call, Mapping) else data @@ -188,6 +193,7 @@ def from_response( typed_error_message=_optional_str(source.get("typed_error_message")) or _optional_str(data.get("typed_error_message")), result=response_value, + target_namespace=target_namespace, ) @classmethod @@ -199,6 +205,9 @@ def from_recorded_payload(cls, value: Any, *, expected: Mapping[str, Any]) -> Ne if value.get("version") != NEXUS_OPERATION_RESULT_VERSION: raise ValueError("recorded Nexus operation result version is not supported") + if value.get("target_namespace") != expected.get("target_namespace"): + raise ValueError("recorded Nexus target namespace does not match the current workflow command") + for key in ("endpoint_name", "service_name", "operation_name", "idempotency_key"): expected_value = expected.get(key) recorded_value = value.get(key) @@ -235,6 +244,7 @@ def from_recorded_payload(cls, value: Any, *, expected: Mapping[str, Any]) -> Ne caller_observed_error_type=_optional_str(value.get("caller_observed_error_type")), typed_error_message=_optional_str(value.get("typed_error_message")), result=value.get("result"), + target_namespace=_optional_str(value.get("target_namespace")), ) if result.is_failure: raise result.to_failure() @@ -287,6 +297,7 @@ def to_recorded_payload(self) -> dict[str, Any]: "endpoint_name": self.endpoint_name, "service_name": self.service_name, "operation_name": self.operation_name, + "target_namespace": self.target_namespace, "idempotency_key": self.request_payload.get("idempotency_key"), "caller_workflow_instance_id": self.caller_workflow_instance_id, "caller_workflow_run_id": self.caller_workflow_run_id, diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index d8c853c..c285675 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -1462,6 +1462,7 @@ async def _resolve_workflow_nexus_commands( idempotency_key=command.idempotency_key, payload_codec=command.payload_codec, caller_namespace=command.caller_namespace, + target_namespace=command.target_namespace, caller_workflow_instance_id=_string_or_none(task.get("workflow_id")), caller_workflow_run_id=_string_or_none(task.get("run_id")), service_sdk_language=command.service_sdk_language, diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index c6e47f7..1b68cdb 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -1196,6 +1196,7 @@ class NexusServiceCall: memo: dict[str, Any] | None = None search_attributes: dict[str, Any] | None = None duplicate_start_policy: str | None = None + target_namespace: str | None = None def to_server_command(self, *args: Any, **kwargs: Any) -> dict[str, Any]: raise RuntimeError( @@ -1209,6 +1210,7 @@ def expected_recorded_metadata(self) -> dict[str, Any]: "service_name": self.service_name, "operation_name": self.operation_name, "idempotency_key": self.idempotency_key, + "target_namespace": self.target_namespace, } def recorded_result(self, value: Any) -> NexusOperationResult: @@ -2402,6 +2404,7 @@ def call_nexus_service( wait_for: str | None = "completed", wait_timeout_seconds: int | None = None, caller_namespace: str | None = None, + target_namespace: str | None = None, service_sdk_language: str | None = None, artifact_tuple: Mapping[str, Any] | None = None, published_artifact_worker_execution: bool | None = None, @@ -2420,7 +2423,12 @@ def call_nexus_service( If the service reports a typed failure, replay raises :class:`~durable_workflow.errors.NexusOperationFailed` at the yield point so workflow code can compensate or let the workflow fail. + + ``target_namespace`` addresses a service in another namespace and + participates in the durable call identity and replay checks. """ + if target_namespace is not None and not target_namespace.strip(): + raise ValueError("target_namespace must be nonempty") args = list(arguments) if arguments is not None else [] if idempotency_key is None: idempotency_key = stable_nexus_idempotency_key( @@ -2431,6 +2439,7 @@ def call_nexus_service( service_name=service_name, operation_name=operation_name, arguments=args, + target_namespace=target_namespace, ) self._nexus_call_counter += 1 return NexusServiceCall( @@ -2444,6 +2453,7 @@ def call_nexus_service( wait_for=wait_for, wait_timeout_seconds=wait_timeout_seconds, caller_namespace=caller_namespace, + target_namespace=target_namespace, service_sdk_language=service_sdk_language, artifact_tuple=dict(artifact_tuple) if artifact_tuple is not None else None, published_artifact_worker_execution=published_artifact_worker_execution, @@ -6589,6 +6599,11 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if result_cursor < len(resolved_results): _assert_next_step_matches(cmd) recorded_value = resolved_results[result_cursor] + if isinstance(recorded_value, Mapping) and recorded_value.get("target_namespace") != cmd.target_namespace: + raise NonDeterministicReplayError( + current_call_sequence, "NexusServiceCall", ["SideEffectRecorded"], + detail="nexus_target_namespace_changed: recorded target differs from the current workflow command", + ) result_cursor += 1 try: next_value = cmd.recorded_result(recorded_value) diff --git a/tests/fixtures/replay_regressions/nexus-target-namespace-changed.json b/tests/fixtures/replay_regressions/nexus-target-namespace-changed.json new file mode 100644 index 0000000..10d5883 --- /dev/null +++ b/tests/fixtures/replay_regressions/nexus-target-namespace-changed.json @@ -0,0 +1,19 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "nexus-target-namespace-changed", + "protocol_version": "1.19", + "bindings": ["python"], + "workflow": {"type": "tests.replay.nexus-target", "input": [], "payload_codec": "avro"}, + "history": [{"event_type": "SideEffectRecorded", "payload": { + "sequence": 1, + "payload_codec": "avro", + "result": "wwHioz3/VYAiNw4WDHNjaGVtYQpqZHVyYWJsZS13b3JrZmxvdy52Mi5zZGstcHl0aG9uLm5leHVzLW9wZXJhdGlvbi1yZXN1bHQOdmVyc2lvbgQCEGFjY2VwdGVkAgEec2VydmljZV9jYWxsX2lkCiBvcmlnaW5hbC1jYWxsLWlkGmVuZHBvaW50X25hbWUKDmdyZWV0ZXIYc2VydmljZV9uYW1lCgxzaGFyZWQcb3BlcmF0aW9uX25hbWUKCmdyZWV0HmlkZW1wb3RlbmN5X2tleQoab3JpZ2luYWwtY2FsbCB0YXJnZXRfbmFtZXNwYWNlCgxzaGFyZWQMc3RhdHVzChJjb21wbGV0ZWQMcmVzdWx0ChpzZXJ2aWNlLXZhbHVlAA==" + }}], + "expected": {"command_sequence": []}, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "message_contains": "nexus_target_namespace_changed", + "workflow_sequence": 1 + } +} diff --git a/tests/test_nexus_namespace.py b/tests/test_nexus_namespace.py new file mode 100644 index 0000000..a0f2e84 --- /dev/null +++ b/tests/test_nexus_namespace.py @@ -0,0 +1,108 @@ +import asyncio +import json + +import httpx +import pytest + +from durable_workflow import Client, serializer, workflow +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.nexus import NexusOperationResult, stable_nexus_idempotency_key +from durable_workflow.workflow import CompleteWorkflow, NexusServiceCall, WorkflowContext, replay + + +@pytest.mark.asyncio +async def test_target_requests_and_concurrent_worker_heartbeat_keep_separate_scopes(): + requests = [] + + async def handler(request): + requests.append(request) + await asyncio.sleep(0) + if request.url.path == "/api/cluster/info": + return httpx.Response(200, json={"namespace": {"name": request.headers["x-namespace"]}}) + return httpx.Response(200, json={"accepted": True, "id": "call-1", "status": "started"}) + + async with Client("https://runtime.example", namespace="tenant-b", control_token="control", worker_token="worker") as client: + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler)) + client._cluster_info = {"namespace": {"name": "tenant-b"}} + result, _ = await asyncio.gather( + client.execute_nexus_operation("greeter", "shared", "greet", ["Ada"], target_namespace="shared", + caller_workflow_instance_id="caller", caller_workflow_run_id="original-run"), + client.heartbeat_worker(worker_id="caller-worker"), + ) + assert client.namespace == "tenant-b" + assert client._cluster_info == {"namespace": {"name": "tenant-b"}} + + nexus = next(request for request in requests if request.url.path.endswith("/execute")) + heartbeat = next(request for request in requests if request.url.path == "/api/worker/heartbeat") + assert nexus.headers["x-namespace"] == "shared" + assert nexus.headers["authorization"] == "Bearer control" + assert heartbeat.headers["x-namespace"] == "tenant-b" + assert heartbeat.headers["authorization"] == "Bearer worker" + body = json.loads(nexus.content) + assert body["caller_namespace"] == "tenant-b" + assert body["caller_workflow_instance_id"] == "caller" + assert body["caller_workflow_run_id"] == "original-run" + assert result.target_namespace == "shared" + assert result.to_recorded_payload()["target_namespace"] == "shared" + + +@pytest.mark.asyncio +async def test_target_scope_discards_caller_discovery_payload_cache_and_provider(): + client = Client("https://runtime.example", namespace="tenant-b", token="shared-token") + client._cluster_info = {"namespace": {"name": "tenant-b"}} + client._runtime_external_payload_transport_resolved = True + target = client._nexus_target_client("shared") + assert target._http is client._http + assert target.namespace == "shared" + assert target._cluster_info is None + assert target._cluster_info_lock is not client._cluster_info_lock + assert not target._runtime_external_payload_transport_resolved + assert target.external_storage is None + assert target.external_storage_cache is not client.external_storage_cache + await client.aclose() + + +@workflow.defn(name="nexus-target-caller") +class NexusTargetCaller: + def run(self, ctx: WorkflowContext, target: str | None): + result = yield ctx.call_nexus_service("greeter", "shared", "greet", ["Ada"], + target_namespace=target, idempotency_key="original-call") + return result.service_call_id + + +@pytest.mark.parametrize("target", [None, "shared"]) +def test_original_target_replays_and_changed_target_is_refused(target): + command = replay(NexusTargetCaller, [], [target], workflow_id="caller", run_id="original-run").commands[0] + assert isinstance(command, NexusServiceCall) + result = NexusOperationResult(accepted=True, service_call_id="call-1", endpoint_name="greeter", + service_name="shared", operation_name="greet", target_namespace=target, + request_payload={"idempotency_key": command.idempotency_key}, status="completed") + history = [{"event_type": "SideEffectRecorded", "payload": { + "sequence": 1, "payload_codec": "avro", "result": serializer.encode(result.to_recorded_payload(), codec="avro"), + }}] + outcome = replay(NexusTargetCaller, history, [target], workflow_id="caller", run_id="original-run", payload_codec="avro") + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == "call-1" + changed = "other" if target else "shared" + with pytest.raises(NonDeterministicReplayError, match="nexus_target_namespace_changed"): + replay(NexusTargetCaller, history, [changed], workflow_id="caller", run_id="original-run", payload_codec="avro") + + +def test_explicit_target_participates_in_call_identity_and_default_stays_compatible(): + identity = dict(workflow_id="caller", run_id="original-run", ordinal=0, endpoint_name="greeter", + service_name="shared", operation_name="greet", arguments=["Ada"]) + legacy = stable_nexus_idempotency_key(**identity) + assert stable_nexus_idempotency_key(**identity, target_namespace=None) == legacy + assert stable_nexus_idempotency_key(**identity, target_namespace="shared") != legacy + assert stable_nexus_idempotency_key(**identity, target_namespace="shared") != stable_nexus_idempotency_key( + **identity, target_namespace="other") + + +@pytest.mark.asyncio +async def test_empty_target_is_refused_before_network_and_worker_credentials_are_not_reused(): + async with Client("https://runtime.example", namespace="tenant-b", worker_token="worker-only") as client: + with pytest.raises(ValueError, match="nonempty"): + await client.execute_nexus_operation("greeter", "shared", "greet", target_namespace=" ") + with pytest.raises(ValueError, match="control-plane requests require"): + await client.execute_nexus_operation("greeter", "shared", "greet", target_namespace="shared") diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 98405b5..dffadaa 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -259,7 +259,18 @@ async def run(self, ctx: WorkflowContext) -> str: return "body was never executed" +@workflow.defn(name="tests.replay.nexus-target") +class NexusTargetWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + result = yield ctx.call_nexus_service( + "greeter", "shared", "greet", ["Ada"], + target_namespace="other", idempotency_key="original-call", + ) + return result.service_call_id + + WORKFLOWS = [ + NexusTargetWorkflow, UnsupportedAsyncWorkflow, ChildPolicyAuthorWorkflow, ColdReplacementSatisfiedConditionWorkflow, diff --git a/tests/test_worker.py b/tests/test_worker.py index 6c26752..13ead6d 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -124,12 +124,13 @@ def run(self, ctx, request): # type: ignore[no-untyped-def] @workflow.defn(name="nexus-worker-wf") class NexusWorkerWorkflow: - def run(self, ctx): # type: ignore[no-untyped-def] + def run(self, ctx, target=None): # type: ignore[no-untyped-def] result = yield ctx.call_nexus_service( "greeter", "shared", "greet", ["Ada"], + target_namespace=target, service_sdk_language="workflow-php", artifact_tuple={"sdk-python": "0.4.95"}, published_artifact_worker_execution=True, @@ -1823,8 +1824,9 @@ async def test_complete_on_resolved_activity(self, mock_client: AsyncMock) -> No assert serializer.decode(commands[0]["result"]["blob"], codec="avro") == "done" @pytest.mark.asyncio + @pytest.mark.parametrize("target", [None, "shared"]) async def test_workflow_nexus_service_call_records_side_effect_and_resumes( - self, mock_client: AsyncMock + self, mock_client: AsyncMock, target: str | None, ) -> None: async def fake_execute(endpoint_name: str, service_name: str, operation_name: str, arguments: list, **kwargs): return NexusOperationResult( @@ -1836,6 +1838,7 @@ async def fake_execute(endpoint_name: str, service_name: str, operation_name: st caller_workflow_instance_id=kwargs["caller_workflow_instance_id"], caller_workflow_run_id=kwargs["caller_workflow_run_id"], service_sdk_language=kwargs["service_sdk_language"], + target_namespace=kwargs["target_namespace"], request_payload={ "arguments": list(arguments), "idempotency_key": kwargs["idempotency_key"], @@ -1858,7 +1861,7 @@ async def fake_execute(endpoint_name: str, service_name: str, operation_name: st "workflow_type": "nexus-worker-wf", "workflow_task_attempt": 1, "history_events": [], - "arguments": serializer.encode([], codec="avro"), + "arguments": serializer.encode([target], codec="avro"), "payload_codec": "avro", } @@ -1869,6 +1872,7 @@ async def fake_execute(endpoint_name: str, service_name: str, operation_name: st assert execute_call.args[:4] == ("greeter", "shared", "greet", ["Ada"]) assert execute_call.kwargs["caller_workflow_instance_id"] == "wf-caller" assert execute_call.kwargs["caller_workflow_run_id"] == "run-caller" + assert execute_call.kwargs["target_namespace"] == target assert execute_call.kwargs["service_sdk_language"] == "workflow-php" assert execute_call.kwargs["raise_on_failure"] is False assert execute_call.kwargs["idempotency_key"].startswith("dw-py-nexus-") From a7c5d91d84add345c13e4b482ed15b518f5e7c6c Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sun, 11 Oct 2026 09:22:07 +0000 Subject: [PATCH 2/2] Keep namespace replay checks within lint limits --- src/durable_workflow/workflow.py | 7 +++++-- tests/test_nexus_namespace.py | 16 ++++++++++++---- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 1b68cdb..85bd674 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -6599,10 +6599,13 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if result_cursor < len(resolved_results): _assert_next_step_matches(cmd) recorded_value = resolved_results[result_cursor] - if isinstance(recorded_value, Mapping) and recorded_value.get("target_namespace") != cmd.target_namespace: + if ( + isinstance(recorded_value, Mapping) + and recorded_value.get("target_namespace") != cmd.target_namespace + ): raise NonDeterministicReplayError( current_call_sequence, "NexusServiceCall", ["SideEffectRecorded"], - detail="nexus_target_namespace_changed: recorded target differs from the current workflow command", + detail="nexus_target_namespace_changed: recorded target differs from the workflow command", ) result_cursor += 1 try: diff --git a/tests/test_nexus_namespace.py b/tests/test_nexus_namespace.py index 7db6734..4ff77ac 100644 --- a/tests/test_nexus_namespace.py +++ b/tests/test_nexus_namespace.py @@ -22,7 +22,9 @@ async def handler(request): return httpx.Response(200, json={"namespace": {"name": request.headers["x-namespace"]}}) return httpx.Response(200, json={"accepted": True, "id": "call-1", "status": "started"}) - async with Client("https://runtime.example", namespace="tenant-b", control_token="control", worker_token="worker") as client: + async with Client( + "https://runtime.example", namespace="tenant-b", control_token="control", worker_token="worker", + ) as client: await client._http.aclose() client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler)) client._cluster_info = {"namespace": {"name": "tenant-b"}} @@ -71,12 +73,16 @@ def handler(request): }) return httpx.Response(200, json={"accepted": True, "id": "call-1", "status": "started"}) - async with Client("https://runtime.example", namespace="tenant-b", control_token="control", worker_token="worker") as client: + async with Client( + "https://runtime.example", namespace="tenant-b", control_token="control", worker_token="worker", + ) as client: await client._http.aclose() client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handler)) client._cluster_info = {"namespace": {"name": "tenant-b"}} client._runtime_external_payload_transport_resolved = True - result = await client.execute_nexus_operation("greeter", "shared", "greet", ["Ada" * 128], target_namespace="shared") + result = await client.execute_nexus_operation( + "greeter", "shared", "greet", ["Ada" * 128], target_namespace="shared", + ) assert client.namespace == "tenant-b" assert client._runtime_external_payload_transport_cache is None @@ -125,7 +131,9 @@ def test_original_target_replays_and_changed_target_is_refused(target): history = [{"event_type": "SideEffectRecorded", "payload": { "sequence": 1, "payload_codec": "avro", "result": serializer.encode(result.to_recorded_payload(), codec="avro"), }}] - outcome = replay(NexusTargetCaller, history, [target], workflow_id="caller", run_id="original-run", payload_codec="avro") + outcome = replay( + NexusTargetCaller, history, [target], workflow_id="caller", run_id="original-run", payload_codec="avro", + ) assert isinstance(outcome.commands[0], CompleteWorkflow) assert outcome.commands[0].result == "call-1" changed = "other" if target else "shared"