Add automatic proxy support for non-serializable objects

This commit implements transparent proxying of non-JSON-serializable
objects returned from worker tasks, along with enhanced proxy
functionality for better ergonomics.

Key changes:
- Non-serializable task outputs are automatically exported as
  worker_object references and wrapped in ProxyObject instances
- ProxyObject (renamed from ProxyHandler) now supports:
  - Immediate attribute access via __getattr__ (returns values or
    nested proxies)
  - Calling proxied callables via __call__
  - Natural chaining: proxy.obj.method(args)
- Auto-generated proxy variables use _appose_auto_N naming pattern

Implementation details:
- Python: message.py encoder detects non-serializable objects and
  creates worker_object references with auto-export
- Groovy: Messages.java includes catch-all converter for same behavior
- Service-side: proxify_worker_objects() recursively converts
  worker_object dicts to ProxyObject instances in task outputs

Tests added:
- test_auto_proxy: Tests automatic proxying with datetime, custom
  classes, and nested objects
- test_callable_proxy: Tests proxying of callable objects (lambdas
  and callable classes)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: ctrueden

src/appose/python_worker.py

@@ -177,6 +177,8 @@ def __init__(self):
 
         # Flag this process as a worker, not a service.
         message._worker_mode = True
+        # Store reference to this worker for auto-export functionality.
+        message._worker_instance = self
 
     def run(self) -> None:
         """

src/appose/service.py

@@ -20,7 +20,7 @@
 
 from .syntax import ScriptSyntax, get as syntax_from_name
 from .util import process
-from .util.message import Args, decode, encode
+from .util.message import Args, decode, encode, proxify_worker_objects
 
 
 class TaskException(Exception):
@@ -685,7 +685,9 @@ def _handle(self, response: Args) -> None:
             self.status = TaskStatus.COMPLETE
             outputs = response.get("outputs")
             if outputs is not None:
-                self.outputs.update(outputs)
+                # Convert any worker_object references to ProxyObject instances.
+                proxified_outputs = proxify_worker_objects(outputs, self.service)
+                self.outputs.update(proxified_outputs)
         elif response_type == ResponseType.CANCELATION:
             self.service._tasks.pop(self.uuid, None)
             self.status = TaskStatus.CANCELED

src/appose/util/message.py

@@ -19,6 +19,12 @@
 # Set to True by python_worker.Worker.__init__().
 _worker_mode = False
 
+# Counter for auto-generated proxy variable names.
+_proxy_counter = 0
+
+# Reference to the worker instance, needed for auto-exporting.
+_worker_instance = None
+
 
 def encode(data: Args) -> str:
     return json.dumps(data, cls=_ApposeJSONEncoder, separators=(",", ":"))
@@ -43,6 +49,23 @@ def default(self, obj):
                 "shape": obj.shape,
                 "shm": obj.shm,
             }
+
+        # If in worker mode and object is not JSON-serializable,
+        # auto-export it and return a worker_object reference.
+        if _worker_mode:
+            global _proxy_counter
+            var_name = f"_appose_auto_{_proxy_counter}"
+            _proxy_counter += 1
+
+            # Export the object so it persists for future tasks.
+            if _worker_instance is not None:
+                _worker_instance.exports[var_name] = obj
+
+            return {
+                "appose_type": "worker_object",
+                "var_name": var_name,
+            }
+
         return super().default(obj)
 
 
@@ -53,5 +76,41 @@ def _appose_object_hook(obj: dict):
         return SharedMemory(name=(obj["name"]), rsize=(obj["rsize"]))
     elif atype == "ndarray":
         return NDArray(obj["dtype"], obj["shape"], obj["shm"])
+    elif atype == "worker_object":
+        # Keep worker_object dicts as-is for now.
+        # They will be converted to proxies by proxify_worker_objects().
+        return obj
     else:
         return obj
+
+
+def proxify_worker_objects(data: Any, service: Any) -> Any:
+    """
+    Recursively convert worker_object dicts to ProxyObject instances.
+
+    This is called on task outputs after JSON deserialization to convert
+    any worker_object references into actual proxy objects.
+
+    Args:
+        data: The data structure (potentially) containing worker_object dicts.
+        service: The Service instance to use for creating proxies.
+
+    Returns:
+        The data with worker_object dicts replaced by ProxyObject instances.
+    """
+    if isinstance(data, dict):
+        if data.get("appose_type") == "worker_object":
+            # Convert this worker_object dict to a ProxyObject.
+            from .proxy import create
+
+            var_name = data["var_name"]
+            return create(service, var_name, queue=None)
+        else:
+            # Recursively process dict values.
+            return {k: proxify_worker_objects(v, service) for k, v in data.items()}
+    elif isinstance(data, list):
+        # Recursively process list elements.
+        return [proxify_worker_objects(item, service) for item in data]
+    else:
+        # Primitive value, return as-is.
+        return data

src/appose/util/proxy.py

@@ -3,26 +3,35 @@
 # SPDX-License-Identifier: BSD-2-Clause
 
 """
