fix: Gate model stages on current process memory

docs/resource-budgets.md

@@ -468,3 +468,53 @@ unchanged. Stdlib tests exercise the actual loaders and handlers with weakrefs,
 constructor and inference barriers, failed replacement, deadlines and identity
 generation changes. Native memory and whole-process admission remain separate
 validation work.
+
+### Cooperative process RSS admission checkpoint
+
+The shared model daemon accepts `TRUEMEMORY_MODEL_SERVER_MAX_RSS_MB` as an
+integer MiB setting, from 0 through 2,147,483,647. Its default is 0, which
+disables this admission policy and performs no RSS measurements. This separate
+setting does not change MCP's historical `ru_maxrss` reporting or the existing
+`TRUEMEMORY_MAX_RSS_MB` setting. A finite deployment default remains uncalibrated.
+
+For a nonzero setting, the byte budget is B = configured MiB * 1,048,576.
+Each admission reads current process RSS U from `psutil.Process().memory_info()`.
+U < B admits the stage; U >= B refuses it. For example, a synthetic 1 MiB budget
+admits 1,048,575 bytes and refuses both 1,048,576 and 1,048,577 bytes. An unavailable
+sensor, sensor exception, missing value, boolean, noninteger or nonpositive RSS
+also refuses when enabled. A later current reading below B can admit work even
+after a higher historical peak. RSS is not added to GPU allocator or physical
+footprint measurements, which can overlap it.
+
+Checks occur under the existing main inference/state or fast encoder owner:
+after obsolete cache references are released and before model construction,
+after construction before cache publication, before each encode/predict slice,
+before embedding CPU transfer and before CPU retries. Recovery records sticky
+CPU selection before possible refusal. A refused embedding transfer invalidates
+the failed accelerator cache, so the next request cannot bypass CPU selection
+through a cache hit. Reranker recovery retains its release-before-replacement
+behavior. Refused requests return `server_busy` with the existing 250 ms retry
+hint, without partial vectors or scores. Fast-lane refusal propagates directly
+instead of falling through to main inference. Ownership release, deadlines,
+fast generation fencing, models, dtypes, dimensions, result ordering and
+existing request/batch/result limits are preserved.
+
+RSS sensor work consumes the request's existing monotonic deadline. The daemon
+checks that deadline before and after sampling, and again after device setup
+before main constructors. If sensing both expires the request and returns an
+unavailable or overbudget value, deadline expiry takes precedence. No constructor,
+transfer or inference retry starts for that expired request. Expiry while
+admitting an embedding CPU transfer still invalidates its failed MPS cache and
+retains sticky CPU selection. Disabled admission performs no sensor reads and
+retains the constructor deadline checks.
+
+This is opt-in sampled admission, not a hard memory ceiling or a reservation.
+Concurrent main and fast lanes can each pass a sample, and one constructor,
+transfer, native call or result allocation can grow beyond B before the next
+sample. Native allocator caches and unified-memory accounting remain outside
+this policy. Synthetic tests cover thresholds, invalid sensors, constructor and
+slice crossings, both CPU recovery routes, refusal without fallback, and zero
+sensor reads when disabled. Finite-default calibration, workspace reservations,
+system headroom, bounded recovery and actual Mac/MPS memory and thermal gates
+remain pending. This checkpoint does not establish that the original 26.24 GB
+Mac symptom is fixed.

tests/test-model-admission-737.py

@@ -330,7 +330,7 @@ def predict(pairs: list, **kwargs: object) -> list:
             self.assertTrue(release.wait(2))
             return [0.5] * len(pairs)
 
-        self.server._get_reranker = lambda name: types.SimpleNamespace(predict=predict)
+        self.server._get_reranker = lambda name, deadline=None: types.SimpleNamespace(predict=predict)
         self.server_module._store_batch_result = lambda result, values, *args: values
         self.server._queue_timeout = 0.05
         payload = b'{"op":"rerank","pairs":[["synthetic","document"]]}'
@@ -480,7 +480,7 @@ def close(self) -> None:
 
     def test_admitted_fast_query_still_runs_while_main_ownership_is_busy(self) -> None:
         self.server._inference_lock.acquire()
-        self.server._get_fast_encoder = lambda tier: types.SimpleNamespace(encode=lambda texts, **kwargs: [[0.25]])
+        self.server._get_fast_encoder = lambda tier, deadline=None: types.SimpleNamespace(encode=lambda texts, **kwargs: [[0.25]])
         try:
             client, future = self.send(b'{"op":"embed","texts":["synthetic"]}')
             self.assertEqual(self.response(client)["vectors"], [[0.25]])

tests/test-model-allocation-preflight-297.py

@@ -246,7 +246,7 @@ def forward(items: list, **kwargs: object) -> Array:
 
             model = types.SimpleNamespace(**{method: forward})
             setattr(self.server, "_get_embed_model" if op == "embed" else "_get_reranker",
-                    lambda _name: model)
+                    lambda _name, deadline=None: model)
             response = self.server.handle_request(self.request(op))
             self.assertTrue(response["ok"])
             self.assertEqual(calls, [[0, 1], [2, 3], [4]])
@@ -265,8 +265,8 @@ def test_native_count_and_later_dimension_changes_remain_errors(self) -> None:
     def test_empty_native_results_keep_their_existing_rank(self) -> None:
         model = types.SimpleNamespace(encode=lambda *args, **kwargs: Array((0,), []),
                                       predict=lambda *args, **kwargs: Array((0,), []))
-        self.server._get_embed_model = lambda tier: model
-        self.server._get_reranker = lambda name: model
+        self.server._get_embed_model = lambda tier, deadline=None: model
+        self.server._get_reranker = lambda name, deadline=None: model
         for op, field in (("embed", "vectors"), ("rerank", "scores")):
             response = self.server.handle_request(self.request(op, count=0))
             self.assertEqual(response[field].shape, (0,))
@@ -330,7 +330,7 @@ def to(self, device: str) -> None:
                         moves.append(device)
                         recovered.set()
 
-                self.server._get_embed_model = lambda tier: Model()
+                self.server._get_embed_model = lambda tier, deadline=None: Model()
                 self.mps.flush_mps_cache.side_effect = lambda: self.assertIsNone(refs[-1]())
                 response = self.server.handle_request(self.request("embed", count=count))
                 self.assertTrue(response["ok"])
@@ -357,7 +357,7 @@ def predict(self, pairs: list, **kwargs: object) -> list:
                     raise RuntimeError("MPS backend out of memory")
                 return [float(value) for value in ids]
 
-        def get_model(name: str) -> Model:
+        def get_model(name: str, deadline: object = None) -> Model:
             if self.server._reranker is None:
                 replacing = "rerank" in self.server._sticky_cpu
                 if replacing:
@@ -421,7 +421,7 @@ def run(request: dict, started: threading.Event | None = None) -> None:
             except Exception as error:
                 outcomes.append(error)
 
-        def load_embed(tier: str) -> object:
+        def load_embed(tier: str, deadline: object = None) -> object:
             loaded.set()
             return types.SimpleNamespace(encode=lambda texts, **kwargs: [[0.5] for _ in texts])
 
@@ -460,7 +460,7 @@ class Model:
                     predict = staticmethod(fail)
                     to = Mock(side_effect=AssertionError("Expired CPU transfer"))
 
-                def load(name: str) -> Model:
+                def load(name: str, deadline: object = None) -> Model:
                     model = Model()
                     models.append(weakref.ref(model))
                     if op == "embed":

tests/test-model-microbatches-735.py

