[Bridge] Vendored the System Nexus WIT for the stream notifier RPCs.

crates/protos/protos/api_upstream/Makefile

@@ -35,7 +35,12 @@ PROTO_PATHS = paths=source_relative:$(PROTO_OUT)
 OAPI_OUT := openapi
 OAPI3_PATH := .components.schemas.Payload
 
-NEX_GEN ?= nex-gen
+# nexgen escapes WIT keywords in field names and scaffolds proto maps and bytes in the
+# forms its generators accept. Pinned by revision so every machine writes the same WIT.
+NEXGEN_GIT ?= https://github.com/moetemp/nexgen
+NEXGEN_REV ?= 6f46948678f764c68bb76f103ab43713047187b2
+NEXGEN_ROOT := $(STAMPDIR)/nexgen-$(NEXGEN_REV)
+NEX_GEN := $(NEXGEN_ROOT)/bin/nexgen
 
 $(PROTO_OUT):
 	mkdir $(PROTO_OUT)
@@ -156,8 +161,8 @@ system-nexus-wit-install:
 	@cd cmd/protoc-gen-system-nexus-wit && go install .
 
 nex-gen-install:
-	printf $(COLOR) "Install nex-gen if missing..."
-	command -v $(NEX_GEN) >/dev/null || cargo install nex-gen --version ^0.1
+	printf $(COLOR) "Install nexgen $(NEXGEN_REV) if missing..."
+	command -v $(NEX_GEN) >/dev/null || CARGO_NET_GIT_FETCH_WITH_CLI=true cargo install --locked --git $(NEXGEN_GIT) --rev $(NEXGEN_REV) --features advanced --root $(NEXGEN_ROOT) nexgen
 
 ##### Clean #####
 clean:

crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml

@@ -17,3 +17,48 @@ services:
           $pythonRef: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse
           $rubyRef: Temporalio::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionResponse
           $typescriptRef: '@temporalio/api/workflowservice/v1.SignalWithStartWorkflowExecutionResponse'
+      AttachStreamCallback:
+        input:
+          $dotnetRef: Temporalio.Api.WorkflowService.V1.AttachStreamCallbackRequest
+          $goRef: go.temporal.io/api/workflowservice/v1.AttachStreamCallbackRequest
+          $javaRef: io.temporal.api.workflowservice.v1.AttachStreamCallbackRequest
+          $pythonRef: temporalio.api.workflowservice.v1.AttachStreamCallbackRequest
+          $rubyRef: Temporalio::Api::WorkflowService::V1::AttachStreamCallbackRequest
+          $typescriptRef: '@temporalio/api/workflowservice/v1.AttachStreamCallbackRequest'
+        output:
+          $dotnetRef: Temporalio.Api.WorkflowService.V1.AttachStreamCallbackResponse
+          $goRef: go.temporal.io/api/workflowservice/v1.AttachStreamCallbackResponse
+          $javaRef: io.temporal.api.workflowservice.v1.AttachStreamCallbackResponse
+          $pythonRef: temporalio.api.workflowservice.v1.AttachStreamCallbackResponse
+          $rubyRef: Temporalio::Api::WorkflowService::V1::AttachStreamCallbackResponse
+          $typescriptRef: '@temporalio/api/workflowservice/v1.AttachStreamCallbackResponse'
+      DetachStreamCallback:
+        input:
+          $dotnetRef: Temporalio.Api.WorkflowService.V1.DetachStreamCallbackRequest
+          $goRef: go.temporal.io/api/workflowservice/v1.DetachStreamCallbackRequest
+          $javaRef: io.temporal.api.workflowservice.v1.DetachStreamCallbackRequest
+          $pythonRef: temporalio.api.workflowservice.v1.DetachStreamCallbackRequest
+          $rubyRef: Temporalio::Api::WorkflowService::V1::DetachStreamCallbackRequest
+          $typescriptRef: '@temporalio/api/workflowservice/v1.DetachStreamCallbackRequest'
+        output:
+          $dotnetRef: Temporalio.Api.WorkflowService.V1.DetachStreamCallbackResponse
+          $goRef: go.temporal.io/api/workflowservice/v1.DetachStreamCallbackResponse
+          $javaRef: io.temporal.api.workflowservice.v1.DetachStreamCallbackResponse
+          $pythonRef: temporalio.api.workflowservice.v1.DetachStreamCallbackResponse
+          $rubyRef: Temporalio::Api::WorkflowService::V1::DetachStreamCallbackResponse
+          $typescriptRef: '@temporalio/api/workflowservice/v1.DetachStreamCallbackResponse'
+      NotifyStream:
+        input:
+          $dotnetRef: Temporalio.Api.WorkflowService.V1.NotifyStreamRequest
+          $goRef: go.temporal.io/api/workflowservice/v1.NotifyStreamRequest
+          $javaRef: io.temporal.api.workflowservice.v1.NotifyStreamRequest
+          $pythonRef: temporalio.api.workflowservice.v1.NotifyStreamRequest
+          $rubyRef: Temporalio::Api::WorkflowService::V1::NotifyStreamRequest
+          $typescriptRef: '@temporalio/api/workflowservice/v1.NotifyStreamRequest'
+        output:
+          $dotnetRef: Temporalio.Api.WorkflowService.V1.NotifyStreamResponse
+          $goRef: go.temporal.io/api/workflowservice/v1.NotifyStreamResponse
+          $javaRef: io.temporal.api.workflowservice.v1.NotifyStreamResponse
+          $pythonRef: temporalio.api.workflowservice.v1.NotifyStreamResponse
+          $rubyRef: Temporalio::Api::WorkflowService::V1::NotifyStreamResponse
+          $typescriptRef: '@temporalio/api/workflowservice/v1.NotifyStreamResponse'

