Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@ to include examples, links to docs, or any other relevant information.

### Fixed

- Nexus-context workflow/activity starts no longer set `on_conflict_options` when there are no links
or callbacks to attach.

### Security

## [1.32.0] - 2026-08-24
Expand Down
25 changes: 16 additions & 9 deletions temporalio/nexus/_operation_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -821,19 +821,17 @@ def _apply_nexus_context_to_start_workflow_request( # pyright: ignore[reportUnu
This is a no-op outside a Nexus operation context. Within one, it attaches
inbound links and configures conflict handling to preserve Nexus metadata.
The Nexus request ID and completion callbacks are added only when the
workflow is backing the Nexus operation.
workflow is backing the Nexus operation. on_conflict_options is populated
only when there are links or callbacks to attach.
"""
nexus_ctx = _try_start_operation_context()
if nexus_ctx is not None:
req.on_conflict_options.attach_request_id = True
req.on_conflict_options.attach_completion_callbacks = True
req.on_conflict_options.attach_links = True

request_links = nexus_ctx._get_request_links()

# Links are duplicated on request for compatibility with older server versions.
req.links.extend(request_links)

callbacks: list[NexusCallback] = []
if _in_nexus_backing_start_context():
req.request_id = nexus_ctx.nexus_context.request_id
callbacks = nexus_ctx._get_callbacks(
Expand All @@ -854,6 +852,11 @@ def _apply_nexus_context_to_start_workflow_request( # pyright: ignore[reportUnu
for callback in callbacks
)

if request_links or callbacks:
req.on_conflict_options.attach_request_id = True

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This option seems like it should be set regardless of if there are links or callbacks.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we do have a validation on the standalone activity implementation in chasm, please see: https://github.com/temporalio/temporal/blame/main/chasm/lib/activity/validator.go#L252-L268 and frontend service makes use of that validation here

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added tests ptal @VegetarianOrc

req.on_conflict_options.attach_completion_callbacks = True
req.on_conflict_options.attach_links = True


def _apply_start_workflow_response_to_nexus_context( # pyright: ignore[reportUnusedFunction]
workflow_handle: temporalio.client.WorkflowHandle[Any, Any],
Expand All @@ -872,16 +875,15 @@ def _apply_nexus_context_to_start_activity_request( # pyright: ignore[reportUnu
the Nexus request ID and configures conflict handling to preserve the Nexus
metadata. Inbound links are attached to the completion callback when the
activity backs the operation and to the request otherwise.
on_conflict_options is populated only when there are links or callbacks to
attach.
"""
nexus_ctx = _try_start_operation_context()
if nexus_ctx is not None:
req.on_conflict_options.attach_request_id = True
req.on_conflict_options.attach_completion_callbacks = True
req.on_conflict_options.attach_links = True

req.request_id = nexus_ctx.nexus_context.request_id
request_links = nexus_ctx._get_request_links()

callbacks: list[NexusCallback] = []
if _in_nexus_backing_start_context():
callbacks = nexus_ctx._get_callbacks(
OperationToken(
Expand All @@ -903,6 +905,11 @@ def _apply_nexus_context_to_start_activity_request( # pyright: ignore[reportUnu
else:
req.links.extend(request_links)

if request_links or callbacks:
req.on_conflict_options.attach_request_id = True
req.on_conflict_options.attach_completion_callbacks = True
req.on_conflict_options.attach_links = True


def _apply_start_activity_response_to_nexus_context( # pyright: ignore[reportUnusedFunction]
activity_id: str,
Expand Down
57 changes: 57 additions & 0 deletions tests/nexus/test_link_propagation.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,39 @@ def nexus_ctx() -> Generator[_TemporalStartOperationContext]:
)


@pytest.fixture
def nexus_ctx_linkless() -> Generator[_TemporalStartOperationContext]:
"""Like `nexus_ctx`, but with no inbound links and no callback URL."""
nexus_context = nexusrpc.handler.StartOperationContext(
service="svc",
operation="op",
headers={},
request_id="req-id",
callback_url=None,
inbound_links=[],
callback_headers={},
task_cancellation=_NexusTaskCancellation(),
)
ctx = temporalio.nexus._operation_context._TemporalStartOperationContext(
nexus_context=nexus_context,
client=mock.MagicMock(namespace=NAMESPACE),
info=lambda: temporalio.nexus.Info(
endpoint="endpoint", namespace=NAMESPACE, task_queue="tq"
),
_runtime_metric_meter=mock.MagicMock(),
_worker_shutdown_event=mock.MagicMock(),
)
token = temporalio.nexus._operation_context._temporal_start_operation_context.set(
ctx
)
try:
yield ctx
finally:
temporalio.nexus._operation_context._temporal_start_operation_context.reset(
token
)


def _make_client_impl(workflow_service: Any) -> _ClientImpl:
client = mock.MagicMock()
client.namespace = NAMESPACE
Expand Down Expand Up @@ -749,6 +782,30 @@ async def test_backing_activity_start_gets_nexus_request_fields() -> None:
assert list(req.completion_callbacks[0].links) == [_inbound_nexus_link()]


@pytest.mark.usefixtures("nexus_ctx_linkless")
async def test_activity_start_omits_on_conflict_options_when_linkless() -> None:
impl = _make_client_impl(mock.MagicMock())

req = await impl._build_start_activity_execution_request(_start_activity_input())

assert req.request_id == "req-id"
assert not req.HasField("on_conflict_options")


@pytest.mark.usefixtures("nexus_ctx_linkless")
async def test_backing_activity_start_omits_on_conflict_options_when_linkless() -> None:
impl = _make_client_impl(mock.MagicMock())

with temporalio.nexus._operation_context._nexus_backing_start_context():
req = await impl._build_start_activity_execution_request(
_start_activity_input()
)

assert req.request_id == "req-id"
assert len(req.completion_callbacks) == 0
assert not req.HasField("on_conflict_options")


# ── handler-level: backlinks land on the StartOperationResponse ──────────────────────────────

# A response link that a handler stashes on ctx.outbound_links, mimicking what a signal RPC inside
Expand Down
Loading