feat: decode an unknown proto enum number as a marked fallback case

GUIDE.md

@@ -825,7 +825,9 @@ by the target SDK or generated operation helpers.
 An enum annotated with `@nexus.proto` likewise remains a generated public enum
 unless a type replacement selects an external model. Fields use the authored
 enum's name and members, even when its name differs from the protobuf enum's.
-Converters bridge the model and protobuf enum by number.
+Converters bridge the model and protobuf enum by number. A case with
+[`@nexus.fallback`](#nexusfallback) takes a protobuf number that the enum
+doesn't know.
 
 ### Transfer-Type Conversion
 
@@ -1791,6 +1793,33 @@ A default cannot be combined with `@nexus.source`.
 
 ---
 
+### @nexus.fallback
+
+**Placement:** Case of an enum with `@nexus.proto` **Syntax:** `@nexus.fallback`
+
+A protobuf enum is open, so a newer peer can send a number that the generated
+enum doesn't know. When a protobuf value converts to the model, such a number
+becomes the marked case. An enum has at most one fallback case, and the
+directive takes no arguments.
+
+```wit
+/// @nexus.proto "temporal.api.enums.v1.StreamStatus"
+enum stream-status {
+  /// @nexus.fallback
+  stream-status-unspecified,
+  stream-status-open,
+  stream-status-closed,
+  stream-status-expired,
+}
+```
+
+Python, TypeScript, and .NET decode an unknown number as the case. Without the
+directive, Python raises `ValueError`, and TypeScript and .NET keep the number.
+Go and Java ignore the directive. nexgen refuses it on an enum without
+`@nexus.proto`, and anywhere other than an enum case.
+
+---
+
 ### @nexus.experimental
 
 **Placement:** Interface, record, variant, or operation (function) **Syntax:**

advanced/samples/dotnet/tests/StreamServiceChecks.cs

@@ -5,6 +5,7 @@
 using Xunit;
 using static Nexgen.DotNetExamples.Tests.TransferTypeTestSupport;
 using StreamServiceExample = Streams::Temporalio.Workflows;
+using WireDescribeStreamResponse = Streams::Temporalio.Api.StreamService.V1.DescribeStreamResponse;
 using WireReadRecordsResponse = Streams::Temporalio.Api.StreamService.V1.ReadRecordsResponse;
 using WireStreamRecord = Streams::Temporalio.Api.Stream.V1.StreamRecord;
 
@@ -71,5 +72,20 @@ public void ReadItemsRoundTripAsTheirRecords()
                 close.Value.Status);
             Assert.Equal(3, decoded.Records[1].Offset);
         }
+
+        // A newer server can send a Stream status that this code doesn't know.
+        [Fact]
+        public void UnknownEnumNumberDecodesAsTheFallbackCase()
+        {
+            var converter = CreateSdkConverter(typeof(StreamServiceExample.DescribeStreamResponse));
+            var wire = new WireDescribeStreamResponse
+            {
+                Status = (Streams::Temporalio.Api.Enums.V1.StreamStatus)9,
+            };
+
+            var decoded = Assert.IsType<StreamServiceExample.DescribeStreamResponse>(
+                converter.FromTransferType(wire));
+            Assert.Equal(StreamServiceExample.StreamStatus.StreamStatusUnspecified, decoded.Status);
+        }
     }
 }

advanced/samples/dotnet/wit/stream-service/Models.cs

