diff --git a/cmd/run.go b/cmd/run.go
index 7bd3898a..f19eeb09 100644
--- a/cmd/run.go
+++ b/cmd/run.go
@@ -43,6 +43,7 @@ const (
 	summaryListenAddr               = "127.0.0.1:0"
 	FeaturePassed                   = "PASSED"
 	featureNamespaceCapabilitiesEnv = "FEATURE_NAMESPACE_CAPABILITIES"
+	devServerDownloadVersion        = "v1.8.3-server-1.32.0-162.0"
 )
 
 func runCmd() *cli.Command {
@@ -433,9 +434,10 @@ func (r *Runner) runBatch(ctx context.Context, batch runBatch) error {
 			return err
 		}
 		server, err := testsuite.StartDevServer(ctx, testsuite.DevServerOptions{
-			LogLevel:      "error",
-			ClientOptions: &client.Options{Namespace: config.Namespace},
-			ExtraArgs:     dynamicConfigArgs,
+			LogLevel:       "error",
+			ClientOptions:  &client.Options{Namespace: config.Namespace},
+			ExtraArgs:      dynamicConfigArgs,
+			CachedDownload: testsuite.CachedDownload{Version: devServerDownloadVersion},
 		})
 		if err != nil {
 			return fmt.Errorf("failed starting devserver: %w", err)
diff --git a/features/system_nexus/signal_with_start/README.md b/features/system_nexus/signal_with_start/README.md
new file mode 100644
index 00000000..036e2ceb
--- /dev/null
+++ b/features/system_nexus/signal_with_start/README.md
@@ -0,0 +1,17 @@
+# Workflow-side signal with start
+
+A workflow can signal another workflow and start it atomically when it is not
+already running. This experimental API is implemented as a System Nexus
+operation.
+
+The basic execution starts a target workflow with its first signal, then invokes
+the operation again for the same workflow ID. It verifies that the target
+receives the start input and both signals, proving that the second operation
+used the existing execution.
+
+A separate execution uses a context-aware payload converter and codec. It
+verifies that Signal-with-Start inner payloads have the target workflow
+serialization context and pass through the codec; the codec rejects the outer
+System Nexus envelope if it is passed to it. Python also performs this execution
+with an in-memory external-storage driver and verifies that inner payloads are
+externalized while the outer envelope is not.
diff --git a/features/system_nexus/signal_with_start/config.json b/features/system_nexus/signal_with_start/config.json
new file mode 100644
index 00000000..3e484711
--- /dev/null
+++ b/features/system_nexus/signal_with_start/config.json
@@ -0,0 +1,10 @@
+{
+  "runVariants": [
+    {
+      "name": "signal-with-start-from-workflow-enabled",
+      "dynamicConfig": {
+        "history.enableSignalWithStartFromWorkflow": true
+      }
+    }
+  ]
+}
diff --git a/features/system_nexus/signal_with_start/feature.cs b/features/system_nexus/signal_with_start/feature.cs
new file mode 100644
index 00000000..90c1cb0d
--- /dev/null
+++ b/features/system_nexus/signal_with_start/feature.cs
@@ -0,0 +1,273 @@
+namespace system_nexus.signal_with_start;
+
+using System.Collections.Concurrent;
+using System.Text.Json;
+using Google.Protobuf;
+using Temporalio.Api.Common.V1;
+using Temporalio.Api.Enums.V1;
+using Temporalio.Client;
+using Temporalio.Converters;
+using Temporalio.Features.Harness;
+using Temporalio.Worker;
+using Temporalio.Workflows;
+
+class Feature : IFeature
+{
+    private static readonly ConcurrentQueue<SerializationRecord> SerializationRecords = new();
+
+    private record ContextValue(string Label);
+
+    private record SerializationRecord(string Stage, string Label, string WorkflowId);
+
+    [Workflow]
+    class TargetWorkflow
+    {
+        private readonly List<string> signals = new();
+
+        [WorkflowRun]
+        public async Task<IReadOnlyCollection<string>> RunAsync(string value)
+        {
+            await Workflow.WaitConditionAsync(() => signals.Count == 2);
+            return [$"started: {value}", .. signals];
+        }
+
+        [WorkflowSignal]
+        public Task AddAsync(string value)
+        {
+            signals.Add($"signal: {value}");
+            return Task.CompletedTask;
+        }
+    }
+
+    [Workflow]
+    class CallerWorkflow
+    {
+        [WorkflowRun]
+        public async Task<string> RunAsync(string targetId, string taskQueue)
+        {
+            await Workflow.SignalWithStartWorkflowAsync(
+                (TargetWorkflow workflow) => workflow.RunAsync("start-value"),
+                workflow => workflow.AddAsync("signal-one"),
+                new(targetId, taskQueue)
+                {
+                    IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting,
+                });
+            await Workflow.SignalWithStartWorkflowAsync(
+                (TargetWorkflow workflow) => workflow.RunAsync("unused-start-value"),
+                workflow => workflow.AddAsync("signal-two"),
+                new(targetId, taskQueue)
+                {
+                    IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting,
+                });
+            return targetId;
+        }
+    }
+
+    // This execution deliberately uses a distinct workflow pair and payload values. It verifies
+    // the converter and codec are applied to the inner Signal-with-Start request payloads with
+    // the target workflow's serialization context.
+    [Workflow]
+    class ContextTargetWorkflow
+    {
+        private readonly List<ContextValue> signals = new();
+
+        [WorkflowRun]
+        public async Task<IReadOnlyCollection<string>> RunAsync(ContextValue value)
+        {
+            await Workflow.WaitConditionAsync(() => signals.Count == 2);
+            return [value.Label, .. signals.Select(signal => signal.Label)];
+        }
+
+        [WorkflowSignal]
+        public Task AddAsync(ContextValue value)
+        {
+            signals.Add(value);
+            return Task.CompletedTask;
+        }
+    }
+
+    [Workflow]
+    class ContextCallerWorkflow
+    {
+        [WorkflowRun]
+        public async Task<string> RunAsync(string targetId, string taskQueue)
+        {
+            await Workflow.SignalWithStartWorkflowAsync(
+                (ContextTargetWorkflow workflow) => workflow.RunAsync(new("context-start")),
+                workflow => workflow.AddAsync(new("context-signal-one")),
+                new(targetId, taskQueue)
+                {
+                    IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting,
+                });
+            await Workflow.SignalWithStartWorkflowAsync(
+                (ContextTargetWorkflow workflow) => workflow.RunAsync(new("unused-context-start")),
+                workflow => workflow.AddAsync(new("context-signal-two")),
+                new(targetId, taskQueue)
+                {
+                    IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting,
+                });
+            return targetId;
+        }
+    }
+
+    class ContextJsonPlainConverter : JsonPlainConverter, IWithSerializationContext<IEncodingConverter>
+    {
+        private readonly string? workflowId;
+
+        public ContextJsonPlainConverter(string? workflowId = null)
+            : base(new()) => this.workflowId = workflowId;
+
+        public IEncodingConverter WithSerializationContext(ISerializationContext context) =>
+            new ContextJsonPlainConverter(WorkflowIdFor(context));
+
+        public override bool TryToPayload(object? value, out Payload? payload)
+        {
+            var converted = base.TryToPayload(value, out payload);
+            if (converted && workflowId != null && value is ContextValue contextValue)
+            {
+                SerializationRecords.Enqueue(new("converter-encode", contextValue.Label, workflowId));
+            }
+            return converted;
+        }
+
+        public override object? ToValue(Payload payload, Type type)
+        {
+            var value = base.ToValue(payload, type);
+            if (workflowId != null && value is ContextValue contextValue)
+            {
+                SerializationRecords.Enqueue(new("converter-decode", contextValue.Label, workflowId));
+            }
+            return value;
+        }
+    }
+
+    class ContextPayloadCodec : IPayloadCodec, IWithSerializationContext<IPayloadCodec>
+    {
+        private readonly string? workflowId;
+
+        public ContextPayloadCodec(string? workflowId = null) => this.workflowId = workflowId;
+
+        public IPayloadCodec WithSerializationContext(ISerializationContext context) =>
+            new ContextPayloadCodec(WorkflowIdFor(context));
+
+        public Task<IReadOnlyCollection<Payload>> EncodeAsync(IReadOnlyCollection<Payload> payloads)
+        {
+            Record("codec-encode", payloads);
+            return Task.FromResult(payloads);
+        }
+
+        public Task<IReadOnlyCollection<Payload>> DecodeAsync(IReadOnlyCollection<Payload> payloads)
+        {
+            Record("codec-decode", payloads);
+            return Task.FromResult(payloads);
+        }
+
+        private void Record(string stage, IReadOnlyCollection<Payload> payloads)
+        {
+            if (workflowId == null)
+            {
+                return;
+            }
+            foreach (var payload in payloads)
+            {
+                // System Nexus envelopes must be traversed, not passed to application codecs.
+                if (payload.Metadata.TryGetValue("messageType", out var messageType) &&
+                    messageType.ToStringUtf8().StartsWith("temporal.api.workflowservice.v1.SignalWithStart"))
+                {
+                    throw new InvalidOperationException("Codec received a Signal-with-Start envelope");
+                }
+                if (!payload.Metadata.TryGetValue("encoding", out var encoding) ||
+                    encoding.ToStringUtf8() != "json/plain")
+                {
+                    continue;
+                }
+                try
+                {
+                    var value = JsonSerializer.Deserialize<ContextValue>(payload.Data.Span);
+                    if (value != null && value.Label.StartsWith("context-"))
+                    {
+                        SerializationRecords.Enqueue(new(stage, value.Label, workflowId));
+                    }
+                }
+                catch (JsonException)
+                {
+                    // This is an unrelated json/plain payload.
+                }
+            }
+        }
+    }
+
+    private static string WorkflowIdFor(ISerializationContext context) =>
+        ((ISerializationContext.IHasWorkflow)context).WorkflowId ??
+        throw new InvalidOperationException("Expected a workflow serialization context");
+
+    public void ConfigureClient(Runner runner, TemporalClientConnectOptions options)
+    {
+        var defaultConverters = ((DefaultPayloadConverter)DataConverter.Default.PayloadConverter)
+            .EncodingConverters
+            .Where(converter => converter is not JsonPlainConverter);
+        options.DataConverter = DataConverter.Default with
+        {
+            PayloadConverter = new DefaultPayloadConverter(
+                [new ContextJsonPlainConverter(), .. defaultConverters]),
+            PayloadCodec = new ContextPayloadCodec(),
+        };
+    }
+
+    public void ConfigureWorker(Runner runner, TemporalWorkerOptions options) =>
+        options
+            .AddWorkflow<CallerWorkflow>()
+            .AddWorkflow<TargetWorkflow>()
+            .AddWorkflow<ContextCallerWorkflow>()
+            .AddWorkflow<ContextTargetWorkflow>();
+
+    public async Task<WorkflowHandle?> ExecuteAsync(Runner runner)
+    {
+        var targetId = $"{runner.PreparedFeature.Dir}-target";
+        return await runner.Client.StartWorkflowAsync(
+            (CallerWorkflow workflow) => workflow.RunAsync(targetId, runner.WorkerOptions.TaskQueue!),
+            runner.NewWorkflowOptions());
+    }
+
+    public async Task CheckResultAsync(Runner runner, WorkflowHandle handle)
+    {
+        var targetId = await handle.GetResultAsync<string>();
+        var target = runner.Client.GetWorkflowHandle<TargetWorkflow, IReadOnlyCollection<string>>(targetId);
+        Assert.Equal(
+            new[] { "started: start-value", "signal: signal-one", "signal: signal-two" },
+            await target.GetResultAsync());
+
+        while (SerializationRecords.TryDequeue(out _))
+        {
+        }
+        var contextTargetId = $"{runner.PreparedFeature.Dir}-context-target";
+        var contextCaller = await runner.Client.StartWorkflowAsync(
+            (ContextCallerWorkflow workflow) =>
+                workflow.RunAsync(contextTargetId, runner.WorkerOptions.TaskQueue!),
+            runner.NewWorkflowOptions());
+        Assert.Equal(contextTargetId, await contextCaller.GetResultAsync<string>());
+        var contextTarget = runner.Client.GetWorkflowHandle<ContextTargetWorkflow, IReadOnlyCollection<string>>(
+            contextTargetId);
+        Assert.Equal(
+            new[] { "context-start", "context-signal-one", "context-signal-two" },
+            await contextTarget.GetResultAsync());
+
+        var records = SerializationRecords.ToArray();
+        foreach (var stage in new[]
+        {
+            "converter-encode",
+            "converter-decode",
+            "codec-encode",
+            "codec-decode",
+        })
+        {
+            var stageRecords = records.Where(record => record.Stage == stage).ToArray();
+            Assert.True(stageRecords.Length >= 3, $"Expected three {stage} records");
+            Assert.All(stageRecords, record => Assert.Equal(contextTargetId, record.WorkflowId));
+            foreach (var label in new[] { "context-start", "context-signal-one", "context-signal-two" })
+            {
+                Assert.Contains(stageRecords, record => record.Label == label);
+            }
+        }
+    }
+}
diff --git a/features/system_nexus/signal_with_start/feature.py b/features/system_nexus/signal_with_start/feature.py
new file mode 100644
index 00000000..244681d6
--- /dev/null
+++ b/features/system_nexus/signal_with_start/feature.py
@@ -0,0 +1,324 @@
+import asyncio
+import json
+from collections.abc import Sequence
+from datetime import timedelta
+from typing import Any
+
+from temporalio import workflow
+from temporalio.api.common.v1 import Payload
+from temporalio.client import WorkflowHandle
+from temporalio.common import WorkflowIDConflictPolicy
+from temporalio.converter import (
+    CompositePayloadConverter,
+    DataConverter,
+    DefaultPayloadConverter,
+    EncodingPayloadConverter,
+    ExternalStorage,
+    JSONPlainPayloadConverter,
+    PayloadCodec,
+    SerializationContext,
+    StorageDriver,
+    StorageDriverClaim,
+    StorageDriverRetrieveContext,
+    StorageDriverStoreContext,
+    WithSerializationContext,
+    WorkflowSerializationContext,
+)
+from temporalio.exceptions import ApplicationError
+
+from harness.python.feature import Runner, register_feature
+
+with workflow.unsafe.imports_passed_through():
+    from features.system_nexus.signal_with_start.support import (
+        ContextValue,
+        serialization_records,
+    )
+
+SYSTEM_ENVELOPE = "__temporal_system_payload"
+CONTEXT_MARKER = "test-signal-with-start-context"
+CODEC_MARKER = "test-signal-with-start-codec"
+
+
+class ContextPayloadConverter(EncodingPayloadConverter, WithSerializationContext):
+    def __init__(
+        self,
+        records: list[tuple[str, str, str]],
+        context: SerializationContext | None = None,
+    ) -> None:
+        self.records = records
+        self.context = context
+
+    @property
+    def encoding(self) -> str:
+        return "test-signal-with-start-context"
+
+    def with_context(self, context: SerializationContext) -> "ContextPayloadConverter":
+        return ContextPayloadConverter(self.records, context)
+
+    def to_payload(self, value: Any) -> Payload | None:
+        if not isinstance(value, ContextValue):
+            return None
+        assert isinstance(self.context, WorkflowSerializationContext)
+        self.records.append(("converter-encode", value.label, self.context.workflow_id))
+        payload = JSONPlainPayloadConverter().to_payload(value)
+        assert payload is not None
+        payload.metadata["encoding"] = self.encoding.encode()
+        payload.metadata[CONTEXT_MARKER] = self.context.workflow_id.encode()
+        return payload
+
+    def from_payload(self, payload: Payload, type_hint: type | None = None) -> Any:
+        assert isinstance(self.context, WorkflowSerializationContext)
+        expected_id = payload.metadata[CONTEXT_MARKER].decode()
+        assert self.context.workflow_id == expected_id
+        value = JSONPlainPayloadConverter().from_payload(payload, ContextValue)
+        assert isinstance(value, ContextValue)
+        self.records.append(("converter-decode", value.label, self.context.workflow_id))
+        return value
+
+
+class ContextPayloadConverterSet(CompositePayloadConverter):
+    def __init__(self) -> None:
+        super().__init__(
+            ContextPayloadConverter(serialization_records),
+            *DefaultPayloadConverter.default_encoding_payload_converters,
+        )
+
+
+class ContextPayloadCodec(PayloadCodec, WithSerializationContext):
+    def __init__(
+        self,
+        records: list[tuple[str, str, str]],
+        context: SerializationContext | None = None,
+    ) -> None:
+        self.records = records
+        self.context = context
+
+    def with_context(self, context: SerializationContext) -> "ContextPayloadCodec":
+        return ContextPayloadCodec(self.records, context)
+
+    async def encode(self, payloads: Sequence[Payload]) -> list[Payload]:
+        return self._visit("codec-encode", payloads)
+
+    async def decode(self, payloads: Sequence[Payload]) -> list[Payload]:
+        return self._visit("codec-decode", payloads)
+
+    def _visit(self, operation: str, payloads: Sequence[Payload]) -> list[Payload]:
+        for payload in payloads:
+            assert SYSTEM_ENVELOPE not in payload.metadata
+            if CONTEXT_MARKER in payload.metadata:
+                assert isinstance(self.context, WorkflowSerializationContext)
+                expected_id = payload.metadata[CONTEXT_MARKER].decode()
+                assert self.context.workflow_id == expected_id
+                value = JSONPlainPayloadConverter().from_payload(payload, ContextValue)
+                assert isinstance(value, ContextValue)
+                self.records.append((operation, value.label, expected_id))
+                payload.metadata[CODEC_MARKER] = expected_id.encode()
+        return list(payloads)
+
+
+class RecordingStorageDriver(StorageDriver):
+    def __init__(self) -> None:
+        self.storage: dict[str, bytes] = {}
+        self.stored_payloads: list[Payload] = []
+
+    def name(self) -> str:
+        return "signal-with-start-test-storage"
+
+    async def store(
+        self, context: StorageDriverStoreContext, payloads: Sequence[Payload]
+    ) -> list[StorageDriverClaim]:
+        _ = context
+        entries: list[tuple[str, bytes]] = []
+        for payload in payloads:
+            assert SYSTEM_ENVELOPE not in payload.metadata
+            key = f"payload-{len(self.storage) + len(entries)}"
+            serialized = payload.SerializeToString()
+            entries.append((key, serialized))
+            snapshot = Payload()
+            snapshot.ParseFromString(serialized)
+            self.stored_payloads.append(snapshot)
+        self.storage.update(entries)
+        await asyncio.sleep(0)
+        return [StorageDriverClaim(claim_data={"key": key}) for key, _ in entries]
+
+    async def retrieve(
+        self,
+        context: StorageDriverRetrieveContext,
+        claims: Sequence[StorageDriverClaim],
+    ) -> list[Payload]:
+        _ = context
+        payloads: list[Payload] = []
+        for claim in claims:
+            serialized = self.storage.get(claim.claim_data["key"])
+            if serialized is None:
+                raise ApplicationError("stored payload not found", non_retryable=True)
+            payload = Payload()
+            payload.ParseFromString(serialized)
+            payloads.append(payload)
+        return payloads
+
+
+storage_driver = RecordingStorageDriver()
+
+
+@workflow.defn
+class TargetWorkflow:
+    def __init__(self) -> None:
+        self.signals: list[str] = []
+
+    @workflow.run
+    async def run(self, value: str) -> list[str]:
+        await workflow.wait_condition(lambda: len(self.signals) == 2)
+        return [f"started: {value}", *self.signals]
+
+    @workflow.signal
+    def add(self, value: str) -> None:
+        self.signals.append(f"signal: {value}")
+
+
+@workflow.defn
+class CallerWorkflow:
+    @workflow.run
+    async def run(self, target_id: str, task_queue: str) -> str:
+        await workflow.signal_with_start_workflow(
+            TargetWorkflow.run,
+            "start-value",
+            id=target_id,
+            task_queue=task_queue,
+            signal=TargetWorkflow.add,
+            signal_args="signal-one",
+            id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
+        )
+        await workflow.signal_with_start_workflow(
+            TargetWorkflow.run,
+            "unused-start-value",
+            id=target_id,
+            task_queue=task_queue,
+            signal=TargetWorkflow.add,
+            signal_args="signal-two",
+            id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
+        )
+        return target_id
+
+
+@workflow.defn
+class ContextTargetWorkflow:
+    def __init__(self) -> None:
+        self.signals: list[ContextValue] = []
+
+    @workflow.run
+    async def run(self, value: ContextValue) -> list[str]:
+        await workflow.wait_condition(lambda: len(self.signals) == 2)
+        return [value.label, *(signal.label for signal in self.signals)]
+
+    @workflow.signal
+    def add(self, value: ContextValue) -> None:
+        self.signals.append(value)
+
+
+@workflow.defn
+class ContextCallerWorkflow:
+    @workflow.run
+    async def run(self, target_id: str, task_queue: str) -> str:
+        await workflow.signal_with_start_workflow(
+            ContextTargetWorkflow.run,
+            ContextValue("start-value"),
+            id=target_id,
+            task_queue=task_queue,
+            signal=ContextTargetWorkflow.add,
+            signal_args=ContextValue("signal-one"),
+            id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
+        )
+        await workflow.signal_with_start_workflow(
+            ContextTargetWorkflow.run,
+            ContextValue("unused-start-value"),
+            id=target_id,
+            task_queue=task_queue,
+            signal=ContextTargetWorkflow.add,
+            signal_args=ContextValue("signal-two"),
+            id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
+        )
+        return target_id
+
+
+async def start(runner: Runner) -> WorkflowHandle:
+    return await runner.client.start_workflow(
+        CallerWorkflow.run,
+        args=[f"{runner.feature.rel_dir}-target", runner.task_queue],
+        id=f"{runner.feature.rel_dir}-caller",
+        task_queue=runner.task_queue,
+        execution_timeout=timedelta(minutes=1),
+    )
+
+
+async def check_result(runner: Runner, handle: WorkflowHandle) -> None:
+    target_id = await handle.result()
+    target = runner.client.get_workflow_handle(target_id)
+    assert await target.result() == [
+        "started: start-value",
+        "signal: signal-one",
+        "signal: signal-two",
+    ]
+
+    serialization_records.clear()
+    storage_driver.storage.clear()
+    storage_driver.stored_payloads.clear()
+    target_id = f"{runner.feature.rel_dir}-context-target"
+    context_handle = await runner.client.start_workflow(
+        ContextCallerWorkflow.run,
+        args=[target_id, runner.task_queue],
+        id=f"{runner.feature.rel_dir}-context-caller",
+        task_queue=runner.task_queue,
+        execution_timeout=timedelta(minutes=1),
+    )
+    assert await context_handle.result() == target_id
+    target = runner.client.get_workflow_handle(target_id)
+    assert await target.result() == ["start-value", "signal-one", "signal-two"]
+
+    expected_labels = {"start-value", "signal-one", "signal-two"}
+    for operation in (
+        "converter-encode",
+        "converter-decode",
+        "codec-encode",
+        "codec-decode",
+    ):
+        records = [record for record in serialization_records if record[0] == operation]
+        assert expected_labels.issubset({record[1] for record in records}), records
+        assert all(record[2] == target_id for record in records)
+
+    stored_context_values = {
+        json.loads(payload.data)["label"]
+        for payload in storage_driver.stored_payloads
+        if CONTEXT_MARKER in payload.metadata
+    }
+    assert expected_labels.issubset(stored_context_values)
+    assert all(
+        payload.metadata[CONTEXT_MARKER] == target_id.encode()
+        for payload in storage_driver.stored_payloads
+        if CONTEXT_MARKER in payload.metadata
+    )
+    assert all(
+        payload.metadata.get(CODEC_MARKER) == target_id.encode()
+        for payload in storage_driver.stored_payloads
+        if CONTEXT_MARKER in payload.metadata
+    )
+
+
+register_feature(
+    workflows=[
+        CallerWorkflow,
+        ContextCallerWorkflow,
+        ContextTargetWorkflow,
+        TargetWorkflow,
+    ],
+    start=start,
+    check_result=check_result,
+    data_converter=DataConverter(
+        payload_converter_class=ContextPayloadConverterSet,
+        payload_codec=ContextPayloadCodec(serialization_records),
+        external_storage=ExternalStorage(
+            drivers=[storage_driver],
+            payload_size_threshold=1,
+        ),
+    ),
+)
diff --git a/features/system_nexus/signal_with_start/support.py b/features/system_nexus/signal_with_start/support.py
new file mode 100644
index 00000000..67d367cd
--- /dev/null
+++ b/features/system_nexus/signal_with_start/support.py
@@ -0,0 +1,9 @@
+import dataclasses
+
+
+@dataclasses.dataclass
+class ContextValue:
+    label: str
+
+
+serialization_records: list[tuple[str, str, str]] = []