fix(test): test_offset_commit_atomicity must not kill the source worker

The test was calling kill_random_worker() without a stage filter, so
~1/3 of runs picked the source worker. Source has parallelism=(1,1)
and no offset-checkpoint logic, so once the source actor dies the
pipeline cannot resume production — and the test ends with ~600
source records never produced, which after the modulo-3 filter shows
up as ~199 records missing (matches the CI failure exactly).

The chaos tests in test_chaos_stress.py already restrict to
stage_id="transform" with a one-line comment explaining why; copy
that here.

This is the *test* bug. The earlier publish-commit race (PR #84
storage.rs fixes) was a real engine bug and stays fixed. The test
also exercises that bug path — but on top of it, this misuse of
kill_random_worker was masking the test as still-broken even after
the engine-side fix landed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jiaoew1991

engine/tests/test_stability_worker_recovery.py

@@ -517,11 +517,18 @@ async def test_offset_commit_atomicity(self, ray_cluster):
             await runner.initialize()
             run_task = asyncio.create_task(runner.run())
 
-            # Kill one worker after initial progress
+            # Kill one worker after initial progress.
+            # Restrict to the transform stage: killing the source worker is
+            # unrecoverable in this pipeline (source has parallelism=(1,1)
+            # and no offset checkpoint), so a kill there bypasses the
+            # offset-commit-atomicity property the test is meant to
+            # validate and just shows up as ~600 source records never
+            # produced. The chaos tests in test_chaos_stress.py already
+            # apply this same restriction for the same reason.
             await wait_for_progress(
                 runner, min_processed=200, timeout=30, collector_name=self.collector_name
             )
-            await kill_random_worker(runner)
+            await kill_random_worker(runner, stage_id="transform")
 
             await asyncio.wait_for(run_task, timeout=60)
         finally: