fix(windows): make worker pipe typing portable

.github/workflows/master.yml

@@ -77,6 +77,15 @@ jobs:
       - name: Set up the environment
         uses: ./.github/actions/setup-python-env
 
+      - name: Check Windows worker types
+        run: >-
+          uv run --locked --no-sync ty check --python-platform win32
+          src/powercontext/builtin/runtime/artifact_processing.py
+          tests/builtin/runtime/test_artifact_processing.py
+
+      - name: Verify spawned worker lifecycle on Windows
+        run: uv run --locked --no-sync pytest -q tests/builtin/runtime/test_artifact_processing.py
+
       - name: Exercise fixtures without Windows long-path support
         shell: pwsh
         run: |

src/powercontext/builtin/runtime/artifact_processing.py

@@ -29,7 +29,6 @@
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from enum import StrEnum
-from multiprocessing.connection import Connection
 from multiprocessing.context import SpawnProcess
 from multiprocessing.process import BaseProcess
 from random import SystemRandom
@@ -141,6 +140,24 @@ def __post_init__(self) -> None:
             CronTrigger.from_crontab(self.cron, timezone=ZoneInfo(self.timezone))
 
 
+class _WorkerResultSender(Protocol):
+    """The child's end of the result pipe; POSIX yields `Connection`, Windows `PipeConnection`."""
+
+    def send(self, obj: object, /) -> None: ...
+
+    def close(self) -> None: ...
+
+
+class _WorkerResultReceiver(Protocol):
+    """The owner's end of the result pipe; POSIX yields `Connection`, Windows `PipeConnection`."""
+
+    def poll(self) -> bool: ...
+
+    def recv(self) -> object: ...
+
+    def close(self) -> None: ...
+
+
 @dataclass(slots=True)
 class _SpawnStartOwner:
     """Retain Popen ownership before BaseProcess can publish the handle."""
@@ -220,8 +237,8 @@ async def _cleanup_cancelled_spawn_start(
     start_task: asyncio.Task[None],
     process: BaseProcess,
     owner: _SpawnStartOwner,
-    receiver: Connection,
-    sender: Connection,
+    receiver: _WorkerResultReceiver,
+    sender: _WorkerResultSender,
 ) -> None:
     """Own and reap a partially-started child before cancellation returns."""
 
@@ -247,8 +264,8 @@ async def _complete_spawn_cleanup(cleanup_task: asyncio.Task[None]) -> None:
 
 async def _cleanup_failed_spawn_start(
     owner: _SpawnStartOwner,
-    receiver: Connection,
-    sender: Connection,
+    receiver: _WorkerResultReceiver,
+    sender: _WorkerResultSender,
 ) -> None:
     """Terminate a child whose Popen exists but was never published by BaseProcess."""
 
@@ -278,7 +295,7 @@ def _wait_for_spawn_popen_exit(popen: SpawnPopen, timeout_seconds: float) -> Non
 
 
 class _SpawnedWorkerHandle:
-    def __init__(self, process: BaseProcess, receiver: Connection) -> None:
+    def __init__(self, process: BaseProcess, receiver: _WorkerResultReceiver) -> None:
         self._process = process
         self._receiver = receiver
         self._closed = False
@@ -1152,7 +1169,7 @@ async def close(self) -> None:
 
 
 def _run_spawned_worker(
-    sender: Connection,
+    sender: _WorkerResultSender,
     entrypoint: WorkerEntrypoint,
     assignment: ArtifactProcessingWorkAssignment,
 ) -> None:

tests/builtin/runtime/test_artifact_processing.py

@@ -17,8 +17,10 @@
 import asyncio
 import os
 import signal
+import sys
 import threading
 import time
+from contextlib import suppress
 from pathlib import Path
 from typing import Any, cast
 
@@ -75,6 +77,58 @@ def _ignore_sigterm(assignment: ArtifactProcessingWorkAssignment) -> ArtifactPro
         time.sleep(0.1)
 
 