@@ -83,9 +83,9 @@ def runtime(monkeypatch: pytest.MonkeyPatch) -> tuple[ms.ModelServer, RecordingM
 
     server = ms.ModelServer()
     model = RecordingModel()
-    monkeypatch.setattr(server, "_get_embed_model", lambda _tier: model)
-    monkeypatch.setattr(server, "_get_fast_encoder", lambda _tier: model)
-    monkeypatch.setattr(server, "_get_reranker", lambda _name: model)
+    monkeypatch.setattr(server, "_get_embed_model", lambda _tier, deadline=None: model)
+    monkeypatch.setattr(server, "_get_fast_encoder", lambda _tier, deadline=None: model)
+    monkeypatch.setattr(server, "_get_reranker", lambda _name, deadline=None: model)
     monkeypatch.setattr(server, "_write_status_file", lambda: None)
     monkeypatch.setattr(server, "_flush_mps_cache", lambda: None)
     monkeypatch.setattr(mps_utils, "flush_mps_cache", lambda: None)
@@ -331,7 +331,7 @@ def release(self) -> None:
 
     server._lock = ObservingLock()
 
-    def load(identity: str) -> RecordingModel:
+    def load(identity: str, deadline: object = None) -> RecordingModel:
         loads.append(identity)
         return model if identity == "primary" else alternate
 
@@ -598,12 +598,14 @@ def load_order_runtime() -> dict:
     path = Path(__file__).resolve().parents[1] / "truememory/model_server.py"
     functions = {"_array_metadata", "_array_base64_size", "_result_wire_size", "_check_result_size",
                  "_batch_limit", "_checked_batch", "_store_batch_result"}
-    classes = {"_EmbedState", "_RequestDeadline", "_RequestDeadlineExceeded", "_ResultTooLarge"}
+    classes = {"_EmbedState", "_RequestDeadline", "_RequestDeadlineExceeded", "_ResultTooLarge",
+               "_ProcessMemoryRefused"}
     methods = {"handle_request", "_handle_request_inner", "_handle_fast_embed", "_embed_global_order",
                "_embed_slice_indices", "_preflight_embed_result", "_resolve_embed_cache",
                "_request_batch_limit", "_after_request_batches", "_recover_embed_oom_locked",
                "_check_embed_recovery_deadline_locked", "_rerank_global_order", "_rerank_slice_indices",
-               "_preflight_rerank_result", "_publish_embed_state", "_retire_stale_fast_encoder"}
+               "_preflight_rerank_result", "_publish_embed_state", "_retire_stale_fast_encoder",
+               "_check_process_memory"}
     body = [ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0)]
     for node in ast.parse(path.read_text(encoding="utf-8")).body:
         if isinstance(node, ast.FunctionDef) and node.name in functions:
@@ -641,6 +643,7 @@ class TestQwenGlobalOrder(unittest.TestCase):
     def runtime(self) -> tuple[dict, object, OrderModel]:
         module = load_order_runtime()
         server = object.__new__(module["ModelServer"])
+        server._max_rss_bytes = 0
         model = OrderModel(module["np"])
         server._embed_state = module["_EmbedState"](model, "base", "qwen3_256")
         server._lock = threading.Lock()
@@ -659,7 +662,7 @@ def runtime(self) -> tuple[dict, object, OrderModel]:
         server._sticky_cpu = set()
         server.loads = 0
 
-        def load(_tier: str) -> OrderModel:
+        def load(_tier: str, deadline: object = None) -> OrderModel:
             server.loads += 1
             return model
 
@@ -943,7 +946,7 @@ def expire() -> None:
                 if phase == "entry":
                     expires = clock.now
                 elif phase == "model-load":
-                    server._get_embed_model = lambda _tier: (expire(), model)[1]
+                    server._get_embed_model = lambda _tier, deadline=None: (expire(), model)[1]
                 elif phase == "flatten-probe":
                     model._can_flatten_inputs = lambda: (expire(), False)[1]
                 elif phase == "lengths":
@@ -1089,6 +1092,7 @@ def runtime(self) -> tuple[dict, object, PairOrderModel]:
         module = load_order_runtime()
         module["np"] = PairOrderNP()
         server = object.__new__(module["ModelServer"])
+        server._max_rss_bytes = 0
         model = PairOrderModel(module["np"])
         server._lock = threading.Lock()
         server._inference_lock = threading.Lock()
@@ -1103,7 +1107,7 @@ def runtime(self) -> tuple[dict, object, PairOrderModel]:
         server.loads, server.models = [], []
         server.after_load = lambda _model: None
 
-        def load(name: str | None) -> PairOrderModel:
+        def load(name: str | None, deadline: object = None) -> PairOrderModel:
             server.loads.append(name)
             loaded = model if not server.models else PairOrderModel(module["np"])
             loaded.default_prompt = model.default_prompt