fix foo.execute_operator() with delegation

fiftyone/operators/delegated.py

@@ -514,11 +514,12 @@ async def _execute_operator(self, doc):
         result = await do_execute_operator(operator, ctx, exhaust=True)
 
         outputs_schema = None
-        request_params = {**context.request_params, "results": result}
         try:
-            outputs = await resolve_type_with_context(
-                request_params, "outputs"
-            )
+            # Resolve output types now
+            ctx.request_params["target"] = "outputs"
+            ctx.request_params["results"] = result
+
+            outputs = resolve_type_with_context(operator, ctx)
             if outputs is not None:
                 outputs_schema = outputs.to_json()
         except (AttributeError, Exception):

fiftyone/operators/executor.py

@@ -378,31 +378,25 @@ async def resolve_type(registry, operator_uri, request_params):
         required_secrets=operator._plugin_secrets,
     )
     await ctx.resolve_secret_values(operator._plugin_secrets)
-    try:
-        return operator.resolve_type(
-            ctx, request_params.get("target", "inputs")
-        )
-    except Exception as e:
-        return ExecutionResult(error=traceback.format_exc())
 
+    return resolve_type_with_context(operator, ctx)
 
-async def resolve_type_with_context(request_params, target=None):
-    """Resolves the "inputs" or "outputs" schema of an operator with the given
-    context.
 
+def resolve_type_with_context(operator, context):
+    """Resolves the "inputs" or "outputs" schema of an operator with the given context.
     Args:
-        request_params: a dictionary of request parameters
-        target (None): the target schema ("inputs" or "outputs")
-
+        operator: the :class:`fiftyone.operators.Operator`
+        context: the :class:`ExecutionContext` of an operator
     Returns:
-        the schema of "inputs" or "outputs"
-        :class:`fiftyone.operators.types.Property` of an operator, or None
+        the schema of "inputs" or "outputs" :class:`fiftyone.operators.types.Property` of
+        an operator, or None
     """
-    computed_target = target or request_params.get("target", None)
-    computed_request_params = {**request_params, "target": computed_target}
-    operator_uri = request_params.get("operator_uri", None)
-    registry = OperatorRegistry()
-    return await resolve_type(registry, operator_uri, computed_request_params)
+    try:
+        return operator.resolve_type(
+            context, context.request_params.get("target", "inputs")
+        )
+    except Exception as e:
+        return ExecutionResult(error=traceback.format_exc())
 
 
 async def resolve_execution_options(registry, operator_uri, request_params):