+if sys.platform == "win32":
+    import _winapi
+
+    # `OpenProcess` reports a pid that no longer names a process as ERROR_INVALID_PARAMETER;
+    # any other failure (ERROR_ACCESS_DENIED above all) means the child may still be running.
+    _ERROR_INVALID_PARAMETER = 87
+
+    def _assert_child_exited(pid: int) -> None:
+        """Windows frees the pid on exit; a handle we can still open must already be signalled."""
+
+        try:
+            handle = _winapi.OpenProcess(_winapi.SYNCHRONIZE, False, pid)
+        except OSError as error:
+            if error.winerror == _ERROR_INVALID_PARAMETER:
+                return
+            raise
+        try:
+            assert _winapi.WaitForSingleObject(handle, 0) == _winapi.WAIT_OBJECT_0
+        finally:
+            _winapi.CloseHandle(handle)
+
+    def _assert_child_reaped(pid: int) -> None:
+        """Windows leaves no zombie for a later wait to collect, so exiting is the whole contract."""
+
+        _assert_child_exited(pid)
+
+else:
+
+    def _assert_child_exited(pid: int) -> None:
+        """POSIX reports a missing process only once the child's exit status has been collected."""
+
+        with pytest.raises(ProcessLookupError):
+            os.kill(pid, 0)
+
+    def _assert_child_reaped(pid: int) -> None:
+        """A reaped child leaves no zombie, so a non-blocking wait can no longer find it."""
+
+        _assert_child_exited(pid)
+        with pytest.raises(ChildProcessError):
+            os.waitpid(pid, os.WNOHANG)
+
+
+# multiprocessing creates the spawn child in `Popen._launch` on POSIX and in `Popen.__init__` on
+# Windows. Both take `(popen, process_obj)` and leave the same attributes (`pid`, `returncode`,
+# `sentinel`, `finalizer`) set on return, so one failure-injection hook fits either seam.
+_SPAWN_LAUNCH_SEAM = "__init__" if sys.platform == "win32" else "_launch"
+
+
+def _real_spawn_launch():
+    return getattr(cast(Any, artifact_processing_module._OwnedSpawnPopen), _SPAWN_LAUNCH_SEAM)
+
+
 async def _wait_until(predicate, *, timeout_seconds: float = 3.0) -> None:
     async with asyncio.timeout(timeout_seconds):
         while not await predicate():  # noqa: ASYNC110 - bounded observation of committed database state
@@ -138,12 +192,7 @@ async def child_is_ready() -> bool:
             assert child_pid is not None
 
             await asyncio.wait_for(supervisor.close(), timeout=SPAWN_TEST_TIMEOUT_SECONDS)
