feat: add signal-with-start-from-workflow features tests

dockerfiles/dynamicconfig/docker.yaml

@@ -22,3 +22,9 @@ frontend.enableUpdateWorkflowExecutionAsyncAccepted:
 frontend.workerVersioningWorkflowAPIs:
   - value: true
     constraints: {}
+history.enableSignalWithStartFromWorkflow:
+  - value: true
+    constraints: {}
+history.enableChasm:
+  - value: true
+    constraints: {}

dockerfiles/go.Dockerfile

@@ -1,5 +1,5 @@
 # Build in a full featured container
-FROM golang:1.24 as build
+FROM golang:1.25 AS build
 
 WORKDIR /app
 

features/signalwithstart/both_workflows_visible/README.md

@@ -0,0 +1,13 @@
+# Signal-with-start from a workflow: payloads, memo, and visibility
+
+A workflow calls `workflow.signal_with_start_workflow(...)` passing a workflow
+input argument, a signal argument, and a memo. This exercises the full
+payload-serialization path (covering what the proto-binary server test verifies,
+since Python encodes these payloads automatically).
+
+Verifies: both the caller and the started target complete; the target returns the
+signal value; and the memo passed in the request is visible on the target.
+
+## Server requirements
+Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and
+`history.enableChasm=true`.

features/signalwithstart/both_workflows_visible/feature.py

@@ -0,0 +1,90 @@
+import uuid
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Optional
+
+from temporalio import workflow
+from temporalio.client import WorkflowHandle
+
+from harness.python.feature import Runner, register_feature
+
+WORKFLOW_INPUT = "workflow-input"
+SIGNAL_VALUE = "signal-input"
+MEMO_KEY = "memo-key"
+MEMO_VALUE = "memo-value"
+
+
+@dataclass
+class SwsResult:
+    workflow_id: str
+    run_id: str
+
+
+@workflow.defn
+class TargetWorkflow:
+    def __init__(self) -> None:
+        self._signal_value: Optional[str] = None
+
+    @workflow.run
+    async def run(self, _input: str) -> str:
+        await workflow.wait_condition(lambda: self._signal_value is not None)
+        assert self._signal_value is not None
+        return self._signal_value
+
+    @workflow.signal
+    def my_signal(self, value: str) -> None:
+        self._signal_value = value
+
+
+@workflow.defn
+class CallerWorkflow:
+    @workflow.run
+    async def run(self, target_id: str) -> SwsResult:
+        handle = await workflow.signal_with_start_workflow(
+            TargetWorkflow.run,
+            WORKFLOW_INPUT,
+            id=target_id,
+            task_queue=workflow.info().task_queue,
+            signal=TargetWorkflow.my_signal,
+            signal_args=SIGNAL_VALUE,
+            memo={MEMO_KEY: MEMO_VALUE},
+        )
+        return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "")
+
+
+async def start(runner: Runner) -> WorkflowHandle:
+    target_id = f"signalwithstart-both-visible-target-{uuid.uuid4()}"
+    return await runner.client.start_workflow(
+        CallerWorkflow.run,
+        target_id,
+        id=f"signalwithstart-both-visible-caller-{uuid.uuid4()}",
+        task_queue=runner.task_queue,
+        execution_timeout=timedelta(minutes=1),
+    )
+
+
+async def check_result(runner: Runner, handle: WorkflowHandle) -> None:
+    # Caller completes and returns the started target's identity.
+    result: SwsResult = await handle.result()
+    assert result.run_id, "expected a non-empty target run id"
+
+    # Target completes after receiving the signal and returns the signal value.
+    target = runner.client.get_workflow_handle(result.workflow_id, run_id=result.run_id)
+    target_result = await target.result()
+    assert target_result == SIGNAL_VALUE, (
+        f"expected target to return {SIGNAL_VALUE!r}, got {target_result!r}"
+    )
+
+    # The memo passed in the signal-with-start request is visible on the target.
+    desc = await target.describe()
+    memo = await desc.memo()
+    assert memo.get(MEMO_KEY) == MEMO_VALUE, (
+        f"expected memo {MEMO_KEY}={MEMO_VALUE}, got {memo}"
+    )
+
+
+register_feature(
+    workflows=[CallerWorkflow, TargetWorkflow],
+    start=start,
+    check_result=check_result,
+)

features/signalwithstart/id_conflict_fail/README.md

@@ -0,0 +1,14 @@
+# Signal-with-start from a workflow: FAIL conflict policy is rejected
+
+A workflow calls `workflow.signal_with_start_workflow(...)` with
+`id_conflict_policy=FAIL`. This policy is not supported for signal-with-start, so
+the server rejects the scheduled Nexus operation. The rejection surfaces as a
+workflow task failure (not a catchable error), so the caller workflow never
+completes.
+
+Verifies: the caller's history contains a workflow task failure whose message
+indicates the conflict policy is not supported.
+
+## Server requirements
+Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and
+`history.enableChasm=true`.

features/signalwithstart/id_conflict_fail/feature.py

@@ -0,0 +1,105 @@
+import uuid
+from datetime import timedelta
+from typing import Optional
+
+from temporalio import workflow
+from temporalio.api.enums.v1 import EventType
+from temporalio.client import WorkflowHandle
+from temporalio.common import WorkflowIDConflictPolicy
+
+from harness.python.feature import Runner, register_feature
+
+SIGNAL_VALUE = "test-signal-value"
+
+_setup: dict = {}
+
+
+@workflow.defn
+class TargetWorkflow:
+    def __init__(self) -> None:
+        self._signal_value: Optional[str] = None
+
+    @workflow.run
+    async def run(self) -> str:
+        await workflow.wait_condition(lambda: self._signal_value is not None)
+        assert self._signal_value is not None
+        return self._signal_value
+
+    @workflow.signal
+    def my_signal(self, value: str) -> None:
+        self._signal_value = value
+
+
+@workflow.defn
+class CallerWorkflow:
+    @workflow.run
+    async def run(self, target_id: str) -> str:
+        # WORKFLOW_ID_CONFLICT_POLICY_FAIL is rejected by the server when
+        # scheduling the Nexus operation (signal-with-required-start is not a
+        # supported operation). This surfaces as a workflow task failure rather
+        # than a catchable error, so this run never completes; the client asserts
+        # on the workflow task failure in history.
+        handle = await workflow.signal_with_start_workflow(
+            TargetWorkflow.run,
+            id=target_id,
+            task_queue=workflow.info().task_queue,
+            signal=TargetWorkflow.my_signal,
+            signal_args=SIGNAL_VALUE,
+            id_conflict_policy=WorkflowIDConflictPolicy.FAIL,
+        )
+        return handle.run_id or ""
+
+
+async def start(runner: Runner) -> WorkflowHandle:
+    target_id = f"signalwithstart-id-conflict-fail-target-{uuid.uuid4()}"
+    _setup["target_id"] = target_id
+
+    # Start a running target to mirror the server test setup.
+    target = await runner.client.start_workflow(
+        TargetWorkflow.run,
+        id=target_id,
+        task_queue=runner.task_queue,
+        execution_timeout=timedelta(minutes=1),
+    )
+    _setup["target_run_id"] = target.result_run_id
+
+    return await runner.client.start_workflow(
+        CallerWorkflow.run,
+        target_id,
+        id=f"signalwithstart-id-conflict-fail-caller-{uuid.uuid4()}",
+        task_queue=runner.task_queue,
+        execution_timeout=timedelta(minutes=1),
+    )
+
+
+async def check_result(runner: Runner, handle: WorkflowHandle) -> None:
+    # The caller never completes; wait for the workflow task failure that rejects
+    # the unsupported conflict policy.
+    event = await runner.wait_for_event(
+        handle,
+        lambda e: e.event_type == EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED,
+        timeout=30.0,
+    )
+    message = event.workflow_task_failed_event_attributes.failure.message
+    assert "not supported" in message.lower(), (
+        f"expected 'not supported' rejection, got: {message}"
+    )
+
+    # Cleanup the caller (stuck failing its workflow task) and the target.
+    for wf_id, run_id in (
+        (handle.id, None),
+        (_setup["target_id"], _setup["target_run_id"]),
+    ):
+        try:
+            await runner.client.get_workflow_handle(wf_id, run_id=run_id).terminate(
+                "cleanup"
+            )
+        except Exception:
+            pass
+
+
+register_feature(
+    workflows=[CallerWorkflow, TargetWorkflow],
+    start=start,
+    check_result=check_result,
+)

features/signalwithstart/id_conflict_terminate_existing/README.md

