Take the engine off the path node code imports from, and let it stop

The worker script is handed to the interpreter by path, so app/flow was
sys.path[0] for every node: `import queue` got the engine's. It now drops
its own directory before anything else imports, and runs with the
deployment's credentials scrubbed out of its environment.

Also: reload builds off the event loop, the pool wakes what is blocked on
it when it stops, a refused metrics flush is kept for the next one rather
than dropped, and the cascade events are paired through failures.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017MeiWk3Yq12n2pTvnQWYvt

backend/tests/flow/test_workers.py

@@ -107,3 +107,71 @@ def test_a_result_that_is_not_json_is_refused(pool):
 def test_compiling_reports_where_the_source_is_wrong(pool):
     assert pool.compile("demo", "broken", "def process(params)\n    return {}\n")
     assert pool.compile("demo", "fine", "def process(params):\n    return {}\n") is None
+
+
+def test_a_node_imports_the_standard_library_not_the_engines_own_modules(pool):
+    # The worker script lives in app/flow, which holds queue.py, secrets.py and
+    # more; the interpreter would put that directory first on sys.path.
+    result = run(
+        pool,
+        "import queue\nimport secrets\n\n\n"
+        "def process(params):\n"
+        "    return {'out': [queue.Queue().qsize(), len(secrets.token_hex(4))]}\n",
+    )
+    assert result == {"out": [0, 8]}
+
+
+def test_the_engines_secrets_are_not_in_a_workers_environment(pool, monkeypatch):
+    monkeypatch.setenv("SECRET_KEY", "not-for-nodes")
+    monkeypatch.setenv("POSTGRES_PASSWORD", "not-for-nodes")
+    monkeypatch.setenv("FLUKSIO_HARMLESS", "fine")
+    # A fresh process, so it is built from the environment set just now.
+    pool.respawn_all()
+
+    result = run(
+        pool,
+        "import os\n\n\n"
+        "def process(params):\n"
+        "    return {'out': [k for k in ('SECRET_KEY', 'POSTGRES_PASSWORD',\n"
+        "                                'FLUKSIO_HARMLESS') if k in os.environ]}\n",
+    )
+    assert result == {"out": ["FLUKSIO_HARMLESS"]}
+
+
+def test_a_pool_can_stop_while_a_node_is_running(pool):
+    # One slot, taken by a node that will not finish on its own, and a second
+    # call queued behind it. The engine's node threads are not daemons, so a
+    # wait here is a shutdown that never completes.
+    outcomes: list[str] = []
+
+    def call(node: str) -> None:
+        try:
+            pool.run(
+                "demo",
+                node,
+                "import time\n\n\ndef process(params):\n    time.sleep(60)\n",
+                {},
+                {},
+                f"demo.{node}",
+                timeout=60,
+            )
+            outcomes.append("returned")
+        except Exception as exc:
+            outcomes.append(type(exc).__name__)
+
+    busy = threading.Thread(target=call, args=("busy",))
+    busy.start()
+    # Let the first one take the slot, so the second is blocked acquiring it.
+    time.sleep(1)
+    waiting = threading.Thread(target=call, args=("waiting",))
+    waiting.start()
+    time.sleep(0.2)
+
+    pool.stop()
+    for thread in (busy, waiting):
+        thread.join(timeout=10)
+        assert not thread.is_alive()
+    assert len(outcomes) == 2
+
+    with pytest.raises(Exception, match="shutting down"):
+        run(pool, "def process(params):\n    return {'out': 1}\n")