feat: let @nexus.source and @nexus.api-omit name their target

GUIDE.md

@@ -1003,6 +1003,42 @@ Nexgen prefixes the helper call, but it does not add arguments. The configured
 package supplies the helper. The `@nexus.output-transform` directive is
 different: its value remains a raw expression.
 
+#### Sources for one target
+
+`target="workflow"` or `target="client"` limits a source to one generation
+target, and `@nexus.api-omit` takes the same `target`. In each target, a field
+is one of three things. A source of that target fills it, the call leaves it
+unset, or the convenience API takes it as a parameter.
+
+```wit
+/// @nexus.source target="workflow" python="workflow_producer()"
+/// @nexus.source target="client" python="client_producer()"
+producer: option<producer>,
+
+/// @nexus.api-omit target="workflow"
+/// @nexus.source target="client" python="local_activity_failed_tasks()"
+failed-tasks: list<s64>,
+```
+
+Target code runs a targeted source, never the model, since code of another
+target can build the same model. The convenience function of an operation runs
+the sources of its target before it builds the request. The operations of a
+target also fill each optional or list field that the request leaves unset, so a
+resource method and SDK code get the sources too. A field that SDK code sets
+keeps its value. A source with no target stays the model's default, and the
+client target takes that field as a parameter.
+
+A source may read `definition`, the definition argument of an operation that
+returns a resource with `@nexus.definition`. Only target code has it, so such a
+source must name its target.
+
+Python has both targets. Go, TypeScript and .NET have only the Workflow target,
+so they read the Workflow source and api-omit, and they ignore the client ones.
+Java has no sources. nexgen refuses an unknown target, two sources for one
+target, a source with no target next to a targeted one, a source and an api-omit
+for the same target, and a resource method parameter that names a sourced or
+api-omitted request field.
+
 ### Omitted Fields
 
 Fields annotated with `@nexus.omit` are excluded entirely from the generated