-Utility functions for creating local proxy objects that provide strongly typed
-access to remote objects living in Appose worker processes.
+Utility functions for creating local proxy objects that provide access to
+remote objects living in Appose worker processes.
 
-A proxy object forwards method calls to a corresponding object in the worker
-process by generating and executing scripts via Tasks. This provides a more
-natural, object-oriented API compared to manually constructing script strings
-for each method invocation.
+A proxy object forwards attribute accesses and method calls to a corresponding
+object in the worker process by generating and executing scripts via Tasks.
+This provides a natural, object-oriented API compared to manually constructing
+script strings for each operation.
 
-Type safety is honor-system based: The interface you provide must match the
-actual methods and signatures of the remote object. If there's a mismatch,
-you'll get runtime errors from the worker process.
+Proxy objects support:
+- Attribute access: proxy.field returns the field value (or another proxy)
+- Method calls: proxy.method(args) invokes the method remotely
+- Chaining: proxy.obj.method(args) works naturally
+- Callables: proxy() invokes the proxied object if it's callable
 
 Usage pattern: First, create and export the remote object via a task,
 then create a proxy to interact with it:
 
     service = env.python()
     service.task("task.export(my_obj=MyClass())").wait_for()
-    proxy = service.proxy("my_obj", MyInterface)
+    proxy = service.proxy("my_obj")
     result = proxy.some_method(42)  # Executes remotely
 
+Automatic proxying: When a task returns a non-JSON-serializable object, it's
+automatically exported and returned as a proxy object:
+
+    counter = service.task("import collections; collections.Counter('abbc')").wait_for().result()
+    # counter is now a ProxyObject wrapping the remote Counter
+    total = counter.total()  # Access the total method and call it
+
 Important: Variables must be explicitly exported using task.export(varName=value)
 in a previous task before they can be proxied. Exported variables persist across
 tasks within the same service.