@@ -522,7 +522,7 @@ internal static DescribeStreamResponse FromTransferType(Temporalio.Api.StreamSer
         {
             return new DescribeStreamResponse()
             {
-                Status = (int)wire.Status == 0 ? null : (StreamStatus)wire.Status,
+                Status = (int)wire.Status == 0 ? null : (System.Enum.IsDefined(typeof(StreamStatus), (int)wire.Status) ? (StreamStatus)wire.Status : StreamStatus.StreamStatusUnspecified),
                 LastOffset = wire.LastOffset == 0 ? null : wire.LastOffset,
                 Retention = wire.Retention == null ? null : Nexgen.Support.ProtoExtensions.FromDurationProto(wire.Retention),
             };
@@ -584,7 +584,7 @@ internal static Execution FromTransferType(Temporalio.Api.Common.V1.Execution wi
         {
             return new Execution()
             {
-                Type = (int)wire.Type == 0 ? null : (ExecutionType)wire.Type,
+                Type = (int)wire.Type == 0 ? null : (System.Enum.IsDefined(typeof(ExecutionType), (int)wire.Type) ? (ExecutionType)wire.Type : ExecutionType.ExecutionTypeUnspecified),
                 BusinessId = string.IsNullOrEmpty(wire.BusinessId) ? null : wire.BusinessId,
                 RunId = string.IsNullOrEmpty(wire.RunId) ? null : wire.RunId,
             };
@@ -655,7 +655,7 @@ internal static Producer FromTransferType(Temporalio.Api.Stream.V1.Producer wire
         {
             return new Producer(wire.Id)
             {
-                Kind = (int)wire.Kind == 0 ? null : (StreamProducerKind)wire.Kind,
+                Kind = (int)wire.Kind == 0 ? null : (System.Enum.IsDefined(typeof(StreamProducerKind), (int)wire.Kind) ? (StreamProducerKind)wire.Kind : StreamProducerKind.StreamProducerKindUnspecified),
                 Epoch = wire.Epoch == null ? null : ProducerEpoch.FromTransferType(wire.Epoch),
                 RunId = string.IsNullOrEmpty(wire.RunId) ? null : wire.RunId,
             };
@@ -979,7 +979,7 @@ internal static StreamClose FromTransferType(Temporalio.Api.Stream.V1.StreamClos
                 Producer = wire.Producer == null ? null : Producer.FromTransferType(wire.Producer),
                 Result = wire.Result == null ? null : Nexgen.Support.ProtoExtensions.FromPayload(wire.Result),
                 Failure = wire.Failure == null ? null : Nexgen.Support.ProtoExtensions.FromFailureProto(wire.Failure),
-                Status = (int)wire.Status == 0 ? null : (StreamCloseStatus)wire.Status,
+                Status = (int)wire.Status == 0 ? null : (System.Enum.IsDefined(typeof(StreamCloseStatus), (int)wire.Status) ? (StreamCloseStatus)wire.Status : StreamCloseStatus.StreamCloseStatusUnspecified),
             };
         }
 
@@ -1266,7 +1266,7 @@ internal static Supersession FromTransferType(Temporalio.Api.Stream.V1.Supersess
         {
             return new Supersession(wire.ProducerId, System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.FailedTasks, item => item)))
             {
-                ProducerKind = (int)wire.ProducerKind == 0 ? null : (StreamProducerKind)wire.ProducerKind,
+                ProducerKind = (int)wire.ProducerKind == 0 ? null : (System.Enum.IsDefined(typeof(StreamProducerKind), (int)wire.ProducerKind) ? (StreamProducerKind)wire.ProducerKind : StreamProducerKind.StreamProducerKindUnspecified),
                 Epoch = wire.Epoch == null ? null : ProducerEpoch.FromTransferType(wire.Epoch),
                 PreviousCompletedTask = wire.PreviousCompletedTask == null ? null : WorkflowTaskEpoch.FromTransferType(wire.PreviousCompletedTask),
                 WithdrawRun = string.IsNullOrEmpty(wire.WithdrawRun) ? null : wire.WithdrawRun,

advanced/samples/inputs/stream-service.wit

@@ -43,6 +43,7 @@ interface stream-service {
 
   /// @nexus.proto "temporal.api.enums.v1.ExecutionType"
   enum execution-type {
+    /// @nexus.fallback
     execution-type-unspecified,
     execution-type-workflow,
     execution-type-activity,
@@ -61,6 +62,7 @@ interface stream-service {
 
   /// @nexus.proto "temporal.api.enums.v1.StreamProducerKind"
   enum stream-producer-kind {
+    /// @nexus.fallback
     stream-producer-kind-unspecified,
     stream-producer-kind-workflow,
     stream-producer-kind-local-activity,
@@ -140,6 +142,7 @@ interface stream-service {
   /// @nexus.doc "How a Stream closed. Only the owner's end on the server terminates or times out."
   /// @nexus.proto "temporal.api.enums.v1.StreamCloseStatus"
   enum stream-close-status {
+    /// @nexus.fallback
     stream-close-status-unspecified,
     stream-close-status-completed,
     stream-close-status-failed,
@@ -214,6 +217,7 @@ interface stream-service {
 
   /// @nexus.proto "temporal.api.enums.v1.StreamStatus"
   enum stream-status {
+    /// @nexus.fallback
     stream-status-unspecified,
     stream-status-open,
     stream-status-closed,
@@ -422,6 +426,7 @@ interface stream-service {
   ///   typescript-extends="ApplicationFailure" typescript-extends-import="@temporalio/common"
   ///   dotnet-extends="Temporalio.Exceptions.ApplicationFailureException"
   enum stream-error-reason {
+    /// @nexus.fallback
     stream-error-reason-unspecified,
     /// @nexus.error
     stream-error-reason-not-found,

advanced/samples/invalid/wit-parser/fallback-arguments/expected.toml

@@ -0,0 +1,8 @@
+descriptors = "temporal"
+
+[error]
+kind = "invalid-wit-directive"
+contains = [
+  "invalid WIT directive `@nexus.fallback` on type `models.alert-level` case `severity-unspecified`",
+  "does not take arguments",
+]

advanced/samples/invalid/wit-parser/fallback-arguments/input.wit

@@ -0,0 +1,14 @@
+package test:fallback-arguments;
+
+world system { export models; }
+
+interface models {
+  /// @nexus.proto "temporal.api.enums.v1.Severity"
+  enum alert-level {
+    /// @nexus.fallback "severity-low"
+    severity-unspecified,
+    severity-high,
+    severity-medium,
+    severity-low,
+  }
+}

advanced/samples/invalid/wit-parser/fallback-on-two-cases/expected.toml

@@ -0,0 +1,8 @@
+descriptors = "temporal"
+
+[error]
+kind = "invalid-wit-directive"
+contains = [
+  "invalid WIT directive `@nexus.fallback` on type `models.alert-level`",
+  "is on `severity-unspecified` and `severity-medium`, but an enum has one fallback case",
+]

advanced/samples/invalid/wit-parser/fallback-on-two-cases/input.wit

@@ -0,0 +1,15 @@
+package test:fallback-on-two-cases;
+
+world system { export models; }
+
+interface models {
+  /// @nexus.proto "temporal.api.enums.v1.Severity"
+  enum alert-level {
+    /// @nexus.fallback
+    severity-unspecified,
+    severity-high,
+    /// @nexus.fallback
+    severity-medium,
+    severity-low,
+  }
+}

advanced/samples/invalid/wit-parser/fallback-without-proto/expected.toml

@@ -0,0 +1,6 @@
+[error]
+kind = "invalid-wit-directive"
+contains = [
+  "invalid WIT directive `@nexus.fallback` on type `models.level`",
+  "is only supported on an enum with `@nexus.proto`",
+]

advanced/samples/invalid/wit-parser/fallback-without-proto/input.wit

@@ -0,0 +1,12 @@
+package test:fallback-without-proto;
+
+world system { export models; }
+
+interface models {
+  enum level {
+    /// @nexus.fallback
+    unknown,
+    low,
+    high,
+  }
+}

advanced/samples/python/tests/test_stream_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import pytest
+from temporalio.api.enums.v1 import stream_pb2
 from temporalio.api.stream.v1 import message_pb2
 from temporalio.api.streamservice.v1 import request_response_pb2
 from temporalio.converter import PayloadConverter
@@ -90,3 +91,26 @@ def test_a_request_without_its_stream_is_refused() -> None:
         ValueError, match="missing required field AppendRecordsRequest.stream"
     ):
         _ = converter.from_transfer_type(wire, AppendRecordsRequest)
+
+
+def test_an_unknown_enum_number_decodes_as_the_fallback_case() -> None:
+    converter = models._DescribeStreamResponseTransferTypeConverter()
+    # A newer server can send a Stream status that this code doesn't know.
+    wire = request_response_pb2.DescribeStreamResponse(
+        status=stream_pb2.StreamStatus.ValueType(9)
+    )
+
+    decoded = converter.from_transfer_type(wire, models.DescribeStreamResponse)
+
+    assert decoded.status is models.StreamStatus.StreamStatusUnspecified
+
+
+def test_an_unknown_number_of_an_enum_with_no_fallback_fails_the_message() -> None:
+    converter = models._ReadRecordsRequestTransferTypeConverter()
+    wire = request_response_pb2.ReadRecordsRequest(
+        stream=message_pb2.StreamRef(id="s-1"),
+        start=stream_pb2.StreamReadStart.ValueType(9),
+    )
+
+    with pytest.raises(ValueError, match="9 is not a valid StreamReadStart"):
+        _ = converter.from_transfer_type(wire, models.ReadRecordsRequest)

advanced/samples/python/wit/stream_service/models.py

@@ -27,13 +27,29 @@ class StreamProducerKind(enum.IntEnum):
     StreamProducerKindStandaloneActivity = 5
 
 
+def _stream_producer_kind_from_proto(value: int) -> StreamProducerKind:
+    try:
+        return StreamProducerKind(value)
+    except ValueError:
+        # A newer peer can send a case that this enum doesn't know.
+        return StreamProducerKind.StreamProducerKindUnspecified
+
+
 class ExecutionType(enum.IntEnum):
     ExecutionTypeUnspecified = 0
     ExecutionTypeWorkflow = 1
     ExecutionTypeActivity = 2
     ExecutionTypeNexusOperation = 3
 
 
+def _execution_type_from_proto(value: int) -> ExecutionType:
+    try:
+        return ExecutionType(value)
+    except ValueError:
+        # A newer peer can send a case that this enum doesn't know.
+        return ExecutionType.ExecutionTypeUnspecified
+
+
 class StreamReadStart(enum.IntEnum):
     StreamReadStartUnspecified = 0
     StreamReadStartBeginning = 1
@@ -49,13 +65,29 @@ class StreamCloseStatus(enum.IntEnum):
     StreamCloseStatusTimedOut = 5
 
 
+def _stream_close_status_from_proto(value: int) -> StreamCloseStatus:
+    try:
+        return StreamCloseStatus(value)
+    except ValueError:
+        # A newer peer can send a case that this enum doesn't know.
+        return StreamCloseStatus.StreamCloseStatusUnspecified
+
+
 class StreamStatus(enum.IntEnum):
     StreamStatusUnspecified = 0
     StreamStatusOpen = 1
     StreamStatusClosed = 2
     StreamStatusExpired = 3
 
 
+def _stream_status_from_proto(value: int) -> StreamStatus:
+    try:
+        return StreamStatus(value)
+    except ValueError:
+        # A newer peer can send a case that this enum doesn't know.
+        return StreamStatus.StreamStatusUnspecified
+
+
 @dataclasses.dataclass(slots=True)
 class CreateStreamRequest:
     """Create a Stream, or get the open one. A closed, expired or another owner's Stream
@@ -166,7 +198,9 @@ def from_transfer_type(
         id = value.id
         return Producer(
             id=id,
-            kind=StreamProducerKind(value.kind) if value.kind != 0 else None,
+            kind=_stream_producer_kind_from_proto(value.kind)
+            if value.kind != 0
+            else None,
             epoch=_ProducerEpochTransferTypeConverter().from_transfer_type(
                 value.epoch, ProducerEpoch
             )
@@ -403,7 +437,7 @@ def from_transfer_type(
         type_hint: type[Execution],
     ) -> Execution:
         return Execution(
-            type=ExecutionType(value.type) if value.type != 0 else None,
+            type=_execution_type_from_proto(value.type) if value.type != 0 else None,
             business_id=value.business_id if bool(value.business_id) else None,
             run_id=value.run_id if bool(value.run_id) else None,
         )
@@ -1090,7 +1124,7 @@ def from_transfer_type(
         return Supersession(
             producer_id=producer_id,
             failed_tasks=list(value.failed_tasks),
-            producer_kind=StreamProducerKind(value.producer_kind)
+            producer_kind=_stream_producer_kind_from_proto(value.producer_kind)
             if value.producer_kind != 0
             else None,
             epoch=_ProducerEpochTransferTypeConverter().from_transfer_type(
@@ -1239,7 +1273,9 @@ def from_transfer_type(
             failure=_support.failure_from_proto(value.failure)
             if value.HasField("failure")
             else None,
-            status=StreamCloseStatus(value.status) if value.status != 0 else None,
+            status=_stream_close_status_from_proto(value.status)
+            if value.status != 0
+            else None,
         )
 
     @typing_extensions.override
@@ -1347,7 +1383,9 @@ def from_transfer_type(
         type_hint: type[DescribeStreamResponse],
     ) -> DescribeStreamResponse:
         return DescribeStreamResponse(
-            status=StreamStatus(value.status) if value.status != 0 else None,
+            status=_stream_status_from_proto(value.status)
+            if value.status != 0
+            else None,
             last_offset=value.last_offset if value.last_offset != 0 else None,
             retention=_support.duration_from_proto(value.retention)
             if value.HasField("retention")

advanced/samples/typescript/tests/stream-service.test.ts

@@ -6,6 +6,8 @@ import { describe, expect, test } from "vitest";
 import {
   type ReadRecordsResponse,
   StreamCloseStatus,
+  StreamStatus,
+  describeStreamResponseTransferTypeConverter,
   readRecordsRequestTransferTypeConverter,
   readRecordsResponseTransferTypeConverter,
 } from "../wit/stream-service/models.ts";
@@ -86,4 +88,15 @@ describe("stream-service generated output", () => {
       readRecordsRequestTransferTypeConverter.toTransferType(model).stream?.id,
     ).toBe("stream-1");
   });
+
+  test("decodes an enum number that it doesn't know as the fallback case", () => {
+    // A newer server can send a Stream status that this code doesn't know.
+    const proto: temporal.api.streamservice.v1.IDescribeStreamResponse = {
+      status: 9 as temporal.api.enums.v1.StreamStatus,
+    };
+
+    const model = describeStreamResponseTransferTypeConverter.fromTransferType(proto);
+
+    expect(model.status).toBe(StreamStatus.StreamStatusUnspecified);
+  });
 });