@@ -0,0 +1,13 @@
+# Signal-with-start from a workflow: TERMINATE_EXISTING conflict policy
+
+A target workflow is already running. A workflow calls
+`workflow.signal_with_start_workflow(...)` with
+`id_conflict_policy=TERMINATE_EXISTING`. The running run is terminated and a new
+run is started.
+
+Verifies: the returned run id differs from the original, and the original run's
+status is TERMINATED.
+
+## Server requirements
+Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and
+`history.enableChasm=true`.

features/signalwithstart/id_conflict_terminate_existing/feature.py

@@ -0,0 +1,105 @@
+import uuid
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Optional
+
+from temporalio import workflow
+from temporalio.client import WorkflowExecutionStatus, WorkflowHandle
+from temporalio.common import WorkflowIDConflictPolicy
+
+from harness.python.feature import Runner, register_feature
+
+SIGNAL_VALUE = "test-signal-value"
+
+_setup: dict = {}
+
+
+@dataclass
+class SwsResult:
+    workflow_id: str
+    run_id: str
+
+
+@workflow.defn
+class TargetWorkflow:
+    def __init__(self) -> None:
+        self._signal_value: Optional[str] = None
+
+    @workflow.run
+    async def run(self) -> str:
+        await workflow.wait_condition(lambda: self._signal_value is not None)
+        assert self._signal_value is not None
+        return self._signal_value
+
+    @workflow.signal
+    def my_signal(self, value: str) -> None:
+        self._signal_value = value
+
+
+@workflow.defn
+class CallerWorkflow:
+    @workflow.run
+    async def run(self, target_id: str) -> SwsResult:
+        handle = await workflow.signal_with_start_workflow(
+            TargetWorkflow.run,
+            id=target_id,
+            task_queue=workflow.info().task_queue,
+            signal=TargetWorkflow.my_signal,
+            signal_args=SIGNAL_VALUE,
+            id_conflict_policy=WorkflowIDConflictPolicy.TERMINATE_EXISTING,
+        )
+        return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "")
+
+
+async def start(runner: Runner) -> WorkflowHandle:
+    target_id = f"signalwithstart-id-conflict-terminate-target-{uuid.uuid4()}"
+
+    # Start the target and leave it running. TERMINATE_EXISTING should terminate
+    # this run and start a new one.
+    target = await runner.client.start_workflow(
+        TargetWorkflow.run,
+        id=target_id,
+        task_queue=runner.task_queue,
+        execution_timeout=timedelta(minutes=1),
+    )
+    _setup["original_run_id"] = target.result_run_id
+
+    return await runner.client.start_workflow(
+        CallerWorkflow.run,
+        target_id,
+        id=f"signalwithstart-id-conflict-terminate-caller-{uuid.uuid4()}",
+        task_queue=runner.task_queue,
+        execution_timeout=timedelta(minutes=1),
+    )
+
+
+async def check_result(runner: Runner, handle: WorkflowHandle) -> None:
+    result: SwsResult = await handle.result()
+    assert result.run_id, "expected a non-empty run id"
+    assert result.run_id != _setup["original_run_id"], (
+        "expected a new run id with TERMINATE_EXISTING"
+    )
+
+    # The original run must have been terminated.
+    original = runner.client.get_workflow_handle(
+        result.workflow_id, run_id=_setup["original_run_id"]
+    )
+    desc = await original.describe()
+    assert desc.status == WorkflowExecutionStatus.TERMINATED, (
+        f"expected original run TERMINATED, got {desc.status}"
+    )
+
+    # Cleanup the freshly-started run.
+    try:
+        await runner.client.get_workflow_handle(
+            result.workflow_id, run_id=result.run_id
+        ).terminate("cleanup")
+    except Exception:
+        pass
+
+
+register_feature(
+    workflows=[CallerWorkflow, TargetWorkflow],
+    start=start,
+    check_result=check_result,
+)

features/signalwithstart/id_conflict_use_existing/README.md

@@ -0,0 +1,12 @@
+# Signal-with-start from a workflow: USE_EXISTING conflict policy
+
+A target workflow is already running. A workflow calls
+`workflow.signal_with_start_workflow(...)` with
+`id_conflict_policy=USE_EXISTING`. The existing run is signaled and its run id is
+returned; no new run is started.
+
+Verifies: the returned run id equals the already-running target's run id.
+
+## Server requirements
+Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and
+`history.enableChasm=true`.