-            child_reaped = False
-            try:
-                os.kill(child_pid, 0)
-            except ProcessLookupError:
-                child_reaped = True
-            assert child_reaped
+            _assert_child_exited(child_pid)
 
             async with ArtifactProcessingSupervisor(
                 database=profile.database,
@@ -170,7 +219,7 @@ async def scenario() -> None:
         spawned = threading.Event()
         release_start = threading.Event()
         child_pid: int | None = None
-        real_launch = cast(Any, artifact_processing_module._OwnedSpawnPopen)._launch
+        real_launch = _real_spawn_launch()
 
         def stalled_launch(popen, process):
             nonlocal child_pid
@@ -179,7 +228,7 @@ def stalled_launch(popen, process):
             spawned.set()
             release_start.wait(timeout=5)
 
-        monkeypatch.setattr(artifact_processing_module._OwnedSpawnPopen, "_launch", stalled_launch)
+        monkeypatch.setattr(artifact_processing_module._OwnedSpawnPopen, _SPAWN_LAUNCH_SEAM, stalled_launch)
         async with SQLiteProfile.open(config, tables=SHARED_TABLES) as profile:
             await _request(profile.database, str(tmp_path / "worker-ready"))
             supervisor = ArtifactProcessingSupervisor(
@@ -209,12 +258,7 @@ def stalled_launch(popen, process):
             finally:
                 release_start.set()
             await asyncio.wait_for(close_task, timeout=SPAWN_TEST_TIMEOUT_SECONDS)
-            child_reaped = False
-            try:
-                os.kill(child_pid, 0)
-            except ProcessLookupError:
-                child_reaped = True
-            assert child_reaped
+            _assert_child_exited(child_pid)
 
             async with ArtifactProcessingSupervisor(
                 database=profile.database,
@@ -241,7 +285,7 @@ async def scenario() -> None:
         tracked_pipes = []
         spawn_context = artifact_processing_module.multiprocessing.get_context("spawn")
         real_pipe = type(spawn_context).Pipe
-        real_launch = cast(Any, artifact_processing_module._OwnedSpawnPopen)._launch
+        real_launch = _real_spawn_launch()
 
         def tracking_pipe(context, duplex=True):
             connections = real_pipe(context, duplex=duplex)
@@ -260,7 +304,7 @@ def stalled_launch(popen, process):
             release_start.wait(timeout=5)
 
         monkeypatch.setattr(type(spawn_context), "Pipe", tracking_pipe)
-        monkeypatch.setattr(artifact_processing_module._OwnedSpawnPopen, "_launch", stalled_launch)
+        monkeypatch.setattr(artifact_processing_module._OwnedSpawnPopen, _SPAWN_LAUNCH_SEAM, stalled_launch)
         async with SQLiteProfile.open(config, tables=SHARED_TABLES) as profile:
             await _request(profile.database, str(ready))
             supervisor = ArtifactProcessingSupervisor(
@@ -306,10 +350,7 @@ async def cleanup_started() -> bool:
                 and not task.done()
                 and task.get_name().startswith("powercontext-artifact-worker-")
             ]
-            with pytest.raises(ProcessLookupError):
-                os.kill(child_pid, 0)
-            with pytest.raises(ChildProcessError):
-                await asyncio.to_thread(os.waitpid, child_pid, os.WNOHANG)
+            _assert_child_reaped(child_pid)
 
             async with ArtifactProcessingSupervisor(
                 database=profile.database,
@@ -333,7 +374,7 @@ async def scenario() -> None:
         launched = threading.Event()
         release_failure = threading.Event()
         child_pid: int | None = None
-        real_launch = cast(Any, artifact_processing_module._OwnedSpawnPopen)._launch
+        real_launch = _real_spawn_launch()
 
         def failing_launch(popen, process):
             nonlocal child_pid
@@ -348,7 +389,7 @@ def failing_launch(popen, process):
                 release_failure.wait(timeout=5)
             raise RuntimeError
 
-        monkeypatch.setattr(artifact_processing_module._OwnedSpawnPopen, "_launch", failing_launch)
+        monkeypatch.setattr(artifact_processing_module._OwnedSpawnPopen, _SPAWN_LAUNCH_SEAM, failing_launch)
         assignment = ArtifactProcessingWorkAssignment(
             binding_name=BINDING,
             scope_id=str(ready),
@@ -376,9 +417,58 @@ def failing_launch(popen, process):
                 await start_task
 
         assert child_pid is not None
-        with pytest.raises(ProcessLookupError):
-            os.kill(child_pid, 0)
-        with pytest.raises(ChildProcessError):
-            await asyncio.to_thread(os.waitpid, child_pid, os.WNOHANG)
+        _assert_child_reaped(child_pid)
+
+    asyncio.run(scenario())
+
+
+def test_cancelled_spawn_start_reaps_the_published_child_before_returning(tmp_path, monkeypatch) -> None:
+    """Cancel once the child is fully published, hooking this module rather than a spawn internal."""
+
+    async def scenario() -> None:
+        ready = tmp_path / "cancelled-start-ready"
+        started = threading.Event()
+        release_start = threading.Event()
+        child_pid: int | None = None
+        real_start = artifact_processing_module._start_owned_spawn_process
+
+        def stalled_start(process, owner):
+            nonlocal child_pid
+            real_start(process, owner)
+            child_pid = process.pid
+            started.set()
+            release_start.wait(timeout=SPAWN_TEST_TIMEOUT_SECONDS)
+
+        monkeypatch.setattr(artifact_processing_module, "_start_owned_spawn_process", stalled_start)
+        assignment = ArtifactProcessingWorkAssignment(
+            binding_name=BINDING,
+            scope_id=str(ready),
+            artifact_family="topic-memory",
+            claimed_request_generation=1,
+            fence=ArtifactProcessingFence(
+                supervisor_group="global",
+                holder_id="holder-a",
+                supervisor_generation=1,
+                lease_mode="single-process",
+            ),
+            worker_id="00000000-0000-4000-8000-000000000001",
+        )
+        start_task = asyncio.create_task(SpawnArtifactProcessingWorkerLauncher(_ignore_sigterm).start(assignment))
+        try:
+            assert await asyncio.to_thread(started.wait, SPAWN_TEST_TIMEOUT_SECONDS)
+            start_task.cancel()
+            await asyncio.sleep(0.05)
+            assert not start_task.done()
+        finally:
+            release_start.set()
+            if not start_task.done() and not start_task.cancelling():
+                start_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await asyncio.wait_for(start_task, timeout=SPAWN_TEST_TIMEOUT_SECONDS)
+
+        assert start_task.cancelled()
+
+        assert child_pid is not None
+        _assert_child_exited(child_pid)
 
     asyncio.run(scenario())