Migrate remaining improvements from appose-java

* Implement TaskException exception class
* Implement 4 Service syntax methods (get_var, put_var, call, proxy)
* Add Task.result() convenience method
* Add Service.syntax() accessor
* Update Environment to wire syntax into services
* Create 3 comprehensive test files (15 tests total)

Co-authored-by: Claude <[email protected]>

Author: ctrueden

src/appose/__init__.py

@@ -218,6 +218,7 @@ def task_listener(event):
 
 from .builder import SimpleBuilder
 from .environment import Environment
+from .service import TaskException  # noqa: F401
 from .shm import NDArray, SharedMemory  # noqa: F401
 
 __version__ = "0.7.3.dev0"

src/appose/environment.py

@@ -36,6 +36,7 @@
 import os
 from pathlib import Path
 
+from .syntax import GroovySyntax, PythonSyntax, ScriptSyntax
 from .util.filepath import find_exe
 from .service import Service
 
@@ -127,6 +128,7 @@ def python(self) -> Service:
             python_exes,
             "-c",
             "import appose.python_worker; appose.python_worker.main()",
+            syntax=PythonSyntax(),
         )
 
     def groovy(
@@ -155,14 +157,18 @@ def groovy(
         :raises IOError: If something goes wrong starting the worker process.
         """
         return self.java(
-            "org.apposed.appose.GroovyWorker", class_path=class_path, jvm_args=jvm_args
+            "org.apposed.appose.GroovyWorker",
+            class_path=class_path,
+            jvm_args=jvm_args,
+            syntax=GroovySyntax(),
         )
 
     def java(
         self,
         main_class: str,
         class_path: list[str] | None = None,
         jvm_args: list[str] | None = None,
+        syntax: ScriptSyntax | None = None,
     ) -> Service:
         # Collect classpath elements into a set, to avoid duplicate entries.
         cp: dict[str] = {}  # NB: Use dict instead of set to maintain insertion order.
@@ -192,9 +198,11 @@ def java(
             "jre/bin/java",
             "jre/bin/java.exe",
         ]
-        return self.service(java_exes, *args)
+        return self.service(java_exes, *args, syntax=syntax)
 
-    def service(self, exes: list[str], *args) -> Service:
+    def service(
+        self, exes: list[str], *args, syntax: ScriptSyntax | None = None
+    ) -> Service:
         """
         Create a service with the given command line arguments.
 
@@ -253,4 +261,4 @@ def service(self, exes: list[str], *args) -> Service:
         all_args.append(exe_path)
         all_args.extend(args)
 
-        return Service(self.base(), all_args)
+        return Service(self.base(), all_args, syntax)

src/appose/service.py

@@ -41,9 +41,33 @@
 from typing import Any, Callable
 from uuid import uuid4
 
+from .syntax import ScriptSyntax
 from .util.types import Args, decode, encode
 
 
+class TaskException(Exception):
+    """
+    Exception raised when a Task fails to complete successfully.
+
+    This exception is raised by Task.wait_for() when the task finishes
+    in a non-successful state (FAILED, CANCELED, or CRASHED).
+    """
+
+    def __init__(self, message: str, task: "Task") -> None:
+        super().__init__(message)
+        self.task = task
+
+    @property
+    def status(self) -> "TaskStatus":
+        """Returns the status of the failed task."""
+        return self.task.status
+
+    @property
+    def task_error(self) -> str | None:
+        """Returns the error message from the task, if available."""
+        return self.task.error
+
+
 class Service:
     """
     An Appose *service* provides access to a linked Appose *worker* running
@@ -54,7 +78,9 @@ class Service:
 
     _service_count: int = 0
 
-    def __init__(self, cwd: str | Path, args: list[str]) -> None:
+    def __init__(
+        self, cwd: str | Path, args: list[str], syntax: ScriptSyntax | None = None
+    ) -> None:
         self._cwd: Path = Path(cwd)
         self._args: list[str] = args[:]
         self._tasks: dict[str, "Task"] = {}
@@ -65,6 +91,16 @@ def __init__(self, cwd: str | Path, args: list[str]) -> None:
         self._stderr_thread: threading.Thread | None = None
         self._monitor_thread: threading.Thread | None = None
         self._debug_callback: Callable[[Any], Any] | None = None
+        self._syntax: ScriptSyntax | None = syntax
+
+    def syntax(self) -> ScriptSyntax | None:
+        """
+        Returns the script syntax associated with this service.
+
+        Returns:
+            The script syntax, or None if not configured.
+        """
+        return self._syntax
 
     def debug(self, debug_callback: Callable[[Any], Any]) -> None:
         """
@@ -126,6 +162,110 @@ def task(
         self.start()
         return Task(self, script, inputs, queue)
 
+    def get_var(self, name: str) -> Any:
+        """
+        Retrieves a variable's value from the worker process's global scope.
+
+        The variable must have been previously exported using task.export()
+        to be accessible across tasks.
+
+        Args:
+            name: The name of the variable to retrieve.
+
+        Returns:
+            The value of the variable.
+
+        Raises:
+            TaskException: If the variable retrieval fails.
+            ValueError: If no script syntax has been configured for this service.
+        """
+        if self._syntax is None:
+            raise ValueError("No script syntax configured for this service")
+        script = self._syntax.get_var(name)
+        task = self.task(script).wait_for()
+        return task.outputs.get("result")
+
+    def put_var(self, name: str, value: Any) -> None:
+        """
+        Sets a variable in the worker process's global scope and exports it
+        for future use across tasks.
+
+        Args:
+            name: The name of the variable to set in the worker process.
+            value: The value to assign to the variable.
+
+        Raises:
+            TaskException: If the variable assignment fails.
+            ValueError: If no script syntax has been configured for this service.
+        """
+        if self._syntax is None:
+            raise ValueError("No script syntax configured for this service")
+        inputs = {"_value": value}
+        script = self._syntax.put_var(name, "_value")
+        self.task(script, inputs).wait_for()
+
+    def call(self, function: str, *args: Any) -> Any:
+        """
+        Calls a function in the worker process with the given arguments and
+        returns the result.
+
+        The function must be accessible in the worker's global scope (either
+        built-in or previously defined/imported).
+
+        Args:
+            function: The name of the function to call in the worker process.
+            *args: The arguments to pass to the function.
+
+        Returns:
+            The result of the function call.
+
+        Raises:
+            TaskException: If the function call fails.
+            ValueError: If no script syntax has been configured for this service.
+        """
+        if self._syntax is None:
+            raise ValueError("No script syntax configured for this service")
+        inputs = {}
+        var_names = []
+        for i, arg in enumerate(args):
+            var_name = f"arg{i}"
+            inputs[var_name] = arg
+            var_names.append(var_name)
+        script = self._syntax.call(function, var_names)
+        task = self.task(script, inputs).wait_for()
+        return task.outputs.get("result")
+
+    def proxy(self, var: str, api: type, queue: str | None = None) -> Any:
+        """
+        Creates a proxy object providing strongly typed access to a remote
+        object in this service's worker process.
+
+        Method calls on the proxy are transparently forwarded to the remote
+        object via Tasks.
+
+        Important: The variable must be explicitly exported using
+        task.export(varName=value) in a previous task. Only exported variables
+        are accessible across tasks within the same service.
+
+        Args:
+            var: The name of the exported variable in the worker process
+                 referencing the remote object.
+            api: The interface/protocol class that the proxy should implement.
+            queue: Optional queue identifier for task execution. Pass "main" to
+                   ensure execution on the worker's main thread.
+
+        Returns:
+            A proxy object that forwards method calls to the remote object.
+
+        Raises:
+            ValueError: If no script syntax has been configured for this service.
+        """
+        if self._syntax is None:
+            raise ValueError("No script syntax configured for this service")
+        from .util.proxy import create
+
+        return create(self, var, queue)
+
     def close(self) -> None:
         """
         Close the worker process's input stream, in order to shut it down.
@@ -349,16 +489,56 @@ def listen(self, listener: Callable[["TaskEvent"], None]) -> None:
 
             self.listeners.append(listener)
 
-    def wait_for(self) -> None:
+    def wait_for(self) -> "Task":
+        """
+        Wait for this task to complete.
+
+        Returns:
+            This task (for method chaining).
+
+        Raises:
+            TaskException: If the task fails, is canceled, or crashes.
+        """
         with self.cv:
             if self.status == TaskStatus.INITIAL:
                 self.start()
 
             if self.status not in (TaskStatus.QUEUED, TaskStatus.RUNNING):
-                return
+                # Task already finished - check if we need to raise
+                if self.status != TaskStatus.COMPLETE:
+                    self._raise_if_failed()
+                return self
 
             self.cv.wait()
 
+        # After waiting, check if task failed
+        self._raise_if_failed()
+        return self
+
+    def result(self) -> Any:
+        """
+        Returns the result of this task.
+
+        This is a convenience method that returns outputs["result"].
+        For tasks that return a single value (e.g., from an expression),
+        that value is stored in outputs["result"].
+
+        Returns:
+            The task's result value.
+        """
+        return self.outputs.get("result")
+
+    def _raise_if_failed(self) -> None:
+        """Raise TaskException if this task is in a failed state."""
+        if self.status == TaskStatus.FAILED:
+            error_msg = self.error if self.error else "Unknown error"
+            raise TaskException(f"Task failed: {error_msg}", self)
+        elif self.status == TaskStatus.CANCELED:
+            raise TaskException("Task was canceled", self)
+        elif self.status == TaskStatus.CRASHED:
+            error_msg = self.error if self.error else "Worker process crashed"
+            raise TaskException(f"Task crashed: {error_msg}", self)
+
     def cancel(self) -> None:
         """
         Send a task cancelation request to the worker process.

tests/test_service.py

@@ -27,10 +27,10 @@
 # #L%
 ###
 
-import os
 
 import appose
-from appose.service import ResponseType, Service, TaskStatus
+from appose.service import TaskStatus
+from tests.test_base import execute_and_assert, maybe_debug
 
 collatz_groovy = """
 // Computes the stopping time of a given value
@@ -155,66 +155,3 @@ def test_main_thread_queue_python():
         assert TaskStatus.COMPLETE == task.status
         thread = task.outputs.get("thread")
         assert thread != "MainThread"
-
-
-def execute_and_assert(service: Service, script: str):
-    task = service.task(script)
-
-    # Record the state of the task for each event that occurs.
-
-    class TaskState:
-        def __init__(self, event):
-            self.response_type = event.response_type
-            self.message = event.message
-            self.current = event.current
-            self.maximum = event.maximum
-            self.status = event.task.status
-            self.error = event.task.error
-
-    events = []
-    task.listen(lambda event: events.append(TaskState(event)))
-
-    # Wait for task to finish.
-    task.wait_for()
-
-    # Validate the execution result.
-    assert TaskStatus.COMPLETE == task.status
-    result = task.outputs["result"]
-    assert 91 == result
-
-    # Validate the events received.
-
-    assert 93 == len(events)
-
-    launch = events[0]
-    assert ResponseType.LAUNCH == launch.response_type
-    assert TaskStatus.RUNNING == launch.status
-    assert launch.message is None
-    assert launch.current is None
-    assert launch.maximum is None
-    assert launch.error is None
-
-    v = 9999
-    for i in range(91):
-        v = v // 2 if v % 2 == 0 else 3 * v + 1
-        update = events[i + 1]
-        assert ResponseType.UPDATE == update.response_type
-        assert TaskStatus.RUNNING == update.status
-        assert f"[{i}] -> {v}" == update.message
-        assert i == update.current
-        assert update.maximum is None
-        assert update.error is None
-
-    completion = events[92]
-    assert ResponseType.COMPLETION == completion.response_type
-    assert TaskStatus.COMPLETE == completion.status
-    assert completion.message is None  # no message from non-UPDATE response
-    assert completion.current is None  # no current from non-UPDATE response
-    assert completion.maximum is None  # no maximum from non-UPDATE response
-    assert completion.error is None
-
-
-def maybe_debug(service):
-    debug = os.getenv("DEBUG")
-    if debug:
-        service.debug(print)