crates/protos/protos/api_upstream/nexus/workflow-service.wit

@@ -138,4 +138,101 @@ interface workflow-service {
   signal-with-start-workflow: func(
     request: signal-with-start-workflow-request,
   ) -> signal-with-start-workflow-response;
+  /// @nexus.proto "temporal.api.enums.v1.StreamOwnerKind"
+  enum stream-owner-kind {
+    stream-owner-kind-unspecified,
+    stream-owner-kind-workflow,
+  }
+
+  /// @nexus.proto "temporal.api.stream.v1.StreamReference"
+  record stream-reference {
+    owner-kind: option<stream-owner-kind>,
+    workflow-id: string,
+    /// @nexus.doc "The id of the first run in the owning Workflow's run chain. It is required, since the server refuses an empty run id."
+    run-id: option<string>,
+    topic: string,
+  }
+
+  /// @nexus.proto "temporal.api.common.v1.Callback.Nexus"
+  record callback-nexus {
+    url: string,
+    header: map<string, string>,
+  }
+
+  /// @nexus.proto "temporal.api.workflowservice.v1.AttachStreamCallbackRequest"
+  record attach-stream-callback-request {
+    /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()"
+    /// @nexus.doc "Namespace of the stream."
+    namespace: string,
+    stream-ref: option<stream-reference>,
+    request-id: string,
+    callback: option<callback-nexus>,
+    /// @nexus.omit
+    identity: placeholder,
+    operation-token: string,
+    start-time: option<timestamp>,
+    /// @nexus.omit
+    workflow-target: placeholder,
+  }
+
+  /// @nexus.proto "temporal.api.workflowservice.v1.AttachStreamCallbackResponse"
+  record attach-stream-callback-response {
+  }
+
+  attach-stream-callback: func(
+    request: attach-stream-callback-request,
+  ) -> attach-stream-callback-response;
+  /// @nexus.proto "temporal.api.workflowservice.v1.DetachStreamCallbackRequest"
+  record detach-stream-callback-request {
+    /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()"
+    /// @nexus.doc "Namespace of the stream."
+    namespace: string,
+    stream-ref: option<stream-reference>,
+    request-id: string,
+    /// @nexus.omit
+    identity: placeholder,
+  }
+
+  /// @nexus.proto "temporal.api.workflowservice.v1.DetachStreamCallbackResponse"
+  record detach-stream-callback-response {
+  }
+
+  detach-stream-callback: func(
+    request: detach-stream-callback-request,
+  ) -> detach-stream-callback-response;
+  use nexus:temporal-types/model@1.0.0.{
+    payload,
+  };
+
+  /// @nexus.proto "temporal.api.workflowservice.v1.NotifyStreamRequest"
+  record notify-stream-request {
+    /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()"
+    /// @nexus.doc "Namespace of the stream."
+    namespace: string,
+    stream-ref: option<stream-reference>,
+    position: string,
+    counter: option<s64>,
+    metadata: map<string, string>,
+    close: option<bool>,
+    close-result: option<payload>,
+    /// @nexus.omit
+    identity: placeholder,
+    close-failure: option<failure>,
+  }
+
+  /// @nexus.proto "temporal.api.workflowservice.v1.NotifyStreamResponse"
+  record notify-stream-response {
+  }
+
+  notify-stream: func(
+    request: notify-stream-request,
+  ) -> notify-stream-response;
+  /// @nexus.proto "google.protobuf.Timestamp"
+  record timestamp {
+    seconds: option<s64>,
+    nanos: option<s32>,
+  }
+  use nexus:temporal-types/model@1.0.0.{
+    failure,
+  };
 }

crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto

@@ -2046,17 +2046,19 @@ service WorkflowService {
     // starts the notifier when the stream has none.
     //
     // (-- api-linter: core::0127::http-annotation=disabled
-    //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
+    //     aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --)
     rpc AttachStreamCallback (AttachStreamCallbackRequest) returns (AttachStreamCallbackResponse) {
+        option (nexusannotations.v1.operation).tags = "exposed";
     }
 
     // DetachStreamCallback removes a callback from the notifier of a stream. The notifier completes
     // the callback's operation as canceled and sends it no more progress. A Nexus handler calls it
     // when the caller cancels the operation.
     //
     // (-- api-linter: core::0127::http-annotation=disabled
-    //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
+    //     aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --)
     rpc DetachStreamCallback (DetachStreamCallbackRequest) returns (DetachStreamCallbackResponse) {
+        option (nexusannotations.v1.operation).tags = "exposed";
     }
 
     // WaitStreamCallback asks the notifier of a stream to wake one idle callback once, when the
@@ -2081,15 +2083,16 @@ service WorkflowService {
     // when the stream has none, so a callback that attaches later still gets the close outcome.
     //
     // (-- api-linter: core::0127::http-annotation=disabled
-    //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
+    //     aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --)
     rpc NotifyStream (NotifyStreamRequest) returns (NotifyStreamResponse) {
+        option (nexusannotations.v1.operation).tags = "exposed";
     }
 
     // DescribeStreamNotifier returns the state of the notifier of a stream and its callbacks. It is
-    // for operators and tools.
+    // for operators and tools, so it is not exposed through System Nexus.
     //
     // (-- api-linter: core::0127::http-annotation=disabled
-    //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
+    //     aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --)
     rpc DescribeStreamNotifier (DescribeStreamNotifierRequest) returns (DescribeStreamNotifierResponse) {
     }
 
@@ -2098,10 +2101,11 @@ service WorkflowService {
     // (`streamnotifier.pollMaxWait`). A client that reads a stream calls it after a read that
     // found nothing, in place of another blocking read on the store. It starts the notifier when the
     // stream has none. The client reads the stream once more after the poll starts, since a
-    // notification that came before the notifier existed was dropped.
+    // notification that came before the notifier existed was dropped. It is a long poll, so it is
+    // not exposed through System Nexus.
     //
     // (-- api-linter: core::0127::http-annotation=disabled
-    //     aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --)
+    //     aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --)
     rpc PollStreamNotifier (PollStreamNotifierRequest) returns (PollStreamNotifierResponse) {
     }