diff --git a/GUIDE.md b/GUIDE.md
index 8ebe62945..f5da84e63 100644
--- a/GUIDE.md
+++ b/GUIDE.md
@@ -713,6 +713,27 @@ return StartedWorkflow(
 If a resource constructor field name cannot be found in either, the generator
 raises `InvalidResource`.
 
+**A result record that holds the resource.** An operation can also return a
+proto-backed record with one `own<T>` field and other fields. The resource binds
+from the record's message as above, and the other fields come from the fields of
+the same name. The record is the API model only, and it declares only the fields
+that its caller sees. Its message travels as a wire record that takes the proto
+message's name. Python, TypeScript, .NET and Go generate it. In Python, the
+record lives in its resource's module.
+
+```wit
+/// @nexus.proto "acme.jobs.v1.CreateJobResponse"
+record create-job-result {
+  job: own<job>,
+  created: option<bool>,
+}
+```
+
+```python
+result = await handle  # CreateJobResponse
+return CreateJobResult(job=Job(job_id=result.job_id), created=result.created)
+```
+
 ### Edge Cases and Error Conditions
 
 #### Same field name in constructor and method parameter
diff --git a/advanced/samples/README.md b/advanced/samples/README.md
index e6f7b763b..65c308241 100644
--- a/advanced/samples/README.md
+++ b/advanced/samples/README.md
@@ -206,4 +206,5 @@ protoc -I descriptors -I "$API" --csharp_out=dotnet/proto streamsample/v1/stream
 ```
 
 The TypeScript `typescript/proto/streamsample.d.ts` mirrors the proto by hand,
-with only the `I<Message>` interfaces that the generated code imports.
+with only the `I<Message>` interfaces and the enums that the generated code
+imports.
diff --git a/advanced/samples/descriptors/stream_service.bin b/advanced/samples/descriptors/stream_service.bin
index 5a375ff37..8dd29f09a 100644
Binary files a/advanced/samples/descriptors/stream_service.bin and b/advanced/samples/descriptors/stream_service.bin differ
diff --git a/advanced/samples/descriptors/streamsample/v1/stream_service.proto b/advanced/samples/descriptors/streamsample/v1/stream_service.proto
index 7a5725cbb..a846bff56 100644
--- a/advanced/samples/descriptors/streamsample/v1/stream_service.proto
+++ b/advanced/samples/descriptors/streamsample/v1/stream_service.proto
@@ -30,9 +30,17 @@ message CreateStreamRequest {
     google.protobuf.Duration retention = 3;
 }
 
+enum StreamStatus {
+    STREAM_STATUS_UNSPECIFIED = 0;
+    STREAM_STATUS_OPEN = 1;
+    STREAM_STATUS_CLOSED = 2;
+    STREAM_STATUS_EXPIRED = 3;
+}
+
 message CreateStreamResponse {
     StreamRef stream = 1;
     bool created = 2;
+    StreamStatus status = 3;
 }
 
 message AppendRecordsRequest {
diff --git a/advanced/samples/dotnet/proto/StreamService.cs b/advanced/samples/dotnet/proto/StreamService.cs
index 0e9119d8d..79b87a0ed 100644
--- a/advanced/samples/dotnet/proto/StreamService.cs
+++ b/advanced/samples/dotnet/proto/StreamService.cs
@@ -34,36 +34,40 @@ static StreamServiceReflection() {
             "bXBvcmFsLmFwaS5jb21tb24udjEuUGF5bG9hZCJ/ChNDcmVhdGVTdHJlYW1S",
             "ZXF1ZXN0EhEKCW5hbWVzcGFjZRgBIAEoCRInCgNyZWYYAiABKAsyGi5zdHJl",
             "YW1zYW1wbGUudjEuU3RyZWFtUmVmEiwKCXJldGVudGlvbhgDIAEoCzIZLmdv",
-            "b2dsZS5wcm90b2J1Zi5EdXJhdGlvbiJTChRDcmVhdGVTdHJlYW1SZXNwb25z",
-            "ZRIqCgZzdHJlYW0YASABKAsyGi5zdHJlYW1zYW1wbGUudjEuU3RyZWFtUmVm",
-            "Eg8KB2NyZWF0ZWQYAiABKAgilwEKFEFwcGVuZFJlY29yZHNSZXF1ZXN0EhEK",
-            "CW5hbWVzcGFjZRgBIAEoCRIqCgZzdHJlYW0YAiABKAsyGi5zdHJlYW1zYW1w",
-            "bGUudjEuU3RyZWFtUmVmEhAKCHNlcXVlbmNlGAMgASgDEi4KB3JlY29yZHMY",
-            "BCADKAsyHS5zdHJlYW1zYW1wbGUudjEuQXBwZW5kUmVjb3JkIkIKFUFwcGVu",
-            "ZFJlY29yZHNSZXNwb25zZRIUCgxmaXJzdF9vZmZzZXQYASABKAMSEwoLbGFz",
-            "dF9vZmZzZXQYAiABKAMihAEKEkNsb3NlU3RyZWFtUmVxdWVzdBIRCgluYW1l",
-            "c3BhY2UYASABKAkSKgoGc3RyZWFtGAIgASgLMhouc3RyZWFtc2FtcGxlLnYx",
-            "LlN0cmVhbVJlZhIvCgZyZXN1bHQYAyABKAsyHy50ZW1wb3JhbC5hcGkuY29t",
-            "bW9uLnYxLlBheWxvYWQiFQoTQ2xvc2VTdHJlYW1SZXNwb25zZSK2AQoSUmVh",
-            "ZFJlY29yZHNSZXF1ZXN0EhEKCW5hbWVzcGFjZRgBIAEoCRIqCgZzdHJlYW0Y",
-            "AiABKAsyGi5zdHJlYW1zYW1wbGUudjEuU3RyZWFtUmVmEhMKC2Zyb21fb2Zm",
-            "c2V0GAMgASgDEg4KBnRvcGljcxgEIAMoCRITCgttYXhfcmVjb3JkcxgFIAEo",
-            "BRInCgR3YWl0GAYgASgLMhkuZ29vZ2xlLnByb3RvYnVmLkR1cmF0aW9uImgK",
-            "E1JlYWRSZWNvcmRzUmVzcG9uc2USLgoHcmVjb3JkcxgBIAMoCzIdLnN0cmVh",
-            "bXNhbXBsZS52MS5TdHJlYW1SZWNvcmQSEwoLbmV4dF9vZmZzZXQYAiABKAMS",
-            "DAoEZG9uZRgDIAEoCCJWChVEZXNjcmliZVN0cmVhbVJlcXVlc3QSEQoJbmFt",
-            "ZXNwYWNlGAEgASgJEioKBnN0cmVhbRgCIAEoCzIaLnN0cmVhbXNhbXBsZS52",
-            "MS5TdHJlYW1SZWYiWwoWRGVzY3JpYmVTdHJlYW1SZXNwb25zZRITCgtsYXN0",
-            "X29mZnNldBgBIAEoAxIsCglyZXRlbnRpb24YAiABKAsyGS5nb29nbGUucHJv",
-            "dG9idWYuRHVyYXRpb25CEqoCD1N0cmVhbVNhbXBsZS5WMWIGcHJvdG8z"));
+            "b2dsZS5wcm90b2J1Zi5EdXJhdGlvbiKCAQoUQ3JlYXRlU3RyZWFtUmVzcG9u",
+            "c2USKgoGc3RyZWFtGAEgASgLMhouc3RyZWFtc2FtcGxlLnYxLlN0cmVhbVJl",
+            "ZhIPCgdjcmVhdGVkGAIgASgIEi0KBnN0YXR1cxgDIAEoDjIdLnN0cmVhbXNh",
+            "bXBsZS52MS5TdHJlYW1TdGF0dXMilwEKFEFwcGVuZFJlY29yZHNSZXF1ZXN0",
+            "EhEKCW5hbWVzcGFjZRgBIAEoCRIqCgZzdHJlYW0YAiABKAsyGi5zdHJlYW1z",
+            "YW1wbGUudjEuU3RyZWFtUmVmEhAKCHNlcXVlbmNlGAMgASgDEi4KB3JlY29y",
+            "ZHMYBCADKAsyHS5zdHJlYW1zYW1wbGUudjEuQXBwZW5kUmVjb3JkIkIKFUFw",
+            "cGVuZFJlY29yZHNSZXNwb25zZRIUCgxmaXJzdF9vZmZzZXQYASABKAMSEwoL",
+            "bGFzdF9vZmZzZXQYAiABKAMihAEKEkNsb3NlU3RyZWFtUmVxdWVzdBIRCglu",
+            "YW1lc3BhY2UYASABKAkSKgoGc3RyZWFtGAIgASgLMhouc3RyZWFtc2FtcGxl",
+            "LnYxLlN0cmVhbVJlZhIvCgZyZXN1bHQYAyABKAsyHy50ZW1wb3JhbC5hcGku",
+            "Y29tbW9uLnYxLlBheWxvYWQiFQoTQ2xvc2VTdHJlYW1SZXNwb25zZSK2AQoS",
+            "UmVhZFJlY29yZHNSZXF1ZXN0EhEKCW5hbWVzcGFjZRgBIAEoCRIqCgZzdHJl",
+            "YW0YAiABKAsyGi5zdHJlYW1zYW1wbGUudjEuU3RyZWFtUmVmEhMKC2Zyb21f",
+            "b2Zmc2V0GAMgASgDEg4KBnRvcGljcxgEIAMoCRITCgttYXhfcmVjb3JkcxgF",
+            "IAEoBRInCgR3YWl0GAYgASgLMhkuZ29vZ2xlLnByb3RvYnVmLkR1cmF0aW9u",
+            "ImgKE1JlYWRSZWNvcmRzUmVzcG9uc2USLgoHcmVjb3JkcxgBIAMoCzIdLnN0",
+            "cmVhbXNhbXBsZS52MS5TdHJlYW1SZWNvcmQSEwoLbmV4dF9vZmZzZXQYAiAB",
+            "KAMSDAoEZG9uZRgDIAEoCCJWChVEZXNjcmliZVN0cmVhbVJlcXVlc3QSEQoJ",
+            "bmFtZXNwYWNlGAEgASgJEioKBnN0cmVhbRgCIAEoCzIaLnN0cmVhbXNhbXBs",
+            "ZS52MS5TdHJlYW1SZWYiWwoWRGVzY3JpYmVTdHJlYW1SZXNwb25zZRITCgts",
+            "YXN0X29mZnNldBgBIAEoAxIsCglyZXRlbnRpb24YAiABKAsyGS5nb29nbGUu",
+            "cHJvdG9idWYuRHVyYXRpb24qegoMU3RyZWFtU3RhdHVzEh0KGVNUUkVBTV9T",
+            "VEFUVVNfVU5TUEVDSUZJRUQQABIWChJTVFJFQU1fU1RBVFVTX09QRU4QARIY",
+            "ChRTVFJFQU1fU1RBVFVTX0NMT1NFRBACEhkKFVNUUkVBTV9TVEFUVVNfRVhQ",
+            "SVJFRBADQhKqAg9TdHJlYW1TYW1wbGUuVjFiBnByb3RvMw=="));
       descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData,
           new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.DurationReflection.Descriptor, global::Temporalio.Api.Common.V1.MessageReflection.Descriptor, },
-          new pbr::GeneratedClrTypeInfo(null, null, new pbr::GeneratedClrTypeInfo[] {
+          new pbr::GeneratedClrTypeInfo(new[] {typeof(global::StreamSample.V1.StreamStatus), }, null, new pbr::GeneratedClrTypeInfo[] {
             new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.StreamRef), global::StreamSample.V1.StreamRef.Parser, new[]{ "Id", "WorkflowId" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.AppendRecord), global::StreamSample.V1.AppendRecord.Parser, new[]{ "Topic", "Body" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.StreamRecord), global::StreamSample.V1.StreamRecord.Parser, new[]{ "Offset", "Topic", "Body" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.CreateStreamRequest), global::StreamSample.V1.CreateStreamRequest.Parser, new[]{ "Namespace", "Ref", "Retention" }, null, null, null, null),
-            new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.CreateStreamResponse), global::StreamSample.V1.CreateStreamResponse.Parser, new[]{ "Stream", "Created" }, null, null, null, null),
+            new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.CreateStreamResponse), global::StreamSample.V1.CreateStreamResponse.Parser, new[]{ "Stream", "Created", "Status" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.AppendRecordsRequest), global::StreamSample.V1.AppendRecordsRequest.Parser, new[]{ "Namespace", "Stream", "Sequence", "Records" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.AppendRecordsResponse), global::StreamSample.V1.AppendRecordsResponse.Parser, new[]{ "FirstOffset", "LastOffset" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::StreamSample.V1.CloseStreamRequest), global::StreamSample.V1.CloseStreamRequest.Parser, new[]{ "Namespace", "Stream", "Result" }, null, null, null, null),
@@ -77,6 +81,16 @@ static StreamServiceReflection() {
     #endregion
 
   }
+  #region Enums
+  public enum StreamStatus {
+    [pbr::OriginalName("STREAM_STATUS_UNSPECIFIED")] Unspecified = 0,
+    [pbr::OriginalName("STREAM_STATUS_OPEN")] Open = 1,
+    [pbr::OriginalName("STREAM_STATUS_CLOSED")] Closed = 2,
+    [pbr::OriginalName("STREAM_STATUS_EXPIRED")] Expired = 3,
+  }
+
+  #endregion
+
   #region Messages
   [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")]
   public sealed partial class StreamRef : pb::IMessage<StreamRef>
@@ -1165,6 +1179,7 @@ public CreateStreamResponse() {
     public CreateStreamResponse(CreateStreamResponse other) : this() {
       stream_ = other.stream_ != null ? other.stream_.Clone() : null;
       created_ = other.created_;
+      status_ = other.status_;
       _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields);
     }
 
@@ -1198,6 +1213,18 @@ public bool Created {
       }
     }
 
+    /// <summary>Field number for the "status" field.</summary>
+    public const int StatusFieldNumber = 3;
+    private global::StreamSample.V1.StreamStatus status_ = global::StreamSample.V1.StreamStatus.Unspecified;
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public global::StreamSample.V1.StreamStatus Status {
+      get { return status_; }
+      set {
+        status_ = value;
+      }
+    }
+
     [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
     [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
     public override bool Equals(object other) {
@@ -1215,6 +1242,7 @@ public bool Equals(CreateStreamResponse other) {
       }
       if (!object.Equals(Stream, other.Stream)) return false;
       if (Created != other.Created) return false;
+      if (Status != other.Status) return false;
       return Equals(_unknownFields, other._unknownFields);
     }
 
@@ -1224,6 +1252,7 @@ public override int GetHashCode() {
       int hash = 1;
       if (stream_ != null) hash ^= Stream.GetHashCode();
       if (Created != false) hash ^= Created.GetHashCode();
+      if (Status != global::StreamSample.V1.StreamStatus.Unspecified) hash ^= Status.GetHashCode();
       if (_unknownFields != null) {
         hash ^= _unknownFields.GetHashCode();
       }
@@ -1250,6 +1279,10 @@ public void WriteTo(pb::CodedOutputStream output) {
         output.WriteRawTag(16);
         output.WriteBool(Created);
       }
+      if (Status != global::StreamSample.V1.StreamStatus.Unspecified) {
+        output.WriteRawTag(24);
+        output.WriteEnum((int) Status);
+      }
       if (_unknownFields != null) {
         _unknownFields.WriteTo(output);
       }
@@ -1268,6 +1301,10 @@ public void WriteTo(pb::CodedOutputStream output) {
         output.WriteRawTag(16);
         output.WriteBool(Created);
       }
+      if (Status != global::StreamSample.V1.StreamStatus.Unspecified) {
+        output.WriteRawTag(24);
+        output.WriteEnum((int) Status);
+      }
       if (_unknownFields != null) {
         _unknownFields.WriteTo(ref output);
       }
@@ -1284,6 +1321,9 @@ public int CalculateSize() {
       if (Created != false) {
         size += 1 + 1;
       }
+      if (Status != global::StreamSample.V1.StreamStatus.Unspecified) {
+        size += 1 + pb::CodedOutputStream.ComputeEnumSize((int) Status);
+      }
       if (_unknownFields != null) {
         size += _unknownFields.CalculateSize();
       }
@@ -1305,6 +1345,9 @@ public void MergeFrom(CreateStreamResponse other) {
       if (other.Created != false) {
         Created = other.Created;
       }
+      if (other.Status != global::StreamSample.V1.StreamStatus.Unspecified) {
+        Status = other.Status;
+      }
       _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields);
     }
 
@@ -1335,6 +1378,10 @@ public void MergeFrom(pb::CodedInputStream input) {
             Created = input.ReadBool();
             break;
           }
+          case 24: {
+            Status = (global::StreamSample.V1.StreamStatus) input.ReadEnum();
+            break;
+          }
         }
       }
     #endif
@@ -1365,6 +1412,10 @@ public void MergeFrom(pb::CodedInputStream input) {
             Created = input.ReadBool();
             break;
           }
+          case 24: {
+            Status = (global::StreamSample.V1.StreamStatus) input.ReadEnum();
+            break;
+          }
         }
       }
     }
diff --git a/advanced/samples/dotnet/wit/stream-service/Models.cs b/advanced/samples/dotnet/wit/stream-service/Models.cs
index 226040f1e..0e8502d4b 100644
--- a/advanced/samples/dotnet/wit/stream-service/Models.cs
+++ b/advanced/samples/dotnet/wit/stream-service/Models.cs
@@ -10,6 +10,68 @@
 namespace Nexgen.StreamService
 {
 
+    [GeneratedCode("nexgen", null)]
+    public enum StreamStatus
+    {
+        StreamStatusUnspecified = 0,
+        StreamStatusOpen = 1,
+        StreamStatusClosed = 2,
+        StreamStatusExpired = 3,
+    }
+
+    [Temporalio.Converters.TemporalTransferTypeConverter(typeof(CreateStreamResponse.TransferTypeConverter))]
+    [GeneratedCode("nexgen", null)]
+    public record CreateStreamResponse
+    {
+        public StreamRef? Stream { get; init; }
+        public bool? Created { get; init; }
+        public StreamStatus? Status { get; init; }
+
+        internal static CreateStreamResponse FromTransferType(StreamSample.V1.CreateStreamResponse wire)
+        {
+            return new CreateStreamResponse()
+            {
+                Stream = wire.Stream == null ? null : StreamRef.FromTransferType(wire.Stream),
+                Created = wire.Created == false ? null : wire.Created,
+                Status = (int)wire.Status == 0 ? null : (StreamStatus)wire.Status,
+            };
+        }
+
+        internal StreamSample.V1.CreateStreamResponse ToTransferType()
+        {
+            var proto = new StreamSample.V1.CreateStreamResponse();
+            if (Stream is { } stream)
+            {
+                proto.Stream = (StreamSample.V1.StreamRef)stream.ToTransferType();
+            }
+            if (Created is { } created)
+            {
+                proto.Created = created;
+            }
+            if (Status is { } status)
+            {
+                proto.Status = (StreamSample.V1.StreamStatus)status;
+            }
+            return proto;
+        }
+
+        /// <summary>
+        /// Converts this model to and from its generated transfer type.
+        /// </summary>
+        public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTransferTypeConverter
+        {
+            /// <summary>Gets the generated transfer type.</summary>
+            public System.Type TransferType => typeof(StreamSample.V1.CreateStreamResponse);
+
+            /// <summary>Converts a model value to its transfer type.</summary>
+            public object? ToTransferType(object? value) => value is null ? null : ((CreateStreamResponse)value).ToTransferType();
+
+            /// <summary>Converts a transfer-type value to this model.</summary>
+            public object? FromTransferType(object? transferType) => transferType is null ? null : CreateStreamResponse.FromTransferType((StreamSample.V1.CreateStreamResponse)transferType);
+        }
+
+    }
+
     [Temporalio.Converters.TemporalTransferTypeConverter(typeof(AppendRecord.TransferTypeConverter))]
     [GeneratedCode("nexgen", null)]
     public record AppendRecord
@@ -286,45 +348,20 @@ public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTrans
 
     }
 
-    [Temporalio.Converters.TemporalTransferTypeConverter(typeof(CreateStreamResult.TransferTypeConverter))]
     [GeneratedCode("nexgen", null)]
     public record CreateStreamResult
     {
-        public StreamRef? Stream { get; init; }
-
-        internal static CreateStreamResult FromTransferType(StreamSample.V1.CreateStreamResponse wire)
-        {
-            return new CreateStreamResult()
-            {
-                Stream = wire.Stream == null ? null : StreamRef.FromTransferType(wire.Stream),
-            };
-        }
-
-        internal StreamSample.V1.CreateStreamResponse ToTransferType()
-        {
-            var proto = new StreamSample.V1.CreateStreamResponse();
-            if (Stream is { } stream)
-            {
-                proto.Stream = (StreamSample.V1.StreamRef)stream.ToTransferType();
-            }
-            return proto;
-        }
-
         /// <summary>
-        /// Converts this model to and from its generated transfer type.
+        /// Initializes a new instance of the <see cref="CreateStreamResult"/> class.
         /// </summary>
-        public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTransferTypeConverter
+        public CreateStreamResult(Stream stream)
         {
-            /// <summary>Gets the generated transfer type.</summary>
-            public System.Type TransferType => typeof(StreamSample.V1.CreateStreamResponse);
-
-            /// <summary>Converts a model value to its transfer type.</summary>
-            public object? ToTransferType(object? value) => value is null ? null : ((CreateStreamResult)value).ToTransferType();
-
-            /// <summary>Converts a transfer-type value to this model.</summary>
-            public object? FromTransferType(object? transferType) => transferType is null ? null : CreateStreamResult.FromTransferType((StreamSample.V1.CreateStreamResponse)transferType);
+            Stream = stream;
         }
 
+        public Stream Stream { get; init; }
+        public bool? Created { get; init; }
+        public StreamStatus? Status { get; init; }
     }
 
     [Temporalio.Converters.TemporalTransferTypeConverter(typeof(DescribeStreamRequest.TransferTypeConverter))]
diff --git a/advanced/samples/dotnet/wit/stream-service/Operations.cs b/advanced/samples/dotnet/wit/stream-service/Operations.cs
index 5f700c6d0..262510dbf 100644
--- a/advanced/samples/dotnet/wit/stream-service/Operations.cs
+++ b/advanced/samples/dotnet/wit/stream-service/Operations.cs
@@ -79,16 +79,16 @@ public static class Operations
         private const string StreamServiceEndpoint = "streams";
 
         [GeneratedCode("nexgen", null)]
-        private static async Task<Stream> CreateStreamAsync(CreateStreamRequest request)
+        private static async Task<CreateStreamResult> CreateStreamAsync(CreateStreamRequest request)
         {
             var client = Workflow.CreateNexusWorkflowClient<IStreamService>(StreamServiceEndpoint);
-            var result = await client.ExecuteNexusOperationAsync<CreateStreamResult>(svc => svc.CreateStream(request)).ConfigureAwait(true);
-            return new Stream((result.Stream ?? throw new System.InvalidOperationException("CreateStream answered without `stream`")));
+            var result = await client.ExecuteNexusOperationAsync<CreateStreamResponse>(svc => svc.CreateStream(request)).ConfigureAwait(true);
+            return new CreateStreamResult(new Stream((result.Stream ?? throw new System.InvalidOperationException("CreateStream answered without `stream`")))) { Created = result.Created, Status = result.Status };
         }
 
         /// <param name="options">Options for the operation.</param>
         [GeneratedCode("nexgen", null)]
-        public static Task<Stream> CreateStreamAsync(CreateStreamOptions? options = null)
+        public static Task<CreateStreamResult> CreateStreamAsync(CreateStreamOptions? options = null)
         {
             var request = new CreateStreamRequest()
             {
diff --git a/advanced/samples/dotnet/wit/stream-service/Services.cs b/advanced/samples/dotnet/wit/stream-service/Services.cs
index 478463df4..4c50e681c 100644
--- a/advanced/samples/dotnet/wit/stream-service/Services.cs
+++ b/advanced/samples/dotnet/wit/stream-service/Services.cs
@@ -19,7 +19,7 @@ internal interface IStreamService
     {
         [GeneratedCode("nexgen", null)]
         [NexusOperation("CreateStream")]
-        CreateStreamResult CreateStream(CreateStreamRequest request);
+        CreateStreamResponse CreateStream(CreateStreamRequest request);
 
         [GeneratedCode("nexgen", null)]
         [NexusOperation("AppendRecords")]
@@ -71,7 +71,7 @@ internal static class NexgenOperationRegistry
 
         internal static IReadOnlyDictionary<(string Service, string Operation), INexgenOperationInfo> Operations { get; } = new Dictionary<(string Service, string Operation), INexgenOperationInfo>
         {
-            [("streamsample.v1.StreamService", "CreateStream")] = new NexgenOperationInfo<CreateStreamRequest, CreateStreamResult>(
+            [("streamsample.v1.StreamService", "CreateStream")] = new NexgenOperationInfo<CreateStreamRequest, CreateStreamResponse>(
                 StreamServiceServiceDefinition.Operations["CreateStream"]),
             [("streamsample.v1.StreamService", "AppendRecords")] = new NexgenOperationInfo<AppendRecordsRequest, AppendRecordsResponse>(
                 StreamServiceServiceDefinition.Operations["AppendRecords"]),
diff --git a/advanced/samples/dotnet/wit/workflow-handle/Models.cs b/advanced/samples/dotnet/wit/workflow-handle/Models.cs
index 60fc0dc8e..77b92b10a 100644
--- a/advanced/samples/dotnet/wit/workflow-handle/Models.cs
+++ b/advanced/samples/dotnet/wit/workflow-handle/Models.cs
@@ -5,7 +5,6 @@
 using System;
 using System.CodeDom.Compiler;
 using System.Collections.Generic;
-using Temporalio.Converters;
 using Nexgen.Support;
 
 namespace Temporalio.Workflows
diff --git a/advanced/samples/inputs/stream-service.wit b/advanced/samples/inputs/stream-service.wit
index 6ad7d10ac..a1d7bb56b 100644
--- a/advanced/samples/inputs/stream-service.wit
+++ b/advanced/samples/inputs/stream-service.wit
@@ -31,6 +31,14 @@ interface stream-service {
     body: option<payload>,
   }
 
+  /// @nexus.proto "streamsample.v1.StreamStatus" typescript-import="#streamsample"
+  enum stream-status {
+    stream-status-unspecified,
+    stream-status-open,
+    stream-status-closed,
+    stream-status-expired,
+  }
+
   /// @nexus.proto "streamsample.v1.CreateStreamRequest" typescript-import="#streamsample"
   record create-stream-request {
     %ref: option<stream-ref>,
@@ -118,9 +126,13 @@ interface stream-service {
     describe: func() -> describe-stream-response;
   }
 
-  // The stream binds from the stream field of the response.
+  // The stream binds from the stream field of the response, and the other fields stay beside it.
   /// @nexus.proto "streamsample.v1.CreateStreamResponse" typescript-import="#streamsample"
-  type create-stream-result = own<%stream>;
+  record create-stream-result {
+    %stream: own<%stream>,
+    created: option<bool>,
+    status: option<stream-status>,
+  }
 
   /// @nexus.operation name="CreateStream"
   create-stream: func(request: create-stream-request) -> create-stream-result;
diff --git a/advanced/samples/invalid/resources/result-record-holder-without-proto/expected.toml b/advanced/samples/invalid/resources/result-record-holder-without-proto/expected.toml
new file mode 100644
index 000000000..0da14ccf4
--- /dev/null
+++ b/advanced/samples/invalid/resources/result-record-holder-without-proto/expected.toml
@@ -0,0 +1,5 @@
+descriptors = "temporal"
+
+[error]
+kind = "invalid-resource"
+contains = ["result record `StartRunResult`", "so it needs `@nexus.proto`"]
diff --git a/advanced/samples/invalid/resources/result-record-holder-without-proto/input.wit b/advanced/samples/invalid/resources/result-record-holder-without-proto/input.wit
new file mode 100644
index 000000000..75a22502f
--- /dev/null
+++ b/advanced/samples/invalid/resources/result-record-holder-without-proto/input.wit
@@ -0,0 +1,23 @@
+package test:holder@1.0.0;
+
+world system {
+  export api;
+}
+
+/// @nexus.endpoint "runs"
+interface api {
+  resource workflow-run {
+    constructor(run-id: option<string>);
+  }
+
+  /// @nexus.proto "temporal.api.workflowservice.v1.GetSystemInfoRequest"
+  record start-run-request {
+  }
+
+  record start-run-result {
+    run: own<workflow-run>,
+    started: option<bool>,
+  }
+
+  start-run: func(request: start-run-request) -> start-run-result;
+}
diff --git a/advanced/samples/python/streamsample/v1/stream_service_pb2.py b/advanced/samples/python/streamsample/v1/stream_service_pb2.py
index 0e5956a09..30bb8138a 100644
--- a/advanced/samples/python/streamsample/v1/stream_service_pb2.py
+++ b/advanced/samples/python/streamsample/v1/stream_service_pb2.py
@@ -26,7 +26,7 @@
 from temporalio.api.common.v1 import message_pb2 as temporal_dot_api_dot_common_dot_v1_dot_message__pb2
 
 
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n$streamsample/v1/stream_service.proto\x12\x0fstreamsample.v1\x1a\x1egoogle/protobuf/duration.proto\x1a$temporal/api/common/v1/message.proto\",\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\"L\n\x0c\x41ppendRecord\x12\r\n\x05topic\x18\x01 \x01(\t\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\"\\\n\x0cStreamRecord\x12\x0e\n\x06offset\x18\x01 \x01(\x03\x12\r\n\x05topic\x18\x02 \x01(\t\x12-\n\x04\x62ody\x18\x03 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\"\x7f\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\'\n\x03ref\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\"S\n\x14\x43reateStreamResponse\x12*\n\x06stream\x18\x01 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08\"\x97\x01\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12\x10\n\x08sequence\x18\x03 \x01(\x03\x12.\n\x07records\x18\x04 \x03(\x0b\x32\x1d.streamsample.v1.AppendRecord\"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03\"\x84\x01\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12/\n\x06result\x18\x03 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\"\x15\n\x13\x43loseStreamResponse\"\xb6\x01\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x0e\n\x06topics\x18\x04 \x03(\t\x12\x13\n\x0bmax_records\x18\x05 \x01(\x05\x12\'\n\x04wait\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\"h\n\x13ReadRecordsResponse\x12.\n\x07records\x18\x01 \x03(\x0b\x32\x1d.streamsample.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08\"V\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\"[\n\x16\x44\x65scribeStreamResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03\x12,\n\tretention\x18\x02 \x01(\x0b\x32\x19.google.protobuf.DurationB\x12\xaa\x02\x0fStreamSample.V1b\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n$streamsample/v1/stream_service.proto\x12\x0fstreamsample.v1\x1a\x1egoogle/protobuf/duration.proto\x1a$temporal/api/common/v1/message.proto\",\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x13\n\x0bworkflow_id\x18\x02 \x01(\t\"L\n\x0c\x41ppendRecord\x12\r\n\x05topic\x18\x01 \x01(\t\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\"\\\n\x0cStreamRecord\x12\x0e\n\x06offset\x18\x01 \x01(\x03\x12\r\n\x05topic\x18\x02 \x01(\t\x12-\n\x04\x62ody\x18\x03 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\"\x7f\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\'\n\x03ref\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\"\x82\x01\n\x14\x43reateStreamResponse\x12*\n\x06stream\x18\x01 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08\x12-\n\x06status\x18\x03 \x01(\x0e\x32\x1d.streamsample.v1.StreamStatus\"\x97\x01\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12\x10\n\x08sequence\x18\x03 \x01(\x03\x12.\n\x07records\x18\x04 \x03(\x0b\x32\x1d.streamsample.v1.AppendRecord\"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03\"\x84\x01\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12/\n\x06result\x18\x03 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\"\x15\n\x13\x43loseStreamResponse\"\xb6\x01\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x0e\n\x06topics\x18\x04 \x03(\t\x12\x13\n\x0bmax_records\x18\x05 \x01(\x05\x12\'\n\x04wait\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\"h\n\x13ReadRecordsResponse\x12.\n\x07records\x18\x01 \x03(\x0b\x32\x1d.streamsample.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08\"V\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12*\n\x06stream\x18\x02 \x01(\x0b\x32\x1a.streamsample.v1.StreamRef\"[\n\x16\x44\x65scribeStreamResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03\x12,\n\tretention\x18\x02 \x01(\x0b\x32\x19.google.protobuf.Duration*z\n\x0cStreamStatus\x12\x1d\n\x19STREAM_STATUS_UNSPECIFIED\x10\x00\x12\x16\n\x12STREAM_STATUS_OPEN\x10\x01\x12\x18\n\x14STREAM_STATUS_CLOSED\x10\x02\x12\x19\n\x15STREAM_STATUS_EXPIRED\x10\x03\x42\x12\xaa\x02\x0fStreamSample.V1b\x06proto3')
 
 _globals = globals()
 _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
@@ -34,6 +34,8 @@
 if not _descriptor._USE_C_DESCRIPTORS:
   _globals['DESCRIPTOR']._loaded_options = None
   _globals['DESCRIPTOR']._serialized_options = b'\252\002\017StreamSample.V1'
+  _globals['_STREAMSTATUS']._serialized_start=1459
+  _globals['_STREAMSTATUS']._serialized_end=1581
   _globals['_STREAMREF']._serialized_start=127
   _globals['_STREAMREF']._serialized_end=171
   _globals['_APPENDRECORD']._serialized_start=173
@@ -42,22 +44,22 @@
   _globals['_STREAMRECORD']._serialized_end=343
   _globals['_CREATESTREAMREQUEST']._serialized_start=345
   _globals['_CREATESTREAMREQUEST']._serialized_end=472
-  _globals['_CREATESTREAMRESPONSE']._serialized_start=474
-  _globals['_CREATESTREAMRESPONSE']._serialized_end=557
-  _globals['_APPENDRECORDSREQUEST']._serialized_start=560
-  _globals['_APPENDRECORDSREQUEST']._serialized_end=711
-  _globals['_APPENDRECORDSRESPONSE']._serialized_start=713
-  _globals['_APPENDRECORDSRESPONSE']._serialized_end=779
-  _globals['_CLOSESTREAMREQUEST']._serialized_start=782
-  _globals['_CLOSESTREAMREQUEST']._serialized_end=914
-  _globals['_CLOSESTREAMRESPONSE']._serialized_start=916
-  _globals['_CLOSESTREAMRESPONSE']._serialized_end=937
-  _globals['_READRECORDSREQUEST']._serialized_start=940
-  _globals['_READRECORDSREQUEST']._serialized_end=1122
-  _globals['_READRECORDSRESPONSE']._serialized_start=1124
-  _globals['_READRECORDSRESPONSE']._serialized_end=1228
-  _globals['_DESCRIBESTREAMREQUEST']._serialized_start=1230
-  _globals['_DESCRIBESTREAMREQUEST']._serialized_end=1316
-  _globals['_DESCRIBESTREAMRESPONSE']._serialized_start=1318
-  _globals['_DESCRIBESTREAMRESPONSE']._serialized_end=1409
+  _globals['_CREATESTREAMRESPONSE']._serialized_start=475
+  _globals['_CREATESTREAMRESPONSE']._serialized_end=605
+  _globals['_APPENDRECORDSREQUEST']._serialized_start=608
+  _globals['_APPENDRECORDSREQUEST']._serialized_end=759
+  _globals['_APPENDRECORDSRESPONSE']._serialized_start=761
+  _globals['_APPENDRECORDSRESPONSE']._serialized_end=827
+  _globals['_CLOSESTREAMREQUEST']._serialized_start=830
+  _globals['_CLOSESTREAMREQUEST']._serialized_end=962
+  _globals['_CLOSESTREAMRESPONSE']._serialized_start=964
+  _globals['_CLOSESTREAMRESPONSE']._serialized_end=985
+  _globals['_READRECORDSREQUEST']._serialized_start=988
+  _globals['_READRECORDSREQUEST']._serialized_end=1170
+  _globals['_READRECORDSRESPONSE']._serialized_start=1172
+  _globals['_READRECORDSRESPONSE']._serialized_end=1276
+  _globals['_DESCRIBESTREAMREQUEST']._serialized_start=1278
+  _globals['_DESCRIBESTREAMREQUEST']._serialized_end=1364
+  _globals['_DESCRIBESTREAMRESPONSE']._serialized_start=1366
+  _globals['_DESCRIBESTREAMRESPONSE']._serialized_end=1457
 # @@protoc_insertion_point(module_scope)
diff --git a/advanced/samples/python/streamsample/v1/stream_service_pb2.pyi b/advanced/samples/python/streamsample/v1/stream_service_pb2.pyi
index f55bb08c8..07b7eba77 100644
--- a/advanced/samples/python/streamsample/v1/stream_service_pb2.pyi
+++ b/advanced/samples/python/streamsample/v1/stream_service_pb2.pyi
@@ -8,6 +8,7 @@ from google.protobuf import descriptor as _descriptor
 from google.protobuf import duration_pb2 as _duration_pb2
 from google.protobuf import message as _message
 from google.protobuf.internal import containers as _containers
+from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper
 from temporalio.api.common.v1 import message_pb2 as _message_pb2
 import builtins as _builtins
 import sys
@@ -20,6 +21,25 @@ else:
 
 DESCRIPTOR: _descriptor.FileDescriptor
 
+class _StreamStatus:
+    ValueType = _typing.NewType("ValueType", _builtins.int)
+    V: _TypeAlias = ValueType  # noqa: Y015
+
+class _StreamStatusEnumTypeWrapper(_enum_type_wrapper._EnumTypeWrapper[_StreamStatus.ValueType], _builtins.type):
+    DESCRIPTOR: _descriptor.EnumDescriptor
+    STREAM_STATUS_UNSPECIFIED: _StreamStatus.ValueType  # 0
+    STREAM_STATUS_OPEN: _StreamStatus.ValueType  # 1
+    STREAM_STATUS_CLOSED: _StreamStatus.ValueType  # 2
+    STREAM_STATUS_EXPIRED: _StreamStatus.ValueType  # 3
+
+class StreamStatus(_StreamStatus, metaclass=_StreamStatusEnumTypeWrapper): ...
+
+STREAM_STATUS_UNSPECIFIED: StreamStatus.ValueType  # 0
+STREAM_STATUS_OPEN: StreamStatus.ValueType  # 1
+STREAM_STATUS_CLOSED: StreamStatus.ValueType  # 2
+STREAM_STATUS_EXPIRED: StreamStatus.ValueType  # 3
+Global___StreamStatus: _TypeAlias = StreamStatus  # noqa: Y015
+
 @_typing.final
 class StreamRef(_message.Message):
     DESCRIPTOR: _descriptor.Descriptor
@@ -124,7 +144,9 @@ class CreateStreamResponse(_message.Message):
 
     STREAM_FIELD_NUMBER: _builtins.int
     CREATED_FIELD_NUMBER: _builtins.int
+    STATUS_FIELD_NUMBER: _builtins.int
     created: _builtins.bool
+    status: Global___StreamStatus.ValueType
     @_builtins.property
     def stream(self) -> Global___StreamRef: ...
     def __init__(
@@ -132,10 +154,11 @@ class CreateStreamResponse(_message.Message):
         *,
         stream: Global___StreamRef | None = ...,
         created: _builtins.bool = ...,
+        status: Global___StreamStatus.ValueType = ...,
     ) -> None: ...
     _HasFieldArgType: _TypeAlias = _typing.Literal["stream", b"stream"]  # noqa: Y015
     def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ...
-    _ClearFieldArgType: _TypeAlias = _typing.Literal["created", b"created", "stream", b"stream"]  # noqa: Y015
+    _ClearFieldArgType: _TypeAlias = _typing.Literal["created", b"created", "status", b"status", "stream", b"stream"]  # noqa: Y015
     def ClearField(self, field_name: _ClearFieldArgType) -> None: ...
     def WhichOneof(self, oneof_group: _Never) -> None: ...
 
diff --git a/advanced/samples/python/wit/stream_service/_resources/__init__.py b/advanced/samples/python/wit/stream_service/_resources/__init__.py
index 06bb70ddd..4cb791aed 100644
--- a/advanced/samples/python/wit/stream_service/_resources/__init__.py
+++ b/advanced/samples/python/wit/stream_service/_resources/__init__.py
@@ -2,8 +2,12 @@
 
 from __future__ import annotations
 
-from .stream import Stream
+from .stream import (
+    Stream,
+    CreateStreamResult,
+)
 
 __all__ = [
     "Stream",
+    "CreateStreamResult",
 ]
diff --git a/advanced/samples/python/wit/stream_service/_resources/stream.py b/advanced/samples/python/wit/stream_service/_resources/stream.py
index 0b0aa9437..280d01a42 100644
--- a/advanced/samples/python/wit/stream_service/_resources/stream.py
+++ b/advanced/samples/python/wit/stream_service/_resources/stream.py
@@ -18,6 +18,7 @@
     ReadRecordsRequest,
     ReadRecordsResponse,
     StreamRef,
+    StreamStatus,
 )
 
 
@@ -76,6 +77,13 @@ async def describe(
         return await handle
 
 
+@dataclasses.dataclass(slots=True)
+class CreateStreamResult:
+    stream: Stream
+    created: bool | None = None
+    status: StreamStatus | None = None
+
+
 async def _append_records(
     request: AppendRecordsRequest,
 ) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]:
diff --git a/advanced/samples/python/wit/stream_service/models.py b/advanced/samples/python/wit/stream_service/models.py
index fcf9313b9..40a75f82c 100644
--- a/advanced/samples/python/wit/stream_service/models.py
+++ b/advanced/samples/python/wit/stream_service/models.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import dataclasses
+import enum
 import typing
 import typing_extensions
 import datetime
@@ -12,6 +13,13 @@
 import temporal_support as _support
 
 
+class StreamStatus(enum.IntEnum):
+    StreamStatusUnspecified = 0
+    StreamStatusOpen = 1
+    StreamStatusClosed = 2
+    StreamStatusExpired = 3
+
+
 @dataclasses.dataclass(slots=True)
 class CreateStreamRequest:
     ref: StreamRef | None = None
@@ -113,13 +121,15 @@ def to_transfer_type(
 
 
 @dataclasses.dataclass(slots=True)
-class CreateStreamResult:
+class CreateStreamResponse:
     stream: StreamRef | None = None
+    created: bool | None = None
+    status: StreamStatus | None = None
 
 
-class _CreateStreamResultTransferTypeConverter(
+class _CreateStreamResponseTransferTypeConverter(
     temporalio.converter.TransferTypeConverter[
-        CreateStreamResult, streamsample.v1.stream_service_pb2.CreateStreamResponse
+        CreateStreamResponse, streamsample.v1.stream_service_pb2.CreateStreamResponse
     ]
 ):
     transfer_type: (
@@ -130,32 +140,40 @@ class _CreateStreamResultTransferTypeConverter(
     def from_transfer_type(
         self,
         value: streamsample.v1.stream_service_pb2.CreateStreamResponse,
-        type_hint: type[CreateStreamResult],
-    ) -> CreateStreamResult:
-        return CreateStreamResult(
+        type_hint: type[CreateStreamResponse],
+    ) -> CreateStreamResponse:
+        return CreateStreamResponse(
             stream=_StreamRefTransferTypeConverter().from_transfer_type(
                 value.stream, StreamRef
             )
             if value.HasField("stream")
             else None,
+            created=value.created if bool(value.created) else None,
+            status=StreamStatus(value.status) if value.status != 0 else None,
         )
 
     @typing_extensions.override
     def to_transfer_type(
         self,
-        value: CreateStreamResult,
+        value: CreateStreamResponse,
     ) -> streamsample.v1.stream_service_pb2.CreateStreamResponse:
         message = streamsample.v1.stream_service_pb2.CreateStreamResponse()
         if value.stream is not None:
             message.stream.CopyFrom(
                 _StreamRefTransferTypeConverter().to_transfer_type(value.stream)
             )
+        if value.created is not None:
+            message.created = value.created
+        if value.status is not None:
+            message.status = streamsample.v1.stream_service_pb2.StreamStatus.ValueType(
+                value.status
+            )
         return message
 
 
 temporalio.converter.transfer_type_convertible(
-    _CreateStreamResultTransferTypeConverter
-)(CreateStreamResult)  # pyright: ignore[reportUnusedCallResult]
+    _CreateStreamResponseTransferTypeConverter
+)(CreateStreamResponse)  # pyright: ignore[reportUnusedCallResult]
 
 
 @dataclasses.dataclass(slots=True)
diff --git a/advanced/samples/python/wit/stream_service/operations/create_stream.py b/advanced/samples/python/wit/stream_service/operations/create_stream.py
index 842fecf70..9bfb27a03 100644
--- a/advanced/samples/python/wit/stream_service/operations/create_stream.py
+++ b/advanced/samples/python/wit/stream_service/operations/create_stream.py
@@ -7,15 +7,18 @@
 
 from ..models import (
     CreateStreamRequest,
-    CreateStreamResult,
+    CreateStreamResponse,
     StreamRef,
 )
-from .._resources import Stream
+from .._resources import (
+    CreateStreamResult,
+    Stream,
+)
 
 
 async def _create_stream(
     request: CreateStreamRequest,
-) -> Stream:
+) -> CreateStreamResult:
     nexus_client = temporalio.workflow.create_nexus_client(
         service="streamsample.v1.StreamService",
         endpoint="streams",
@@ -23,13 +26,17 @@ async def _create_stream(
     handle = await nexus_client.start_operation(
         operation="CreateStream",
         input=request,
-        output_type=CreateStreamResult,
+        output_type=CreateStreamResponse,
     )
     result = await handle
     if result.stream is None:
         raise RuntimeError("CreateStream answered without `stream`")
-    return Stream(
-        stream=result.stream,
+    return CreateStreamResult(
+        stream=Stream(
+            stream=result.stream,
+        ),
+        created=result.created,
+        status=result.status,
     )
 
 
@@ -37,7 +44,7 @@ async def create_stream(
     *,
     ref: StreamRef | None = None,
     retention: datetime.timedelta | None = None,
-) -> Stream:
+) -> CreateStreamResult:
     request = CreateStreamRequest(
         ref=ref,
         retention=retention,
diff --git a/advanced/samples/python/wit/stream_service/services.py b/advanced/samples/python/wit/stream_service/services.py
index 4b76e9d3f..3ad9eadb2 100644
--- a/advanced/samples/python/wit/stream_service/services.py
+++ b/advanced/samples/python/wit/stream_service/services.py
@@ -9,7 +9,7 @@
     CloseStreamRequest,
     CloseStreamResponse,
     CreateStreamRequest,
-    CreateStreamResult,
+    CreateStreamResponse,
     DescribeStreamRequest,
     DescribeStreamResponse,
     ReadRecordsRequest,
@@ -21,7 +21,7 @@
 class StreamService:
     create_stream: Operation[
         CreateStreamRequest,
-        CreateStreamResult,
+        CreateStreamResponse,
     ] = Operation(name="CreateStream")
 
     append_records: Operation[
diff --git a/advanced/samples/typescript/proto/streamsample.d.ts b/advanced/samples/typescript/proto/streamsample.d.ts
index e1756542f..27fc59bae 100644
--- a/advanced/samples/typescript/proto/streamsample.d.ts
+++ b/advanced/samples/typescript/proto/streamsample.d.ts
@@ -27,9 +27,17 @@ export namespace streamsample {
       retention?: google.protobuf.IDuration | null;
     }
 
+    enum StreamStatus {
+      STREAM_STATUS_UNSPECIFIED = 0,
+      STREAM_STATUS_OPEN = 1,
+      STREAM_STATUS_CLOSED = 2,
+      STREAM_STATUS_EXPIRED = 3,
+    }
+
     interface ICreateStreamResponse {
       stream?: IStreamRef | null;
       created?: boolean | null;
+      status?: StreamStatus | null;
     }
 
     interface IAppendRecordsRequest {
diff --git a/advanced/samples/typescript/wit/stream-service/models.ts b/advanced/samples/typescript/wit/stream-service/models.ts
index 947cd35e9..8b4262595 100644
--- a/advanced/samples/typescript/wit/stream-service/models.ts
+++ b/advanced/samples/typescript/wit/stream-service/models.ts
@@ -4,6 +4,7 @@ import type { streamsample } from "#streamsample";
 import type * as common from "@temporalio/common";
 import type Long from "long";
 import * as support from "../../support/temporal_model_converters";
+import type { Stream } from "./resources";
 
 function requiredField<T>(
   value: T | null | undefined,
@@ -16,33 +17,49 @@ function requiredField<T>(
   return value;
 }
 
-export interface CreateStreamResult {
+export enum StreamStatus {
+  StreamStatusUnspecified = 0,
+  StreamStatusOpen = 1,
+  StreamStatusClosed = 2,
+  StreamStatusExpired = 3,
+}
+
+export interface CreateStreamResponse {
   stream?: StreamRef;
+  created?: boolean;
+  status?: StreamStatus;
 }
 
-function createStreamResultFromProto(
+function createStreamResponseFromProto(
   proto: streamsample.v1.ICreateStreamResponse,
-): CreateStreamResult {
+): CreateStreamResponse {
   return {
     stream: proto.stream == null ? undefined : streamRefFromProto(proto.stream),
+    created: proto.created ?? undefined,
+    status: proto.status == null ? undefined : (proto.status as number as StreamStatus),
   };
 }
 
-function createStreamResultToProto(
-  model: CreateStreamResult,
+function createStreamResponseToProto(
+  model: CreateStreamResponse,
 ): streamsample.v1.ICreateStreamResponse {
   return {
     stream: model.stream == null ? undefined : streamRefToProto(model.stream),
+    created: model.created,
+    status:
+      model.status == null
+        ? undefined
+        : (model.status as number as streamsample.v1.StreamStatus),
   };
 }
 
-export const createStreamResultTransferTypeConverter = {
-  fromTransferType(value: streamsample.v1.ICreateStreamResponse): CreateStreamResult {
-    return createStreamResultFromProto(value);
+export const createStreamResponseTransferTypeConverter = {
+  fromTransferType(value: streamsample.v1.ICreateStreamResponse): CreateStreamResponse {
+    return createStreamResponseFromProto(value);
   },
 
-  toTransferType(value: CreateStreamResult): streamsample.v1.ICreateStreamResponse {
-    return createStreamResultToProto(value);
+  toTransferType(value: CreateStreamResponse): streamsample.v1.ICreateStreamResponse {
+    return createStreamResponseToProto(value);
   },
 };
 export interface StreamRef {
@@ -443,3 +460,8 @@ export const describeStreamRequestTransferTypeConverter = {
     return describeStreamRequestToProto(value);
   },
 };
+export interface CreateStreamResult {
+  stream: Stream;
+  created?: boolean;
+  status?: StreamStatus;
+}
diff --git a/advanced/samples/typescript/wit/stream-service/operations/create-stream.ts b/advanced/samples/typescript/wit/stream-service/operations/create-stream.ts
index 1ffdca0db..bf8482e1a 100644
--- a/advanced/samples/typescript/wit/stream-service/operations/create-stream.ts
+++ b/advanced/samples/typescript/wit/stream-service/operations/create-stream.ts
@@ -2,7 +2,7 @@
 
 import * as workflow from "@temporalio/workflow";
 import { streamService } from "../services";
-import type { CreateStreamRequest } from "../models";
+import type { CreateStreamRequest, CreateStreamResult } from "../models";
 import { Stream } from "../resources";
 
 function requiredField<T>(
@@ -19,7 +19,9 @@ function requiredField<T>(
 /**
  * @param request - Request for the operation.
  */
-export async function createStream(request: CreateStreamRequest): Promise<Stream> {
+export async function createStream(
+  request: CreateStreamRequest,
+): Promise<CreateStreamResult> {
   const client = workflow.createNexusServiceClient({
     service: streamService,
     endpoint: "streams",
@@ -29,5 +31,9 @@ export async function createStream(request: CreateStreamRequest): Promise<Stream
     request,
   );
   const result = await handle.result();
-  return new Stream(requiredField(result.stream, "resource", "stream"));
+  return {
+    stream: new Stream(requiredField(result.stream, "resource", "stream")),
+    created: result.created,
+    status: result.status,
+  };
 }
diff --git a/advanced/samples/typescript/wit/stream-service/services.ts b/advanced/samples/typescript/wit/stream-service/services.ts
index 7dbef6e3b..e2adb9914 100644
--- a/advanced/samples/typescript/wit/stream-service/services.ts
+++ b/advanced/samples/typescript/wit/stream-service/services.ts
@@ -2,7 +2,7 @@
 
 import * as nexus from "nexus-rpc";
 import {
-  createStreamResultTransferTypeConverter,
+  createStreamResponseTransferTypeConverter,
   createStreamRequestTransferTypeConverter,
   appendRecordsResponseTransferTypeConverter,
   appendRecordsRequestTransferTypeConverter,
@@ -14,7 +14,7 @@ import {
   describeStreamRequestTransferTypeConverter,
 } from "./models";
 import type {
-  CreateStreamResult,
+  CreateStreamResponse,
   CreateStreamRequest,
   AppendRecordsResponse,
   AppendRecordsRequest,
@@ -27,10 +27,10 @@ import type {
 } from "./models";
 
 export const streamService = nexus.service("streamsample.v1.StreamService", {
-  createStream: nexus.operation<CreateStreamRequest, CreateStreamResult>({
+  createStream: nexus.operation<CreateStreamRequest, CreateStreamResponse>({
     name: "CreateStream",
     inputType: { transferTypeConverter: createStreamRequestTransferTypeConverter },
-    outputType: { transferTypeConverter: createStreamResultTransferTypeConverter },
+    outputType: { transferTypeConverter: createStreamResponseTransferTypeConverter },
   }),
   appendRecords: nexus.operation<AppendRecordsRequest, AppendRecordsResponse>({
     name: "AppendRecords",
diff --git a/changelog/added/holding-heron.md b/changelog/added/holding-heron.md
new file mode 100644
index 000000000..5fb767524
--- /dev/null
+++ b/changelog/added/holding-heron.md
@@ -0,0 +1 @@
+An operation can return a proto-backed record that holds a resource in one `own<T>` field, in Python, TypeScript, .NET and Go. The resource binds from the record's message, the other fields come from the same message, and the message travels as a wire record with the same name.
diff --git a/src/generator/dotnet.rs b/src/generator/dotnet.rs
index d382268f5..939bd47fa 100644
--- a/src/generator/dotnet.rs
+++ b/src/generator/dotnet.rs
@@ -15,7 +15,8 @@ use crate::planning::{
     PlannedSpec, PlannedType, operation_input_model,
 };
 use crate::planning::{
-    RequestPlan, RequestPlanField, RequestPlanSource, ResolvedResourceBindingSource,
+    PlannedResourceHolderField, RequestPlan, RequestPlanField, RequestPlanSource,
+    ResolvedResourceBindingSource,
 };
 use crate::spec::{
     AliasTypeSpec, EnumSpec, ExternalTypeSourceSpec, ExternalTypeSpec, FlagsSpec,
@@ -904,11 +905,14 @@ impl<'a> ApiPlanner<'a> {
             let expression =
                 operation_transform_expression(operation, "request", "result", self.api_plan);
             output.push_str("        return ");
-            if let Some(resource) = &operation.data.output_resource_return {
-                output.push_str(&csharp_type_name(&resource.resource_type_name));
-                output.push_str(".BindClient(");
-                output.push_str(&expression);
-                output.push_str(", _client)");
+            if operation.data.output_resource_return.is_some() {
+                output.push_str(&resource_return_expression(
+                    operation,
+                    "request",
+                    "result",
+                    self.api_plan,
+                    Some("_client"),
+                ));
             } else {
                 output.push_str(&expression);
             }
@@ -1108,11 +1112,14 @@ impl<'a> ApiPlanner<'a> {
             let expression =
                 operation_transform_expression(operation, "request", "result", self.api_plan);
             output.push_str("        return ");
-            if endpoint_parameter && let Some(resource) = &operation.data.output_resource_return {
-                output.push_str(&csharp_type_name(&resource.resource_type_name));
-                output.push_str(".BindClient(");
-                output.push_str(&expression);
-                output.push_str(", client)");
+            if endpoint_parameter && operation.data.output_resource_return.is_some() {
+                output.push_str(&resource_return_expression(
+                    operation,
+                    "request",
+                    "result",
+                    self.api_plan,
+                    Some("client"),
+                ));
             } else {
                 output.push_str(&expression);
             }
@@ -1196,11 +1203,14 @@ impl<'a> ApiPlanner<'a> {
             let expression =
                 operation_transform_expression(operation, "request", "result", self.api_plan);
             output.push_str("        return ");
-            if let Some(resource) = &operation.data.output_resource_return {
-                output.push_str(&csharp_type_name(&resource.resource_type_name));
-                output.push_str(".BindClient(");
-                output.push_str(&expression);
-                output.push_str(", RequireNexgenClient())");
+            if operation.data.output_resource_return.is_some() {
+                output.push_str(&resource_return_expression(
+                    operation,
+                    "request",
+                    "result",
+                    self.api_plan,
+                    Some("RequireNexgenClient()"),
+                ));
             } else {
                 output.push_str(&expression);
             }
@@ -1306,7 +1316,12 @@ impl<'a> ApiPlanner<'a> {
             return type_name.to_string();
         }
         if let Some(resource) = &operation.data.output_resource_return {
-            return csharp_type_name(&resource.resource_type_name);
+            return csharp_type_name(
+                resource
+                    .holder
+                    .as_ref()
+                    .map_or(&resource.resource_type_name, |holder| &holder.model_name),
+            );
         }
         self.operation_output_type(operation)
     }
@@ -3642,42 +3657,93 @@ fn operation_transform_expression(
             .replace("request", request_expr)
             .replace("result", result_expr);
     }
-    if let Some(resource) = &operation.data.output_resource_return {
-        let args = resource
-            .bindings
-            .iter()
-            .map(|binding| {
-                let expr = resource_return_binding_expr(
-                    resource,
-                    binding,
-                    request_expr,
-                    result_expr,
-                    api_plan,
-                );
-                // A response model field is always nullable, so a required one is checked.
-                match &binding.source {
-                    ResolvedResourceBindingSource::ResultField {
-                        proto_field_name, ..
-                    } if !binding.optional => format!(
-                        "({expr} ?? throw new System.InvalidOperationException({}))",
-                        csharp_string_literal(&format!(
-                            "{} answered without `{proto_field_name}`",
-                            operation.wire_name
-                        ))
-                    ),
-                    _ => expr,
-                }
-            })
-            .collect::<Vec<_>>();
-        return format!(
-            "new {}({})",
-            csharp_type_name(&resource.resource_type_name),
-            args.join(", ")
-        );
+    if operation.data.output_resource_return.is_some() {
+        return resource_return_expression(operation, request_expr, result_expr, api_plan, None);
     }
     result_expr.to_string()
 }
 
+/// The resource that an operation returns, bound to `client` when one is given. A result
+/// record that holds the resource gets it in its resource field, and each other field from
+/// the field of the same name in the wire record.
+fn resource_return_expression(
+    operation: &PlannedOperation,
+    request_expr: &str,
+    result_expr: &str,
+    api_plan: &PlannedSpec,
+    client: Option<&str>,
+) -> String {
+    let resource = operation
+        .data
+        .output_resource_return
+        .as_ref()
+        .expect("the operation returns a resource");
+    let type_name = csharp_type_name(&resource.resource_type_name);
+    let args = resource
+        .bindings
+        .iter()
+        .map(|binding| {
+            let expr = resource_return_binding_expr(
+                resource,
+                binding,
+                request_expr,
+                result_expr,
+                api_plan,
+            );
+            // A response model field is always nullable, so a required one is checked.
+            match &binding.source {
+                ResolvedResourceBindingSource::ResultField {
+                    proto_field_name, ..
+                } if !binding.optional => format!(
+                    "({expr} ?? throw new System.InvalidOperationException({}))",
+                    csharp_string_literal(&format!(
+                        "{} answered without `{proto_field_name}`",
+                        operation.wire_name
+                    ))
+                ),
+                _ => expr,
+            }
+        })
+        .collect::<Vec<_>>();
+    let mut value = format!("new {type_name}({})", args.join(", "));
+    if let Some(client) = client {
+        value = format!("{type_name}.BindClient({value}, {client})");
+    }
+    let Some(holder) = &resource.holder else {
+        return value;
+    };
+    let field_value = |field: &PlannedResourceHolderField| {
+        if field.resource {
+            value.clone()
+        } else {
+            format!("{result_expr}.{}", csharp_type_name(&field.name))
+        }
+    };
+    let required = holder
+        .fields
+        .iter()
+        .filter(|field| field.required)
+        .map(field_value)
+        .collect::<Vec<_>>();
+    let optional = holder
+        .fields
+        .iter()
+        .filter(|field| !field.required)
+        .map(|field| format!("{} = {}", csharp_type_name(&field.name), field_value(field)))
+        .collect::<Vec<_>>();
+    let mut expression = format!(
+        "new {}({})",
+        csharp_type_name(&holder.model_name),
+        required.join(", ")
+    );
+    if !optional.is_empty() {
+        expression.push_str(" { ");
+        expression.push_str(&optional.join(", "));
+        expression.push_str(" }");
+    }
+    expression
+}
+
 fn resource_return_binding_expr(
     resource: &crate::planning::PlannedOperationResourceReturn,
     binding: &PlannedOperationResourceFieldBinding,
diff --git a/src/generator/go.rs b/src/generator/go.rs
index cc61b1a5c..0e493f490 100644
--- a/src/generator/go.rs
+++ b/src/generator/go.rs
@@ -472,7 +472,13 @@ fn planned_value_type(value_type: &PlannedType, spec: &PlannedSpec) -> PlannedVa
             info: local_type_info(&variant_type.full_name),
             name: variant_type.name.clone(),
         }),
-        TypeSpec::Resource(_) => PlannedValueType::Unknown,
+        TypeSpec::Resource(resource) => PlannedValueType::External {
+            type_name: SelectedTextSpec {
+                value: Some(resource.type_name.clone()),
+                import: None,
+            },
+            fallback: Box::new(PlannedValueType::Unknown),
+        },
         TypeSpec::External(ExternalTypeSpec::Proto(PlannedProtoType::Message(message))) => {
             PlannedValueType::Message(PlannedMessageType {
                 info: message.proto.clone(),
@@ -2633,6 +2639,22 @@ impl GoVisibilityMarker<'_> {
             }
             PlannedOperationOutput::Resource { .. } | PlannedOperationOutput::None => {}
         }
+        // A result record that holds a resource is what the caller gets.
+        if let Some(holder) = operation
+            .data
+            .output_resource_return
+            .as_ref()
+            .and_then(|resource_return| resource_return.holder.as_ref())
+            && self.public_keys.insert(holder.record_full_name.clone())
+            && let Some(record) = self.api_plan.record(&holder.record_full_name)
+        {
+            for (_, field) in record.public_fields() {
+                self.mark_operation_field_kind(
+                    &planned_field_kind(&field.field_type, self.api_plan),
+                    field.name.as_str(),
+                );
+            }
+        }
     }
 
     fn mark_resource_public_surface(&mut self, resource: &PlannedResource) {
diff --git a/src/generator/proto/go.rs b/src/generator/proto/go.rs
index 85898226f..2b7b8f3a5 100644
--- a/src/generator/proto/go.rs
+++ b/src/generator/proto/go.rs
@@ -814,6 +814,9 @@ struct RenderedResourceReturn {
     local_lines: Vec<String>,
     field_initializers: Vec<RenderedResourceFieldInitializer>,
     requires_data_converter: bool,
+    /// The result record that holds the resource: its type, the resource field, and each
+    /// other field with its value.
+    holder: Option<(String, String, Vec<(String, String)>)>,
 }
 
 #[derive(Debug)]
@@ -1285,20 +1288,38 @@ pub(in crate::generator) fn render_operation_function_proto(
                 output.push_str("\t\t\treturn\n");
                 output.push_str("\t\t}\n");
                 for line in &resource_return.local_lines {
+                    // The lines run in the future's goroutine, which sets the error.
+                    if line.trim() == "return nil, err" {
+                        let indent = &line[..line.len() - line.trim_start().len()];
+                        output.push_str(&format!(
+                            "\t\t{indent}resultSettable.SetError(err)\n\t\t{indent}return\n"
+                        ));
+                        continue;
+                    }
                     output.push_str("\t\t");
                     output.push_str(line);
                     output.push('\n');
                 }
-                output.push_str("\t\tvalue := New");
-                output.push_str(&resource_return.resource_type_name);
-                output.push('(');
+                let mut resource = format!("New{}(", resource_return.resource_type_name);
                 for (index, initializer) in resource_return.field_initializers.iter().enumerate() {
                     if index > 0 {
-                        output.push_str(", ");
+                        resource.push_str(", ");
+                    }
+                    resource.push_str(&initializer.expr);
+                }
+                resource.push(')');
+                output.push_str("\t\tvalue := ");
+                match &resource_return.holder {
+                    Some((type_name, resource_field, fields)) => {
+                        output.push_str(&format!("{type_name}{{{resource_field}: {resource}"));
+                        for (name, expr) in fields {
+                            output.push_str(&format!(", {name}: {expr}"));
+                        }
+                        output.push('}');
                     }
-                    output.push_str(&initializer.expr);
+                    None => output.push_str(&resource),
                 }
-                output.push_str(")\n");
+                output.push('\n');
             },
         );
     } else {
@@ -1435,11 +1456,67 @@ fn build_rendered_resource_return(
         field_initializers.push(RenderedResourceFieldInitializer { expr });
     }
 
+    let holder = match &resource_return.holder {
+        Some(holder) => {
+            let record = api_plan
+                .record(&holder.record_full_name)
+                .expect("the plan has the holder record");
+            let mut resource_field = String::new();
+            let mut fields = Vec::new();
+            for field in &holder.fields {
+                if field.resource {
+                    resource_field = go_field_name(&field.name);
+                    continue;
+                }
+                let planned = &record.fields[&field.proto_name];
+                let resource_field = PlannedResourceField {
+                    name: field.name.clone(),
+                    optional: !field.required,
+                    kind: planned.field_type.clone(),
+                    function: None,
+                };
+                let (lines, expr, uses_data_converter) = resource_return_proto_field_source(
+                    &resource_field,
+                    "result",
+                    &field.proto_name,
+                    api_plan,
+                    backend,
+                    package,
+                )
+                .map_err(|reason| Error::UnsupportedGoProtoConversion {
+                    context: format!("result record field `{}.{}`", holder.model_name, field.name),
+                    reason,
+                })?;
+                requires_data_converter |= uses_data_converter;
+                local_lines.extend(lines);
+                // An optional scalar field of the record is a pointer, and the zero value
+                // means unset, as in the record's model.
+                let field_kind = planned_field_kind(&resource_field.kind, api_plan);
+                let field_type = resolve_resource_field_kind(&field_kind, true, package);
+                let expr = match public_default_punning_zero_for_field(&field_kind, &field_type) {
+                    Some(zero) if !field.required => {
+                        let local = format!("held{}", go_field_name(&field.name));
+                        local_lines.push(format!("var {local} {field_type}"));
+                        local_lines.push(format!("if value := {expr}; value != {zero} {{"));
+                        local_lines.push(format!("\t{local} = &value"));
+                        local_lines.push("}".to_string());
+                        local
+                    }
+                    _ => expr,
+                };
+                fields.push((go_field_name(&field.name), expr));
+            }
+            Some((holder.model_name.clone(), resource_field, fields))
+        }
+        None => None,
+    };
+
     Ok(RenderedResourceReturn {
         resource_type_name: resource_return.resource_type_name.clone(),
         local_lines,
         field_initializers,
         requires_data_converter,
+        holder,
     })
 }
 
diff --git a/src/generator/python.rs b/src/generator/python.rs
index 613a2f8a2..f83bc0b85 100644
--- a/src/generator/python.rs
+++ b/src/generator/python.rs
@@ -15,9 +15,9 @@ use crate::language::Language;
 use crate::planning::{
     PlannedFamily, PlannedOperationResourceFieldBinding, PlannedOperationResourceReturn,
     PlannedProtoType, PlannedProtoTypeInfo, PlannedRecordType, PlannedResource,
-    PlannedResourceField, PlannedResourceMethod, PlannedResourceMethodBindingSpec,
-    PlannedResourceMethodResultKind, PlannedSpec, PlannedType, message_model_name,
-    operation_input_model, operation_output_direct_result,
+    PlannedResourceField, PlannedResourceHolder, PlannedResourceMethod,
+    PlannedResourceMethodBindingSpec, PlannedResourceMethodResultKind, PlannedSpec, PlannedType,
+    message_model_name, operation_input_model, operation_output_direct_result,
 };
 use crate::planning::{RequestPlan, ResolvedResourceBindingSource};
 use crate::spec::{
@@ -524,7 +524,12 @@ impl<'a> ApiPlanner<'a> {
             });
             self.resolve_message_value_conversion(&model_type)?;
         }
-        let model_refs = self.models.values().collect::<Vec<_>>();
+        let holders = resource_holders(api_plan);
+        let model_refs = self
+            .models
+            .values()
+            .filter(|model| !holders.contains_key(&model.full_name))
+            .collect::<Vec<_>>();
         let variant_refs = self.variants.values().collect::<Vec<_>>();
         let model_fragments =
             self.render_model_fragments(model_refs.as_slice(), variant_refs.as_slice())?;
@@ -602,6 +607,11 @@ impl<'a> ApiPlanner<'a> {
                     .iter()
                     .map(|resource| resource.type_name.clone())
             })
+            .chain(
+                resource_holders(self.api_plan)
+                    .into_values()
+                    .map(|(model_name, _)| model_name),
+            )
             .collect::<BTreeSet<_>>()
             .into_iter()
             .collect::<Vec<_>>();
@@ -684,7 +694,7 @@ impl<'a> ApiPlanner<'a> {
         if !resource_names.is_empty() {
             files.insert(
                 "_resources/__init__.py",
-                render_resources_package_init(services),
+                render_resources_package_init(services, self.api_plan),
                 GeneratedFileOrigin::fixed("generated Python resources package initializer"),
             )?;
         }
@@ -814,6 +824,17 @@ impl<'a> ApiPlanner<'a> {
             body.push_str("\n\n");
         }
         self.render_resource(&mut body, service, resource, bound_operations);
+        for (full_name, (_, resource_type)) in resource_holders(self.api_plan) {
+            if resource_type != resource.type_name {
+                continue;
+            }
+            let model = &self.models[&full_name];
+            body.push_str("\n\n");
+            render_record_model(&mut body, model, None);
+            for field in &model.fields {
+                module_imports.extend(field.imports.module_imports.iter().cloned());
+            }
+        }
         if !bound_operations.is_empty() && service.endpoint.is_some() {
             body.push_str("\n\n");
             for (index, bound_operation) in bound_operations.iter().enumerate() {
@@ -872,9 +893,15 @@ impl<'a> ApiPlanner<'a> {
                 self.model_hoists,
             );
         }
+        let own_names = resource_holders(self.api_plan)
+            .into_values()
+            .filter(|(_, resource_type)| *resource_type == resource.type_name)
+            .map(|(model_name, _)| model_name)
+            .chain(std::iter::once(resource.type_name.clone()))
+            .collect::<BTreeSet<_>>();
         let used_resource_names = used_python_symbol_imports(&body, resource_names)
             .into_iter()
-            .filter(|name| name != &resource.type_name)
+            .filter(|name| !own_names.contains(name))
             .collect::<Vec<_>>();
         if !used_resource_names.is_empty() {
             render_named_python_import(&mut output, ".", &used_resource_names);
@@ -1252,9 +1279,13 @@ impl<'a> ApiPlanner<'a> {
         let overload_output_annotation = output_transform
             .and_then(|transform| transform.type_name.value().map(str::to_string))
             .or_else(|| {
-                output_resource_return
-                    .as_ref()
-                    .map(|resource| resource.resource_type_name.clone())
+                output_resource_return.as_ref().map(|resource| {
+                    resource
+                        .holder
+                        .as_ref()
+                        .map(|holder| holder.model_name.clone())
+                        .unwrap_or_else(|| resource.resource_type_name.clone())
+                })
             })
             .unwrap_or_else(|| output_annotation_default.clone());
         let unpacked_input = if let (Some(input), Some(rendered_input)) = (input, &rendered_input) {
@@ -3962,11 +3993,48 @@ fn render_models_module(
     Ok(Some(output))
 }
 
-fn render_resources_package_init(services: &[RenderedService<'_>]) -> String {
+/// The result records that hold a resource, by full name, with the model name and the
+/// resource type. Each one lives in its resource's module, since the resource module imports
+/// the models and the record uses the resource class.
+fn resource_holders(api_plan: &PlannedSpec) -> BTreeMap<String, (String, String)> {
+    api_plan
+        .services
+        .iter()
+        .flat_map(|service| service.operations.iter())
+        .filter_map(|operation| operation.data.output_resource_return.as_ref())
+        .filter_map(|resource_return| {
+            resource_return.holder.as_ref().map(|holder| {
+                (
+                    holder.record_full_name.clone(),
+                    (
+                        holder.model_name.clone(),
+                        resource_return.resource_type_name.clone(),
+                    ),
+                )
+            })
+        })
+        .collect()
+}
+
+fn render_resources_package_init(
+    services: &[RenderedService<'_>],
+    api_plan: &PlannedSpec,
+) -> String {
     let resources = services
         .iter()
         .flat_map(|service| service.resources.iter())
         .collect::<Vec<_>>();
+    let holders = resource_holders(api_plan);
+    let names_of = |resource: &PlannedResource| {
+        std::iter::once(resource.type_name.clone())
+            .chain(
+                holders
+                    .values()
+                    .filter(|(_, resource_type)| *resource_type == resource.type_name)
+                    .map(|(model_name, _)| model_name.clone()),
+            )
+            .collect::<Vec<_>>()
+    };
     let mut output = String::new();
     render_generated_file_header(&mut output);
     output.push('\n');
@@ -3974,14 +4042,16 @@ fn render_resources_package_init(services: &[RenderedService<'_>]) -> String {
         render_named_python_import(
             &mut output,
             &format!(".{}", resource_module_name(resource)),
-            std::slice::from_ref(&resource.type_name),
+            &names_of(resource),
         );
     }
     output.push_str("\n__all__ = [\n");
     for resource in &resources {
-        output.push_str("    ");
-        output.push_str(&python_string_literal(&resource.type_name));
-        output.push_str(",\n");
+        for name in names_of(resource) {
+            output.push_str("    ");
+            output.push_str(&python_string_literal(&name));
+            output.push_str(",\n");
+        }
     }
     output.push_str("]\n");
     output
@@ -6874,6 +6944,10 @@ fn render_request_only_operation_function(
     } else if let Some(resource_return) = &operation.output_resource_return {
         output.push_str("    result = await handle\n");
         render_resource_return_guards(output, "    ", operation, resource_return);
+        if let Some(holder) = &resource_return.holder {
+            render_resource_holder_return(output, resource_return, holder);
+            return;
+        }
         if service.endpoint.is_none() {
             output.push_str("    resource = ");
         } else {
@@ -6896,6 +6970,41 @@ fn render_request_only_operation_function(
     }
 }
 
+/// The holder record gets the resource in its resource field, and each other field from the
+/// field of the same name in the wire record.
+fn render_resource_holder_return(
+    output: &mut String,
+    resource_return: &PlannedOperationResourceReturn,
+    holder: &PlannedResourceHolder,
+) {
+    output.push_str("    return ");
+    output.push_str(&holder.model_name);
+    output.push_str("(\n");
+    for field in &holder.fields {
+        let name = python_field_name(&field.name);
+        output.push_str("        ");
+        output.push_str(&name);
+        if !field.resource {
+            output.push_str("=result.");
+            output.push_str(&name);
+            output.push_str(",\n");
+            continue;
+        }
+        output.push('=');
+        output.push_str(&resource_return.resource_type_name);
+        output.push_str("(\n");
+        for binding in &resource_return.bindings {
+            output.push_str("            ");
+            output.push_str(&python_field_name(&binding.field_name));
+            output.push('=');
+            output.push_str(&resource_return_binding_expr_python(binding));
+            output.push_str(",\n");
+        }
+        output.push_str("        ),\n");
+    }
+    output.push_str("    )\n");
+}
+
 fn render_unpacked_operation_docstring(
     output: &mut String,
     operation: &RenderedOperation<'_>,
diff --git a/src/generator/typescript.rs b/src/generator/typescript.rs
index b64182311..254d22d99 100644
--- a/src/generator/typescript.rs
+++ b/src/generator/typescript.rs
@@ -517,9 +517,13 @@ impl<'a> ApiPlanner<'a> {
             output_annotation: output_transform
                 .and_then(|transform| transform.type_name.value().map(str::to_string))
                 .or_else(|| {
-                    output_resource_return
-                        .as_ref()
-                        .map(|resource| resource.resource_type_name.clone())
+                    output_resource_return.as_ref().map(|resource| {
+                        resource
+                            .holder
+                            .as_ref()
+                            .map_or(&resource.resource_type_name, |holder| &holder.model_name)
+                            .clone()
+                    })
                 })
                 .unwrap_or(output_annotation_default),
             output_transform_type_import: output_transform
@@ -3891,6 +3895,16 @@ fn render_endpoint_service_operation_body(
         output.push_str("    return ");
         output.push_str(transform_expr);
         output.push_str(";\n");
+    } else if let Some(resource_return) = &operation.output_resource_return
+        && resource_return.holder.is_some()
+    {
+        output.push_str("    const result = await handle.result();\n");
+        output.push_str("    return ");
+        output.push_str(&typescript_resource_return_value(
+            resource_return,
+            Some("this.client"),
+        ));
+        output.push_str(";\n");
     } else if let Some(resource_return) = &operation.output_resource_return {
         output.push_str("    const result = await handle.result();\n");
         output.push_str("    return ");
@@ -4270,6 +4284,20 @@ fn render_models_module(
         api_plan,
         &body,
     );
+    // A result record that holds a resource uses its class.
+    let held_resources = api_plan
+        .services
+        .iter()
+        .flat_map(|service| service.operations.iter())
+        .filter_map(|operation| operation.data.output_resource_return.as_ref())
+        .filter(|resource_return| resource_return.holder.is_some())
+        .map(|resource_return| resource_return.resource_type_name.clone())
+        .collect::<BTreeSet<_>>()
+        .into_iter()
+        .collect::<Vec<_>>();
+    if !held_resources.is_empty() {
+        render_type_imports(&mut imports, "./resources", &held_resources);
+    }
     render_generated_module(imports, body)
 }
 
@@ -6177,6 +6205,44 @@ fn render_request_plan_typescript(plan: &RequestPlan) -> String {
     )
 }
 
+/// The resource that an operation returns, bound to `client` when one is given, or the
+/// result record that holds it. The record takes each other field from the field of the
+/// same name in the wire record.
+fn typescript_resource_return_value(
+    resource_return: &PlannedOperationResourceReturn,
+    client: Option<&str>,
+) -> String {
+    let args = resource_return
+        .bindings
+        .iter()
+        .map(resource_return_binding_expr_typescript)
+        .collect::<Vec<_>>()
+        .join(", ");
+    let mut value = format!("new {}({args})", resource_return.resource_type_name);
+    if let Some(client) = client {
+        value = format!(
+            "{}({value}, {client})",
+            resource_client_bind_function_name_for_type(&resource_return.resource_type_name)
+        );
+    }
+    let Some(holder) = &resource_return.holder else {
+        return value;
+    };
+    let fields = holder
+        .fields
+        .iter()
+        .map(|field| {
+            let name = typescript_generated_field_name(&field.name);
+            if field.resource {
+                format!("{name}: {value}")
+            } else {
+                format!("{name}: result.{name}")
+            }
+        })
+        .collect::<Vec<_>>();
+    format!("{{ {} }}", fields.join(", "))
+}
+
 fn resource_return_binding_expr_typescript(
     binding: &PlannedOperationResourceFieldBinding,
 ) -> String {
@@ -6372,6 +6438,13 @@ fn render_operation_function(
             output.push_str("  return ");
             output.push_str(transform_expr);
             output.push_str(";\n");
+        } else if let Some(resource_return) = &operation.output_resource_return
+            && resource_return.holder.is_some()
+        {
+            // The answer is the wire message, so the record and its resource are built from it.
+            output.push_str("  return ");
+            output.push_str(&typescript_resource_return_value(resource_return, None));
+            output.push_str(";\n");
         } else if let Some(resource_return) = &operation.output_resource_return {
             // The answer is the resource's wire message, so the resource is built from it.
             let args = resource_return
@@ -6415,6 +6488,16 @@ fn render_operation_function(
         output.push_str("  return ");
         output.push_str(transform_expr);
         output.push_str(";\n");
+    } else if let Some(resource_return) = &operation.output_resource_return
+        && resource_return.holder.is_some()
+    {
+        output.push_str("  const result = await handle.result();\n");
+        output.push_str("  return ");
+        output.push_str(&typescript_resource_return_value(
+            resource_return,
+            service.endpoint.is_none().then_some("endpoint"),
+        ));
+        output.push_str(";\n");
     } else if let Some(resource_return) = &operation.output_resource_return {
         output.push_str("  const result = await handle.result();\n");
         if service.endpoint.is_none() {
diff --git a/src/planning/authored_validation.rs b/src/planning/authored_validation.rs
index fc9338b15..c38cbad8a 100644
--- a/src/planning/authored_validation.rs
+++ b/src/planning/authored_validation.rs
@@ -14,7 +14,7 @@ use crate::generator::python;
 use crate::language::Language;
 use crate::spec::{
     ApiSpec, AuthoredFamily, ExternalTypeBindingSpec, ExternalTypeSourceSpec, ExternalTypeSpec,
-    RecordFieldVisibility, RecordSpec, ServiceSpec, TypeSpec,
+    RecordFieldSpec, RecordFieldVisibility, RecordSpec, ServiceSpec, TypeSpec,
 };
 use crate::spec::{ApiSpecLeaf, CompilerPass};
 
@@ -492,7 +492,7 @@ fn validate_message_model_config(
         .record
         .fields
         .iter()
-        .filter(|(_, field)| field.required)
+        .filter(|(_, field)| field.required && !is_resource_field(field))
         .map(|(field_name, _)| field_name)
     {
         if oneof_by_name(&oneofs, field_name).is_some() {
@@ -512,8 +512,8 @@ fn validate_message_model_config(
         }
         validate_model_override_field(message_name, field_name, message)?;
     }
-    for field_name in model_config.record.fields.keys() {
-        if oneof_by_name(&oneofs, field_name).is_some() {
+    for (field_name, field) in &model_config.record.fields {
+        if oneof_by_name(&oneofs, field_name).is_some() || is_resource_field(field) {
             continue;
         }
         validate_model_override_field(message_name, field_name, message)?;
@@ -551,7 +551,7 @@ fn validate_record_fields(
     language: Language,
 ) -> Result<()> {
     for (field_name, field) in &model_config.record.fields {
-        if oneof_by_name(oneofs, field_name).is_some() {
+        if oneof_by_name(oneofs, field_name).is_some() || is_resource_field(field) {
             continue;
         }
         validate_model_override_field(message_name, field_name, message)?;
@@ -639,13 +639,17 @@ fn validate_record_fields(
         }
     }
 
+    // A record that holds a resource declares only what its caller sees. The resource binds
+    // the fields that it needs, and the other fields of the message are left out.
+    let holds_resource = model_config.record.fields.values().any(is_resource_field);
     let mut seen_generated_names: BTreeMap<String, String> = BTreeMap::new();
     for field in &message.descriptor.field {
         let proto_name = field
             .name
             .as_deref()
             .expect("descriptor fields should be named");
-        if !model_config.is_field_omitted(proto_name)
+        if !holds_resource
+            && !model_config.is_field_omitted(proto_name)
             && !model_config.record.fields.contains_key(proto_name)
             && !oneofs.iter().any(|oneof| {
                 model_config.record.fields.contains_key(oneof.name)
@@ -701,6 +705,9 @@ fn validate_authored_field_types(
             continue;
         }
 
+        if is_resource_field(field_spec) {
+            continue;
+        }
         let field = validate_model_override_field(message_name, field_name, message)?;
         if model_config.is_field_omitted(field_name) {
             continue;
@@ -1223,6 +1230,12 @@ fn validate_model_override_field<'a>(
         })
 }
 
+/// A resource field holds a resource that an operation result builds from the message, so
+/// it has no proto field of its own.
+fn is_resource_field(field: &RecordFieldSpec) -> bool {
+    matches!(field.field_type.without_option(), TypeSpec::Resource(_))
+}
+
 fn validate_model_required_field(
     message_name: &str,
     field_name: &str,
diff --git a/src/planning/mod.rs b/src/planning/mod.rs
index 19e0dd256..918b0cee9 100644
--- a/src/planning/mod.rs
+++ b/src/planning/mod.rs
@@ -37,8 +37,8 @@ pub(crate) use proto::{message_model_name, relative_descriptor_name};
 pub(crate) use reachability::ReachabilityPass;
 pub(crate) use resource_resolution::{
     RequestPlan, RequestPlanField, RequestPlanSource, ResolvedResourceBindingSource,
-    ResolvedResourceMethodBinding, ResolvedResourceReturnSpec, ResolvedResourceSpec,
-    ResourceBoundData, ResourceBoundFamily, ResourceResolutionPass,
+    ResolvedResourceHolder, ResolvedResourceMethodBinding, ResolvedResourceReturnSpec,
+    ResolvedResourceSpec, ResourceBoundData, ResourceBoundFamily, ResourceResolutionPass,
 };
 pub(crate) use selected_validation::SelectedValidationPass;
 pub(crate) use selection::LanguageSelectionPass;
@@ -119,6 +119,25 @@ pub(crate) enum PlanningMode {
 pub(crate) struct PlannedOperationResourceReturn {
     pub(crate) resource_type_name: String,
     pub(crate) bindings: Vec<PlannedOperationResourceFieldBinding>,
+    pub(crate) holder: Option<PlannedResourceHolder>,
+}
+
+/// The result record that holds the returned resource, by model and field names. Its other
+/// fields have the same names in the wire record.
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct PlannedResourceHolder {
+    pub(crate) record_full_name: String,
+    pub(crate) model_name: String,
+    /// The record's fields in order. The resource field holds the resource.
+    pub(crate) fields: Vec<PlannedResourceHolderField>,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct PlannedResourceHolderField {
+    pub(crate) name: String,
+    pub(crate) proto_name: String,
+    pub(crate) required: bool,
+    pub(crate) resource: bool,
 }
 
 #[derive(Debug, Clone, PartialEq)]
diff --git a/src/planning/operation_lowering.rs b/src/planning/operation_lowering.rs
index b98daed73..bf95a23bf 100644
--- a/src/planning/operation_lowering.rs
+++ b/src/planning/operation_lowering.rs
@@ -59,12 +59,22 @@ impl CompilerPass<OperationBoundFamily, OperationLoweredFamily> for OperationLow
     ) -> Result<ApiSpecLeaf<OperationLoweredFamily>> {
         let mut spec = leaf.spec.map_names(OperationLoweringMapper);
         let mut generated_records = Vec::new();
+        let mut holders = Vec::new();
 
         for service in &mut spec.services {
             for operation in &mut service.operations {
                 let Some(resource_return) = operation.data.output_resource_return.as_ref() else {
                     continue;
                 };
+                if let Some(holder) = &resource_return.holder {
+                    holders.push((
+                        service.name.clone(),
+                        operation.name.clone(),
+                        holder.clone(),
+                        resource_return.clone(),
+                    ));
+                    continue;
+                }
                 let Some(TypeSpec::Resource(resource)) = operation.output_type() else {
                     continue;
                 };
@@ -104,6 +114,33 @@ impl CompilerPass<OperationBoundFamily, OperationLoweredFamily> for OperationLow
             }
         }
 
+        for (service_name, operation_name, holder, resource_return) in holders {
+            let (wire_full_name, wire_record) = lower_resource_holder(
+                &spec,
+                &service_name,
+                &operation_name,
+                &holder,
+                &resource_return,
+            )?;
+            if let Some(TypeDeclEntry {
+                declaration: TypeDeclSpec::Record(record),
+                ..
+            }) = spec.types.get_mut(&holder.record_name)
+            {
+                // The holder is the API model only. Its message travels as the wire record.
+                record.source = None;
+            }
+            let operation = spec
+                .services
+                .iter_mut()
+                .filter(|service| service.name == service_name)
+                .flat_map(|service| service.operations.iter_mut())
+                .find(|operation| operation.name == operation_name)
+                .expect("the holder operation is in its service");
+            operation.output = Some(TypeSpec::Record(Symbol::new(wire_full_name.clone())));
+            generated_records.push((wire_full_name, wire_record));
+        }
+
         for (full_name, record) in generated_records {
             spec.types
                 .entry(full_name)
@@ -119,6 +156,69 @@ impl CompilerPass<OperationBoundFamily, OperationLoweredFamily> for OperationLow
     }
 }
 
+/// The wire record of a result record that holds a resource. It takes the name of the proto
+/// message, and has the holder's other fields plus the fields that the resource binds from.
+fn lower_resource_holder(
+    spec: &crate::spec::ApiSpec<OperationLoweredFamily>,
+    service_name: &str,
+    operation_name: &str,
+    holder: &super::ResolvedResourceHolder,
+    resource_return: &super::ResolvedResourceReturnSpec,
+) -> Result<(String, RecordSpec<OperationLoweredFamily>)> {
+    let record = spec
+        .record(&holder.record_name)
+        .expect("the holder record is declared");
+    let wire_name = resource_return
+        .output_message_name
+        .rsplit('.')
+        .next()
+        .unwrap_or(&resource_return.output_message_name)
+        .to_string();
+    if spec
+        .types
+        .values()
+        .any(|entry| {
+            matches!(&entry.declaration, TypeDeclSpec::Record(other) if other.name == wire_name)
+        })
+    {
+        return Err(Error::InvalidResource {
+            service: service_name.to_string(),
+            resource: resource_return.resource_name.to_upper_camel_case(),
+            reason: format!(
+                "operation `{operation_name}` needs the name `{wire_name}` for the wire record \
+                 of `{}`, and a record already has it",
+                record.name
+            ),
+        });
+    }
+    let mut fields = IndexMap::new();
+    for (key, field) in &record.fields {
+        if key == &holder.field_key {
+            fields.extend(resource_result_fields(resource_return));
+        } else {
+            fields.insert(key.clone(), field.clone());
+        }
+    }
+    let full_name = format!("__generated.{service_name}.{wire_name}");
+    Ok((
+        full_name.clone(),
+        RecordSpec {
+            name: wire_name,
+            full_name,
+            doc: record.doc.clone(),
+            source: Some(ExternalTypeSourceSpec::Proto(ProtoTypeSpec {
+                proto: Symbol::new(resource_return.output_message_name.clone()),
+                reference: Default::default(),
+                type_name: Default::default(),
+            })),
+            experimental: record.experimental,
+            flatten_in_api: false,
+            fields,
+            data: (),
+        },
+    ))
+}
+
 fn resource_result_fields(
     resource_return: &super::ResolvedResourceReturnSpec,
 ) -> IndexMap<String, RecordFieldSpec<OperationLoweredFamily>> {
diff --git a/src/planning/reachability.rs b/src/planning/reachability.rs
index d727c759a..073234e40 100644
--- a/src/planning/reachability.rs
+++ b/src/planning/reachability.rs
@@ -51,6 +51,14 @@ fn prune(spec: &mut PlannedSpec) {
                     .flatten()
                     .cloned(),
             );
+            if let Some(holder) = operation
+                .data
+                .output_resource_return
+                .as_ref()
+                .and_then(|resource_return| resource_return.holder.as_ref())
+            {
+                add_name(spec, &holder.record_full_name, &mut reachable);
+            }
         }
         for resource in &service.resources {
             enqueue_resource(&resource.data, &mut pending);
diff --git a/src/planning/resource_resolution.rs b/src/planning/resource_resolution.rs
index 72cdf9849..4cd468180 100644
--- a/src/planning/resource_resolution.rs
+++ b/src/planning/resource_resolution.rs
@@ -121,6 +121,16 @@ pub struct ResolvedResourceReturnSpec {
     pub resource_name: String,
     pub output_message_name: String,
     pub bindings: Vec<ResolvedResourceFieldBinding>,
+    /// Set when the operation returns a record that holds the resource in one field.
+    pub holder: Option<ResolvedResourceHolder>,
+}
+
+/// A result record with one resource field. The resource binds from the record's message,
+/// and the other fields of the record come from the same message.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ResolvedResourceHolder {
+    pub record_name: String,
+    pub field_key: String,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -186,12 +196,19 @@ pub(crate) fn resolve_service_resources(
 
     let mut operation_returns = BTreeMap::new();
     for operation in &service.operations {
-        let Some(resource_name) = operation_output_resource_name(service, operation) else {
-            continue;
+        let holder = operation_output_resource_holder(spec, service, operation)?;
+        let resource_name = match &holder {
+            Some((_, _, resource_name, _)) => *resource_name,
+            None => {
+                let Some(resource_name) = operation_output_resource_name(service, operation) else {
+                    continue;
+                };
+                if operation_output_resource_wire_type(operation).is_none() {
+                    continue;
+                }
+                resource_name
+            }
         };
-        if operation_output_resource_wire_type(operation).is_none() {
-            continue;
-        }
         let resource = service
             .resource(resource_name)
             .ok_or_else(|| Error::InvalidResource {
@@ -212,7 +229,11 @@ pub(crate) fn resolve_service_resources(
                 ),
             });
         };
-        let Some(output_message_name) = operation_output_resource_message_name(operation) else {
+        let Some(output_message_name) = holder
+            .as_ref()
+            .map(|(_, _, _, proto_name)| *proto_name)
+            .or_else(|| operation_output_resource_message_name(operation))
+        else {
             return Err(Error::InvalidResource {
                 service: service.name.clone(),
                 resource: resource.name.to_upper_camel_case(),
@@ -248,6 +269,10 @@ pub(crate) fn resolve_service_resources(
                 resource_name: resource.name.clone(),
                 output_message_name: output_message_name.to_string(),
                 bindings,
+                holder: holder.map(|(record_name, field_key, _, _)| ResolvedResourceHolder {
+                    record_name: record_name.to_string(),
+                    field_key: field_key.to_string(),
+                }),
             },
         );
     }
@@ -755,6 +780,69 @@ fn operation_output_resource_name<'a>(
         .map(|_| resource_name.as_str())
 }
 
+/// The proto-backed result record of an operation that holds a resource of the service, as
+/// its record name, field key, resource name and proto message name.
+fn operation_output_resource_holder<'a>(
+    spec: &'a ApiSpec<SelectedFamily>,
+    service: &'a ServiceSpec<SelectedFamily>,
+    operation: &'a OperationSpec<SelectedFamily>,
+) -> Result<Option<(&'a str, &'a str, &'a str, &'a str)>> {
+    let record = match operation.output_type() {
+        Some(TypeSpec::Record(record_name)) => spec.record(record_name.as_str()),
+        Some(TypeSpec::External(ExternalTypeSpec::Proto(proto_name))) => {
+            spec.record_for_proto(proto_name.as_str())
+        }
+        _ => None,
+    };
+    let Some(record) = record else {
+        return Ok(None);
+    };
+    let resource_fields = record
+        .fields
+        .iter()
+        .filter_map(|(key, field)| match field.field_type.without_option() {
+            TypeSpec::Resource(resource) => Some((key.as_str(), resource.as_str())),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    let [(field_key, resource_name)] = resource_fields.as_slice() else {
+        if resource_fields.is_empty() {
+            return Ok(None);
+        }
+        return Err(Error::InvalidResource {
+            service: service.name.clone(),
+            resource: resource_fields[0].1.to_upper_camel_case(),
+            reason: format!(
+                "result record `{}` of operation `{}` holds more than one resource",
+                record.name, operation.name
+            ),
+        });
+    };
+    let Some(proto_name) = record
+        .source
+        .as_ref()
+        .and_then(|source| source.proto_type())
+    else {
+        return Err(Error::InvalidResource {
+            service: service.name.clone(),
+            resource: resource_name.to_upper_camel_case(),
+            reason: format!(
+                "result record `{}` of operation `{}` holds a resource, so it needs `@nexus.proto`",
+                record.name, operation.name
+            ),
+        });
+    };
+    if service.resource(resource_name).is_none() {
+        return Ok(None);
+    }
+    Ok(Some((
+        record.full_name.as_str(),
+        field_key,
+        resource_name,
+        proto_name.as_str(),
+    )))
+}
+
 fn operation_output_ref(operation: &OperationSpec<SelectedFamily>) -> Option<&str> {
     operation.output_type()?.reference()
 }
diff --git a/src/planning/type_planning.rs b/src/planning/type_planning.rs
index 67ac8a980..43a2593e4 100644
--- a/src/planning/type_planning.rs
+++ b/src/planning/type_planning.rs
@@ -665,6 +665,7 @@ impl<'a> TypePlanningContext<'a> {
             serialization_context: operation.serialization_context.clone(),
             data: PlannedOperationData {
                 output_resource_return: plan_operation_resource_return(
+                    &self.spec,
                     operation.data.output_resource_return.as_ref(),
                 ),
             },
@@ -1280,9 +1281,27 @@ impl<'a> TypePlanningContext<'a> {
 }
 
 fn plan_operation_resource_return(
+    spec: &ApiSpec<OperationLoweredFamily>,
     output_resource_return: Option<&ResolvedResourceReturnSpec>,
 ) -> Option<PlannedOperationResourceReturn> {
     output_resource_return.map(|resource_return| PlannedOperationResourceReturn {
+        holder: resource_return.holder.as_ref().and_then(|holder| {
+            let record = spec.record(&holder.record_name)?;
+            Some(PlannedResourceHolder {
+                record_full_name: record.full_name.clone(),
+                model_name: record.name.clone(),
+                fields: record
+                    .fields
+                    .iter()
+                    .map(|(key, field)| PlannedResourceHolderField {
+                        name: field.name.clone(),
+                        proto_name: key.clone(),
+                        required: field.required,
+                        resource: *key == holder.field_key,
+                    })
+                    .collect(),
+            })
+        }),
         resource_type_name: resource_return.resource_name.to_upper_camel_case(),
         bindings: resource_return
             .bindings