wip: xxx: introduce a way to submit and listen isolated calls

src/isolate/backends/remote.py

@@ -102,7 +102,7 @@ def run(
         executable: BasicCallable,
         *args: Any,
         **kwargs: Any,
-    ) -> CallResultType:
+    ) -> CallResultType:  # type: ignore
         if self._channel is None:
             self._acquire_channel()
 

src/isolate/server/definitions/server.proto

@@ -7,6 +7,9 @@ service Isolate {
     // Run the given function on the specified environment. Streams logs
     // and the result originating from that function.
     rpc Run (BoundFunction) returns (stream PartialRunResult) {}
+
+    // Submit a function to be run with callbacks for the logs and the result.
+    rpc Submit (SubmitRequest) returns (SubmitResponse) {}
 }
 
 message BoundFunction {
@@ -23,3 +26,21 @@ message EnvironmentDefinition {
     // Whether to force-create this environment or not.
     bool force = 3;
 }
+
+message SubmitRequest {
+    // The function to run.
+    BoundFunction function = 1;
+
+    // The HTTP callback to call with every partial result. It will be a POST request
+    // with serialized PartialRunResult(s) as the body.
+    string callback = 2;
+}
+
+message SubmitResponse {
+    // Reserved for future use.
+}
+
+message PartialRunResults {
+    // The results of the function.
+    repeated PartialRunResult results = 1;
+}

src/isolate/server/definitions/server_pb2.pyi

@@ -105,3 +105,69 @@ class EnvironmentDefinition(google.protobuf.message.Message):
     ) -> None: ...
 
 global___EnvironmentDefinition = EnvironmentDefinition
+
+@typing_extensions.final
+class SubmitRequest(google.protobuf.message.Message):
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    FUNCTION_FIELD_NUMBER: builtins.int
+    CALLBACK_FIELD_NUMBER: builtins.int
+    @property
+    def function(self) -> global___BoundFunction:
+        """The function to run."""
+    callback: builtins.str
+    """The HTTP callback to call with every partial result. It will be a POST request
+    with serialized PartialRunResult(s) as the body.
+    """
+    def __init__(
+        self,
+        *,
+        function: global___BoundFunction | None = ...,
+        callback: builtins.str = ...,
+    ) -> None: ...
+    def HasField(
+        self, field_name: typing_extensions.Literal["function", b"function"]
+    ) -> builtins.bool: ...
+    def ClearField(
+        self,
+        field_name: typing_extensions.Literal[
+            "callback", b"callback", "function", b"function"
+        ],
+    ) -> None: ...
+
+global___SubmitRequest = SubmitRequest
+
+@typing_extensions.final
+class SubmitResponse(google.protobuf.message.Message):
+    """Reserved for future use."""
+
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    def __init__(
+        self,
+    ) -> None: ...
+
+global___SubmitResponse = SubmitResponse
+
+@typing_extensions.final
+class PartialRunResults(google.protobuf.message.Message):
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    RESULTS_FIELD_NUMBER: builtins.int
+    @property
+    def results(
+        self,
+    ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[
+        common_pb2.PartialRunResult
+    ]:
+        """The results of the function."""
+    def __init__(
+        self,
+        *,
+        results: collections.abc.Iterable[common_pb2.PartialRunResult] | None = ...,
+    ) -> None: ...
+    def ClearField(
+        self, field_name: typing_extensions.Literal["results", b"results"]
+    ) -> None: ...
+
+global___PartialRunResults = PartialRunResults

src/isolate/server/definitions/server_pb2_grpc.py

@@ -20,6 +20,11 @@ def __init__(self, channel):
             request_serializer=server__pb2.BoundFunction.SerializeToString,
             response_deserializer=common__pb2.PartialRunResult.FromString,
         )
+        self.Submit = channel.unary_unary(
+            "/Isolate/Submit",
+            request_serializer=server__pb2.SubmitRequest.SerializeToString,
+            response_deserializer=server__pb2.SubmitResponse.FromString,
+        )
 
 
 class IsolateServicer(object):
@@ -33,6 +38,12 @@ def Run(self, request, context):
         context.set_details("Method not implemented!")
         raise NotImplementedError("Method not implemented!")
 
+    def Submit(self, request, context):
+        """Submit a function to be run with callbacks for the logs and the result."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details("Method not implemented!")
+        raise NotImplementedError("Method not implemented!")
+
 
 def add_IsolateServicer_to_server(servicer, server):
     rpc_method_handlers = {
@@ -41,6 +52,11 @@ def add_IsolateServicer_to_server(servicer, server):
             request_deserializer=server__pb2.BoundFunction.FromString,
             response_serializer=common__pb2.PartialRunResult.SerializeToString,
         ),
+        "Submit": grpc.unary_unary_rpc_method_handler(
+            servicer.Submit,
+            request_deserializer=server__pb2.SubmitRequest.FromString,
+            response_serializer=server__pb2.SubmitResponse.SerializeToString,
+        ),
     }
     generic_handler = grpc.method_handlers_generic_handler(
         "Isolate", rpc_method_handlers
@@ -80,3 +96,32 @@ def Run(
             timeout,
             metadata,
         )
+
+    @staticmethod
+    def Submit(
+        request,
+        target,
+        options=(),
+        channel_credentials=None,
+        call_credentials=None,
+        insecure=False,
+        compression=None,
+        wait_for_ready=None,
+        timeout=None,
+        metadata=None,
+    ):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            "/Isolate/Submit",
+            server__pb2.SubmitRequest.SerializeToString,
+            server__pb2.SubmitResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+        )