Bound warm workers' host memory and account local RAM

A worker over FLOW_WORKER_MAX_RSS_MB (--worker-max-rss) when its node
returns is retired instead of kept warm, in the shared pool and in every
declared one; 0, the default, keeps them all. The engine now accounts its
own memory (FLOW_RAM / --ram, else the box or the cgroup limit less 1 GB),
so a node's `ram` declaration holds locally. A declared environment the
shared pool already has is that pool rather than a duplicate set of warm
workers.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

backend/tests/flow/test_workers.py

@@ -659,6 +659,9 @@ def test_a_declared_environment_gets_a_pool_of_its_own(pool):
     assert pool.for_env({"OMP_NUM_THREADS": "3"}) is not child
     # Nothing to derive means the shared pool, with no second process anywhere.
     assert pool.for_env({}) is pool
+    # Nor does deriving what the shared pool's workers were started with.
+    pool.env["OMP_NUM_THREADS"] = "4"
+    assert pool.for_env({"OMP_NUM_THREADS": "4"}) is pool
 
 
 def test_a_child_worker_is_started_with_what_it_was_given(pool):
@@ -698,6 +701,23 @@ def test_retiring_the_cards_leaves_the_other_pools_warm(pool):
     assert threads._generation == before[1]
 
 
+def test_a_worker_over_the_memory_ceiling_is_not_kept_warm(pool):
+    """A library keeps what it allocated, so the process is what gives it back."""
+    code = "import os\n\n\ndef process():\n    return {'pid': os.getpid()}\n"
+    # Any interpreter is past a megabyte, so every call retires its worker.
+    tight = PythonWorkerPool(python=sys.executable, size=1, max_rss_mb=1)
+    tight.start()
+    try:
+        first = tight.run("demo", "pid", code, {}, "demo.pid", timeout=10)
+        assert tight.run("demo", "pid", code, {}, "demo.pid", timeout=10) != first
+        # The pools a declaration gets are held to the same ceiling.
+        assert tight.for_env({"OMP_NUM_THREADS": "2"}).max_rss_mb == 1
+    finally:
+        tight.stop()
+    # Without one, the same slot answers twice from the same process.
+    assert run(pool, code, node="pid") == run(pool, code, node="pid")
+
+
 def test_cancelling_reaches_a_node_running_in_a_child(pool):
     child = pool.for_env({"OMP_NUM_THREADS": "2"})
     started = threading.Event()