Add Service.init() for early worker initialization

Enables execution of scripts before the worker's I/O loop starts,
providing a general solution for imports that interfere with I/O
operations (e.g., numpy hang on Windows - numpy/numpy#24290).

The init script is passed to workers via a temporary file referenced
by the APPOSE_INIT_SCRIPT environment variable. GroovyWorker and
python_worker both support this mechanism.

As per:
* apposed/appose-java@fb6b861
* apposed/appose-java@5e3eb7f

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

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

Author: ctrueden

src/appose/service.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import subprocess
+import tempfile
 import threading
 from enum import Enum
 from pathlib import Path
@@ -56,6 +57,7 @@ def __init__(self, cwd: str | Path, env_vars: dict[str, str] | None = None, *arg
         self._stderr_thread: threading.Thread | None = None
         self._monitor_thread: threading.Thread | None = None
         self._debug_callback: Callable[[Any], Any] | None = None
+        self._init_script: str | None = None
         self._syntax: ScriptSyntax | None = None
 
     def debug(self, debug_callback: Callable[[Any], Any]) -> None:
@@ -68,6 +70,32 @@ def debug(self, debug_callback: Callable[[Any], Any]) -> None:
         """
         self._debug_callback = debug_callback
 
+    def init(self, script: str) -> "Service":
+        """
+        Register a script to be executed when the worker process first starts up,
+        before any tasks are processed. This is useful for early initialization that
+        must happen before the worker's main loop begins, such as importing libraries
+        that may interfere with I/O operations.
+
+        Example: On Windows, importing numpy can hang when stdin is open for reading
+        (described at https://github.com/numpy/numpy/issues/24290).
+        Using service.init("import numpy") works around this by importing
+        numpy before the worker's I/O loop starts.
+
+        Args:
+            script: The script code to execute during worker initialization.
+
+        Returns:
+            This service object, for chaining method calls.
+
+        Raises:
+            RuntimeError: If the service has already started.
+        """
+        if self._process is not None:
+            raise RuntimeError("Service already started")
+        self._init_script = script
+        return self
+
     def start(self) -> None:
         """
         Explicitly launch the worker process associated with this service.
@@ -84,6 +112,19 @@ def start(self) -> None:
 
         prefix = f"Appose-Service-{self._service_id}"
 
+        # If an init script is provided, write it to a temporary file
+        # and pass its path via environment variable.
+        if self._init_script:
+            with tempfile.NamedTemporaryFile(
+                mode="w",
+                encoding="utf-8",
+                prefix="appose-init-",
+                suffix=".txt",
+                delete=False,
+            ) as init_file:
+                init_file.write(self._init_script)
+                self._env_vars["APPOSE_INIT_SCRIPT"] = init_file.name
+
         self._process = process.builder(self._cwd, self._env_vars, *self._args)
         self._stdout_thread = threading.Thread(
             target=self._stdout_loop, name=f"{prefix}-Stdout"

tests/test_service.py

@@ -131,3 +131,62 @@ def test_main_thread_queue_python():
         assert TaskStatus.COMPLETE == task.status
         thread = task.outputs.get("thread")
         assert thread != "MainThread"
+
+
+def test_init():
+    """Tests that init script is executed before tasks run."""
+    env = appose.system()
+    with env.python().init("init_value = 'initialized'") as service:
+        maybe_debug(service)
+
+        # Verify that the init script was executed and the variable is accessible.
+        task = service.task("init_value").wait_for()
+        assert TaskStatus.COMPLETE == task.status
+
+        result = task.result()
+        assert result == "initialized", "Init script should set init_value variable"
+
+
+def test_init_numpy():
+    """Tests that NumPy works on every platform, even Windows."""
+    env = (
+        appose.pixi()
+        .base("target/envs/test-init-numpy")
+        .conda("numpy=2.3.4")
+        .pypi("appose==0.7.2")
+        .log_debug()
+        .build()
+    )
+    with env.python().init("import numpy") as service:
+        maybe_debug(service)
+
+        task = service.task(
+            "import numpy\n"
+            "narr = numpy.random.default_rng(seed=1337).random([3, 5])\n"
+            "[float(v) for v in narr.flatten()]"
+        ).wait_for()
+        assert TaskStatus.COMPLETE == task.status
+
+        result = task.outputs.get("result")
+        assert isinstance(result, list)
+        expected = [
+            0.8781019003,
+            0.1855279616,
+            0.9209004548,
+            0.9465658637,
+            0.8745080903,
+            0.1157427629,
+            0.1937316623,
+            0.3417371975,
+            0.4957909002,
+            0.8983712328,
+            0.0064586191,
+            0.2274114670,
+            0.7936549524,
+            0.4142867178,
+            0.0838144031,
+        ]
+        for i, expected_val in enumerate(expected):
+            actual_val = result[i]
+            assert isinstance(actual_val, (int, float))
+            assert abs(actual_val - expected_val) < 1e-10