@@ -1028,7 +1064,10 @@ making it a public operation parameter.
 headers: option<header>,
 ```
 
-`@nexus.api-omit` cannot be combined with `@nexus.omit` or `@nexus.source`.
+`@nexus.api-omit` cannot be combined with `@nexus.omit` or `@nexus.source`. A
+targeted api-omit can be combined with a source for the other target, as in
+[Sources for one target](#sources-for-one-target). A required field can't be
+api-omitted, since the call would leave it unset, but a list or a map can.
 
 ### Flattened Records
 
@@ -1526,8 +1565,10 @@ The methods of the resource take the RPC options too. The Workflow operations
 raise `ValueError` when either option is set, since the options apply to calls
 outside a Workflow only.
 
-The client target has no sources yet, so a `@nexus.source` field is a parameter
-of the client call.
+A `@nexus.source` with no target is the model's default, which can read the
+Workflow context, so the client call takes that field as a parameter. A source
+with `target="client"` fills the field in the client target. See
+[Sources for one target](#sources-for-one-target).
 
 ---
 
@@ -1887,7 +1928,7 @@ Cannot be combined with `@nexus.source`, `@nexus.type`, `@nexus.function`,
 ### @nexus.source
 
 **Placement:** Record field (within a `@nexus.proto` record) **Syntax:**
-`@nexus.source python="<helper-call>" typescript="<helper-call>" go="<helper-call>" dotnet="<helper-call>"`
+`@nexus.source [target="workflow|client"] python="<helper-call>" typescript="<helper-call>" go="<helper-call>" dotnet="<helper-call>"`
 
 Provides a field value through a helper in the package specified by
 `--support-package`. The field is not an argument in operation convenience APIs.
@@ -1902,7 +1943,8 @@ namespace: string,
 
 The sample's Python model uses `_support.workflow_namespace` as its default
 factory. Go calls `support.WorkflowNamespace(ctx)`. Both helpers come from the
-configured support package.
+configured support package. `target` limits a source to one generation target.
+See [Sources for one target](#sources-for-one-target).
 
 Cannot be combined with `@nexus.default`.
 

advanced/samples/README.md

@@ -177,7 +177,8 @@ clients. The read poll names the client target, since Workflow code reads from a
 Local Activity. The Stream and the writer are generic in the record body type
 `body-t`, so a handle decodes the bodies of a read as its own type argument. The
 create of a Stream takes a definition, which is a type of the sample support
-code, since no SDK has `StreamDefinition` yet.
+code, since no SDK has `StreamDefinition` yet. The requests take a source for
+each target, which the sample support code gives in place of Brook's runtime.
 
 - WIT: [`inputs/stream-service.wit`](inputs/stream-service.wit),
   [`inputs/stream-writer.wit`](inputs/stream-writer.wit)

advanced/samples/dotnet/Nexgen.DotNetWorkflowServiceDocs.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <Compile Remove="bin/**/*.cs;obj/**/*.cs;tests/**/*.cs;json_schema/**/*.cs;*.cs;wit/**/*.cs;TemporalSupport/**/*.cs" />
+    <Compile Remove="bin/**/*.cs;obj/**/*.cs;tests/**/*.cs;json_schema/**/*.cs;*.cs;wit/**/*.cs;TemporalSupport/**/*.cs;StreamSupport/**/*.cs" />
     <!-- The generated interceptor partials require matching SDK changes. -->
     <Compile Include="wit/workflow-service/**/*.cs"
              Exclude="wit/workflow-service/SystemNexusWorkflowOutboundInterceptor.cs;wit/workflow-service/Support/*.cs" />

advanced/samples/dotnet/StreamSupport/StreamSupport.cs

@@ -0,0 +1,17 @@
+using Temporalio.Workflows;
+
+namespace Nexgen.Support
+{
+    // Sources of the stream sample. They stand in for the ones of Brook's runtime.
+    internal static class StreamWorkflowContext
+    {
+        internal static string StreamId<T>(StreamDefinition<T> definition) =>
+            $"{Workflow.Info.WorkflowId}:{Workflow.Info.FirstExecutionRunId}:{definition.Name}";
+
+        internal static Nexgen.StreamService.Producer Owner() =>
+            new($"wf:{Workflow.Info.RunId}") { RunId = Workflow.Info.RunId };
+
+        internal static Nexgen.StreamService.Producer Producer() =>
+            new($"wf:{Workflow.Info.RunId}") { RunId = Workflow.Info.RunId };
+    }
+}

advanced/samples/dotnet/tests/Nexgen.DotNetExamples.Tests.csproj

@@ -21,6 +21,7 @@
     <Compile Include="../wit/proto-oneof/Models.cs" Link="GeneratedProtoOneof/Models.cs" />
     <Compile Include="../wit/proto-generic/Models.cs" Link="GeneratedProtoGeneric/Models.cs" />
     <Compile Include="../wit/stream-service/*.cs" Link="GeneratedStreamService/%(Filename)%(Extension)" />
+    <Compile Include="../StreamSupport/*.cs" Link="StreamSupport/%(Filename)%(Extension)" />
   </ItemGroup>
 
   <ItemGroup>

advanced/samples/dotnet/tests/StreamServiceChecks.cs

@@ -81,11 +81,10 @@ public void ReadItemsRoundTripAsTheirRecords()
             StreamServiceExample.Stream<Turn> stream) =>
             stream.ReadAsync(null, null, null, null, null);
 
-        // A definition gives the type argument of a create.
+        // A definition gives the type argument of a create, and the Workflow source gives the id.
         private static Task<StreamServiceExample.Stream<Turn>> CreateTurns() =>
             StreamServiceExample.Operations.CreateStreamAsync(
-                new Nexgen.Support.StreamDefinition<Turn>("turns"),
-                new StreamServiceExample.CreateStreamOptions("stream-1"));
+                new Nexgen.Support.StreamDefinition<Turn>("turns"));
 
         // A newer server can send a Stream status that this code doesn't know.
         [Fact]

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

@@ -140,23 +140,19 @@ public record AppendRecordsRequest<TBody>
         /// <summary>
         /// Initializes a new instance of the <see cref="AppendRecordsRequest{TBody}"/> class.
         /// </summary>
-        public AppendRecordsRequest(StreamRef stream, IReadOnlyList<AppendRecord<TBody>> records, IReadOnlyList<long> failedTasks)
+        public AppendRecordsRequest(StreamRef stream, IReadOnlyList<AppendRecord<TBody>> records, Producer? producer)
         {
             Stream = stream;
             Records = records;
-            FailedTasks = failedTasks;
+            Producer = producer;
         }
 
         public StreamRef Stream { get; init; }
         public IReadOnlyList<AppendRecord<TBody>> Records { get; init; }
         /// <summary>
         /// Local Activity only. The run's failed or timed out Workflow Tasks.
         /// </summary>
-        public IReadOnlyList<long> FailedTasks { get; init; }
-        /// <summary>
-        /// Required outside a Workflow. In a Workflow the transport sets it.
-        /// </summary>
-        public Producer? Producer { get; init; }
+        public IReadOnlyList<long> FailedTasks { get; init; } = System.Array.Empty<long>();
         /// <summary>
         /// The batch number in this epoch, 1 or more. It is part of the dedupe key.
         /// </summary>
@@ -169,12 +165,16 @@ public AppendRecordsRequest(StreamRef stream, IReadOnlyList<AppendRecord<TBody>>
         /// Retry run only. The previous run, whose Records this withdraws.
         /// </summary>
         public string? WithdrawRun { get; init; }
+        /// <summary>
+        /// Required outside a Workflow. In a Workflow the transport sets it.
+        /// </summary>
+        public Producer? Producer { get; init; }
 
         internal static AppendRecordsRequest<TBody> FromTransferType(Temporalio.Api.StreamService.V1.AppendRecordsRequest wire)
         {
-            return new AppendRecordsRequest<TBody>(StreamRef.FromTransferType(wire.Stream), System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.Records, item => AppendRecord<TBody>.FromTransferType(item))), System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.FailedTasks, item => item)))
+            return new AppendRecordsRequest<TBody>(StreamRef.FromTransferType(wire.Stream), System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.Records, item => AppendRecord<TBody>.FromTransferType(item))), wire.Producer == null ? null : Producer.FromTransferType(wire.Producer))
             {
-                Producer = wire.Producer == null ? null : Producer.FromTransferType(wire.Producer),
+                FailedTasks = System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.FailedTasks, item => item)),
                 Sequence = wire.Sequence == 0 ? null : wire.Sequence,
                 PreviousCompletedTask = wire.PreviousCompletedTask == null ? null : WorkflowTaskEpoch.FromTransferType(wire.PreviousCompletedTask),
                 WithdrawRun = string.IsNullOrEmpty(wire.WithdrawRun) ? null : wire.WithdrawRun,
@@ -184,13 +184,13 @@ internal static AppendRecordsRequest<TBody> FromTransferType(Temporalio.Api.Stre
         internal Temporalio.Api.StreamService.V1.AppendRecordsRequest ToTransferType()
         {
             var proto = new Temporalio.Api.StreamService.V1.AppendRecordsRequest();
-            proto.Stream = (Temporalio.Api.Stream.V1.StreamRef)Stream.ToTransferType();
-            proto.Records.Add(System.Linq.Enumerable.Select(Records, item => (Temporalio.Api.Stream.V1.AppendRecord)item.ToTransferType()));
-            proto.FailedTasks.Add(System.Linq.Enumerable.Select(FailedTasks, item => item));
             if (Producer is { } producer)
             {
                 proto.Producer = (Temporalio.Api.Stream.V1.Producer)producer.ToTransferType();
             }
+            proto.Stream = (Temporalio.Api.Stream.V1.StreamRef)Stream.ToTransferType();
+            proto.Records.Add(System.Linq.Enumerable.Select(Records, item => (Temporalio.Api.Stream.V1.AppendRecord)item.ToTransferType()));
+            proto.FailedTasks.Add(System.Linq.Enumerable.Select(FailedTasks, item => item));
             if (Sequence is { } sequence)
             {
                 proto.Sequence = sequence;
@@ -280,34 +280,34 @@ public record CloseStreamRequest
         /// <summary>
         /// Initializes a new instance of the <see cref="CloseStreamRequest"/> class.
         /// </summary>
-        public CloseStreamRequest(StreamRef stream, IReadOnlyList<long> failedTasks)
+        public CloseStreamRequest(StreamRef stream, Producer? producer)
         {
             Stream = stream;
-            FailedTasks = failedTasks;
+            Producer = producer;
         }
 
         public StreamRef Stream { get; init; }
         /// <summary>
         /// Workflow only. The run's failed or timed out Workflow Tasks.
         /// </summary>
-        public IReadOnlyList<long> FailedTasks { get; init; }
-        public Producer? Producer { get; init; }
+        public IReadOnlyList<long> FailedTasks { get; init; } = System.Array.Empty<long>();
         public object? Result { get; init; }
         public System.Exception? Failure { get; init; }
         /// <summary>
-        /// Empty is completed, or failed with a failure. A caller may set canceled.
+        /// Empty is completed, failed with a failure, or canceled with a cancel.
         /// </summary>
         public StreamCloseStatus? Status { get; init; }
         /// <summary>
         /// Workflow only. The previous completed Workflow Task of the run.
         /// </summary>
         public WorkflowTaskEpoch? PreviousCompletedTask { get; init; }
+        public Producer? Producer { get; init; }
 
         internal static CloseStreamRequest FromTransferType(Temporalio.Api.StreamService.V1.CloseStreamRequest wire)
         {
-            return new CloseStreamRequest(StreamRef.FromTransferType(wire.Stream), System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.FailedTasks, item => item)))
+            return new CloseStreamRequest(StreamRef.FromTransferType(wire.Stream), wire.Producer == null ? null : Producer.FromTransferType(wire.Producer))
             {
-                Producer = wire.Producer == null ? null : Producer.FromTransferType(wire.Producer),
+                FailedTasks = System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.FailedTasks, item => item)),
                 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,
@@ -318,12 +318,12 @@ internal static CloseStreamRequest FromTransferType(Temporalio.Api.StreamService
         internal Temporalio.Api.StreamService.V1.CloseStreamRequest ToTransferType()
         {
             var proto = new Temporalio.Api.StreamService.V1.CloseStreamRequest();
-            proto.Stream = (Temporalio.Api.Stream.V1.StreamRef)Stream.ToTransferType();
-            proto.FailedTasks.Add(System.Linq.Enumerable.Select(FailedTasks, item => item));
             if (Producer is { } producer)
             {
                 proto.Producer = (Temporalio.Api.Stream.V1.Producer)producer.ToTransferType();
             }
+            proto.Stream = (Temporalio.Api.Stream.V1.StreamRef)Stream.ToTransferType();
+            proto.FailedTasks.Add(System.Linq.Enumerable.Select(FailedTasks, item => item));
             if (Result is { } result)
             {
                 proto.Result = Nexgen.Support.ProtoExtensions.ToPayload(result);
@@ -402,15 +402,12 @@ public record CreateStreamRequest
         /// <summary>
         /// Initializes a new instance of the <see cref="CreateStreamRequest"/> class.
         /// </summary>
-        public CreateStreamRequest(string streamId)
+        public CreateStreamRequest(string streamId, Producer? owner)
         {
             StreamId = streamId;
+            Owner = owner;
         }
 
-        /// <summary>
-        /// Required on Redis and memory. On native, empty asks the server for an id.
-        /// </summary>
-        public string StreamId { get; init; }
         /// <summary>
         /// The age after which a Record is dropped. Empty is the Store default.
         /// </summary>
@@ -420,21 +417,24 @@ public CreateStreamRequest(string streamId)
         /// </summary>
         public string? WithdrawRun { get; init; }
         /// <summary>
-        /// The Workflow run that owns the Stream. The Workflow transport sets it.
-        /// </summary>
-        public Producer? Owner { get; init; }
-        /// <summary>
         /// The Stream's ref, with the owner fields.
         /// </summary>
         public StreamRef? Ref { get; init; }
+        /// <summary>
+        /// The id of the Stream. Required, and never empty.
+        /// </summary>
+        public string StreamId { get; init; }
+        /// <summary>
+        /// The Workflow run that owns the Stream. The Workflow transport sets it.
+        /// </summary>
+        public Producer? Owner { get; init; }
 
         internal static CreateStreamRequest FromTransferType(Temporalio.Api.StreamService.V1.CreateStreamRequest wire)
         {
-            return new CreateStreamRequest(wire.StreamId)
+            return new CreateStreamRequest(wire.StreamId, wire.Owner == null ? null : Producer.FromTransferType(wire.Owner))
             {
                 Retention = wire.Retention == null ? null : Nexgen.Support.ProtoExtensions.FromDurationProto(wire.Retention),
                 WithdrawRun = string.IsNullOrEmpty(wire.WithdrawRun) ? null : wire.WithdrawRun,
-                Owner = wire.Owner == null ? null : Producer.FromTransferType(wire.Owner),
                 Ref = wire.Ref == null ? null : StreamRef.FromTransferType(wire.Ref),
             };
         }
@@ -443,6 +443,10 @@ internal Temporalio.Api.StreamService.V1.CreateStreamRequest ToTransferType()
         {
             var proto = new Temporalio.Api.StreamService.V1.CreateStreamRequest();
             proto.StreamId = StreamId;
+            if (Owner is { } owner)
+            {
+                proto.Owner = (Temporalio.Api.Stream.V1.Producer)owner.ToTransferType();
+            }
             if (Retention is { } retention)
             {
                 proto.Retention = retention.ToProto();
@@ -451,10 +455,6 @@ internal Temporalio.Api.StreamService.V1.CreateStreamRequest ToTransferType()
             {
                 proto.WithdrawRun = withdrawRun;
             }
-            if (Owner is { } owner)
-            {
-                proto.Owner = (Temporalio.Api.Stream.V1.Producer)owner.ToTransferType();
-            }
             if (Ref is { } @ref)
             {
                 proto.Ref = (Temporalio.Api.Stream.V1.StreamRef)@ref.ToTransferType();