@@ -77,34 +86,45 @@ def create(service: Service, var: str, queue: str | None = None) -> Any:
         RuntimeError: If a proxied method call fails in the worker process.
     """
 
-    class ProxyHandler:
+    class ProxyObject:
         def __init__(self, service: Service, var: str, queue: str | None):
             self._service = service
             self._var = var
             self._queue = queue
 
         def __getattr__(self, name: str):
-            def method(*args):
-                # Construct map of input arguments.
-                inputs = {}
-                arg_names = []
-                for i, arg in enumerate(args):
-                    arg_name = f"arg{i}"
-                    inputs[arg_name] = arg
-                    arg_names.append(arg_name)
-
-                # Use the service's ScriptSyntax to generate the method invocation script.
-                # This allows support for different languages with varying syntax.
-                syntax.validate(self._service)
-                script = self._service._syntax.invoke_method(self._var, name, arg_names)
-
-                try:
-                    task = self._service.task(script, inputs, self._queue)
-                    task.wait_for()
-                    return task.result()
-                except Exception as e:
-                    raise RuntimeError(str(e)) from e
-
-            return method
-
-    return ProxyHandler(service, var, queue)  # type: ignore
+            # Immediately evaluate the attribute access on the worker.
+            attr_expr = f"{self._var}.{name}"
+
+            try:
+                task = self._service.task(attr_expr, queue=self._queue)
+                task.wait_for()
+                result = task.result()
+                # If result is a worker_object, it will already be a ProxyObject
+                # thanks to proxify_worker_objects() in Task._handle().
+                return result
+            except Exception as e:
+                raise RuntimeError(str(e)) from e
+
+        def __call__(self, *args):
+            # Invoke the proxied object as a callable.
+            # Construct map of input arguments.
+            inputs = {}
+            arg_names = []
+            for i, arg in enumerate(args):
+                arg_name = f"arg{i}"
+                inputs[arg_name] = arg
+                arg_names.append(arg_name)
+
+            # Use the service's ScriptSyntax to generate the call script.
+            syntax.validate(self._service)
+            script = self._service._syntax.call(self._var, arg_names)
+
+            try:
+                task = self._service.task(script, inputs, self._queue)
+                task.wait_for()
+                return task.result()
+            except Exception as e:
+                raise RuntimeError(str(e)) from e
+
+    return ProxyObject(service, var, queue)  # type: ignore

tests/test_syntax.py

@@ -173,3 +173,107 @@ def walk(self, rate):
             fish.dive(100) == "Swam down 100 deep"
             or fish.dive(100) == "Swam down 100.0 deep"
         )
+
+
+def test_auto_proxy():
+    """Test automatic proxying of non-serializable task outputs."""
+    env = appose.system()
+    with env.python() as service:
+        maybe_debug(service)
+
+        # Return a non-serializable object from a task - should auto-proxy
+        # datetime is not JSON-serializable
+        dt = service.task("import datetime\ndatetime.datetime(2024, 1, 15, 10, 30, 45)").wait_for().result()
+
+        # dt should be a proxy object now
+        assert dt is not None
+
+        # Access attributes on the proxied datetime
+        assert dt.year == 2024
+        assert dt.month == 1
+        assert dt.day == 15
+
+        # Call a method
+        iso_str = dt.isoformat()
+        assert "2024-01-15T10:30:45" == iso_str
+
+        # Access a field that returns a primitive type
+        # Counter doesn't have simple fields, so let's create a custom class
+        custom = service.task("""
+class CustomClass:
+    def __init__(self):
+        self.value = 42
+        self.name = "test"
+
+    def get_double(self):
+        return self.value * 2
+
+CustomClass()
+""").wait_for().result()
+
+        # Access primitive fields
+        assert custom.value == 42
+        assert custom.name == "test"
+
+        # Call a method
+        assert custom.get_double() == 84
+
+        # Test nested object access
+        nested = service.task("""
+class Inner:
+    def __init__(self):
+        self.data = "inner_data"
+
+    def process(self, x):
+        return f"processed: {x}"
+
+class Outer:
+    def __init__(self):
+        self.inner = Inner()
+        self.label = "outer"
+
+Outer()
+""").wait_for().result()
+
+        # Access nested object
+        assert nested.label == "outer"
+        inner = nested.inner
+        assert inner.data == "inner_data"
+        assert inner.process("test") == "processed: test"
+
+        # Or chain it all together
+        result = nested.inner.process("chained")
+        assert result == "processed: chained"
+
+
+def test_callable_proxy():
+    """Test that proxied callable objects work correctly."""
+    env = appose.system()
+    with env.python() as service:
+        maybe_debug(service)
+
+        # Create a lambda function
+        func = service.task("lambda x, y: x + y").wait_for().result()
+
+        # Call it
+        result = func(10, 32)
+        assert result == 42
+
+        # Create a custom callable class
+        callable_obj = service.task("""
+class Adder:
+    def __init__(self, offset):
+        self.offset = offset
+
+    def __call__(self, x):
+        return x + self.offset
+
+Adder(100)
+""").wait_for().result()
+
+        # Call the callable object
+        result = callable_obj(23)
+        assert result == 123
+
+        # Access a field on the callable object
+        assert callable_obj.offset == 100