serve: an engine left behind by an earlier run is waited for until it has given its memory back (#4051)

The server ends an engine an earlier run lost (strata_control/procs.py) before it
starts the next one, and waited 10 s after the kill.  An engine with tens of GB
of pinned memory takes longer to give it back, so the new engine's memory guard
saw 50 GiB still in use and loaded a 40 GiB arena into 10 GiB.  The wait is now
up to 120 s, with a progress line every 10 s; one still there after that is said
in plain words (what it is, why starting now is risky, what to do) and the start
goes on: nothing is refused, the engine waits for RAM itself and warns.

Also documents what the parser repairs (#4078 #4427 #4077) and the arena RAM wait
in docs/DETAILS.md.  Tests: serve/test_leftover_engines.py and a fake-clock case
in tests/terminal/test_procs.py.

docs/DETAILS.md

@@ -501,6 +501,16 @@ applies at the next start. Network, key, MCP and program keys are not editable t
   guessed: `bash` is not `shell`, and a reordered name (`server1_label_values` for `label_values_mcp_server1`) is not the declared one.
   A call whose envelope drifted by a token (`<=NAME>`,
   `<function:NAME>` or `<NAME>` instead of `<function=NAME>`) is read as the call when NAME is a declared tool.
+- **What the parser repairs, and what it leaves.** A string argument that quotes a whole tool call, or repeats its own
+  `</parameter><parameter=NAME>`, stays exact (#4078). A call whose `</function>` was left out is read as the call (#4427), in the
+  answer and inside the thinking. An argument the tool declares as an array or an object, or declares nothing for, is read as
+  written when it has a slip in it: a trailing or doubled comma, single quotes or `True`/`False`/`None`, a bare string between two
+  members of an object (`"status": "pending", "in_progress", "pending": false`), brackets left open at the end. The server window
+  says `N tool call argument(s) repaired` and `/metrics/prometheus` counts them in `strata:tool_args_repaired_total`. Text that
+  is not an array or an object, a string cut short and a string-typed parameter are never touched. After the thinking has ended at
+  a closing think tag, a second one at the start of a line in the answer is dropped (the model quoted the tag while thinking, #4077):
+  the call that follows is kept. The reasoning between the two went out as the answer already, because a stream cannot take text
+  back.
 - **A call that came back as text because its name was not declared (#3796).** The text stays exactly as the model wrote it and
   `finish_reason` stays `stop` (or `end_turn` on `/v1/messages`). Next to it the reply carries a list, only when there is something in it:
   `"strata_rejected_tool_calls": [{"name": "bash", "reason": "undeclared"}]` (at the top level of a Chat Completions reply, in
@@ -698,6 +708,15 @@ the engine (v2 `memory.max` and `memory.high`, v1 `memory.limit_in_bytes`, less
 `--memory-limit-mib` cap. The startup line `strata generate: memory guard: limit X GiB (source: flag|cgroup|meminfo) ...` names
 the source. It covers parking and session files only, not the startup sizing of the arena, the resident budget or the Windows commit check.
 
+**When the arena does not fit in the RAM that is free (#4051).** The engine checks the RAM before it reads the experts into the
+arena. A memory limit below the arena (a container) stops the start with the numbers. Less free RAM than the arena (an engine
+that was still exiting, another program) is waited for up to 20 s, and the log says so (`STRATA_ARENA_RAM_WAIT_S` changes the
+seconds, `0` loads at once); if it is still short the load goes on with a warning that says how much, never a refusal. Below half
+of the arena the warning adds that the whole PC can stall in memory reclaim, and that a systemd `MemoryHigh=` below the RAM keeps
+one program from taking all of it. The server, before it starts an engine, ends one that an earlier run left behind and now waits up
+to 120 s for it to be gone (it used to wait 10 s): an engine with tens of GB of pinned memory needs longer to give it back, and the
+next one started beside it ran the PC out of RAM (#4051).
+
 Snapshots hold running state, checkpoints, used K/V pages and draft-layer K/V. They add host RAM, not VRAM. Oldest parked entries
 are evicted first. A new conversation that starts at a checkpoint inside a parked one (sub-agents that share a system prompt)
 restores that prefix and leaves the parked conversation in place (the log says "borrowed"). Oversized snapshots or host

serve/server.py

@@ -128,17 +128,29 @@ def rttime_limit_note() -> str:
             "systemd-run --user --collect -p LimitRTTIME=infinity -- strata serve ...")
 
 
+LEFTOVER_SETTLE_S = 120.0       # #4051: how long an engine ended at a start is waited for to give its memory back
+
+
 def end_leftover_engines(exe: str) -> None:
     """Before an engine starts: end the ones a lost server left running (strata_control/procs.py). Only a process
     this install recorded whose starter is gone, or an orphan of exactly this executable running --serve, and only
-    after its PID, start time and command line are checked again at the moment of the signal. Never raises."""
+    after its PID, start time and command line are checked again at the moment of the signal. Never raises.
+    #4051: the new engine is not started while the old one still holds its memory (tens of GB take longer than the
+    10 s this used to wait): a 40 GiB arena loaded beside 50 GiB the old engine had not returned put a 60 GiB PC into
+    memory reclaim for good.  One still there after LEFTOVER_SETTLE_S is said in plain words, and the start goes on
+    (the engine waits for the RAM itself and warns; nothing is refused)."""
     if os.environ.get("STRATA_KEEP_LEFTOVERS") == "1":
         return
     try:
         from strata_control import procs
         say = lambda text: print(text, flush=True)        # noqa: E731
-        procs.end_leftovers(say=say)
-        procs.end_orphans(exe, say=say)
+        stuck = [r["pid"] for r in procs.end_leftovers(say=say, settle=LEFTOVER_SETTLE_S)["stuck"]]
+        stuck += [r["pid"] for r in procs.end_orphans(exe, say=say, settle=LEFTOVER_SETTLE_S) if r["outcome"] == "stuck"]
+        if stuck:
+            print(f"[strata] WARNING: the engine from an earlier run (process {', '.join(map(str, stuck))}) was told to stop "
+                  f"but still holds its memory after {LEFTOVER_SETTLE_S:.0f} s. Starting the next one now can leave this PC "
+                  "short of RAM and very slow: wait until that process is gone (it is shown by `ps`/Task Manager), "
+                  "or restart the PC if it does not go.", flush=True)
     except Exception as e:  # noqa: BLE001 - a failed check must not stop a start
         print(f"[strata] could not check for engines left by an earlier run: {type(e).__name__}", flush=True)
 

serve/test_leftover_engines.py

@@ -0,0 +1,57 @@
+"""#4051: an engine an earlier run left behind is waited for until it has given its memory back.
+
+The new engine's arena is as big as the old one's; started beside it, a 60 GiB PC went into memory reclaim for good.  The
+server ends the old engine before it starts the next (strata_control.procs) and now waits for it to be gone for as long as
+a big engine needs; one that is still there after that is said in plain words, and the start goes on (nothing is refused).
+
+    python -m unittest serve.test_leftover_engines -v
+"""
+import contextlib
+import io
+import unittest
+from unittest import mock
+
+from serve import server
+from strata_control import procs
+
+
+class LeftoverEngines(unittest.TestCase):
+    def run_it(self, leftovers=None, orphans=None, env=None):
+        out = io.StringIO()
+        with mock.patch.object(procs, "end_leftovers", return_value=leftovers or {"stuck": []}) as ends, \
+                mock.patch.object(procs, "end_orphans", return_value=orphans or []) as orph, \
+                mock.patch.dict("os.environ", env or {}, clear=False), contextlib.redirect_stdout(out):
+            server.end_leftover_engines("/x/strata")
+        return out.getvalue(), ends, orph
+
+    def test_the_wait_is_long_enough_for_a_big_engine(self):
+        _, ends, orph = self.run_it()
+        self.assertGreaterEqual(server.LEFTOVER_SETTLE_S, 60)
+        self.assertEqual(ends.call_args.kwargs["settle"], server.LEFTOVER_SETTLE_S)
+        self.assertEqual(orph.call_args.kwargs["settle"], server.LEFTOVER_SETTLE_S)
+
+    def test_nothing_left_says_nothing(self):
+        text, _, _ = self.run_it(orphans=[{"pid": 7, "outcome": "ended"}])
+        self.assertEqual(text, "")
+
+    def test_an_engine_that_is_still_there_is_said_and_the_start_goes_on(self):
+        text, _, _ = self.run_it(leftovers={"stuck": [{"pid": 4242, "role": "engine"}]},
+                                 orphans=[{"pid": 4343, "outcome": "stuck"}])
+        self.assertIn("4242, 4343", text)
+        self.assertIn("WARNING", text)
+        self.assertIn("wait until that process is gone", text)
+
+    def test_keep_leftovers_leaves_them_alone(self):
+        _, ends, orph = self.run_it(env={"STRATA_KEEP_LEFTOVERS": "1"})
+        ends.assert_not_called()
+        orph.assert_not_called()
+
+    def test_a_failed_check_never_stops_a_start(self):
+        out = io.StringIO()
+        with mock.patch.object(procs, "end_leftovers", side_effect=RuntimeError("x")), contextlib.redirect_stdout(out):
+            server.end_leftover_engines("/x/strata")
+        self.assertIn("could not check", out.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()

strata_control/procs.py

@@ -302,8 +302,10 @@ def _signal_verified(record: dict, sig: int) -> bool:
     return True
 
 
-def end_one(record: dict, *, grace: float = 8.0) -> str:
-    """End one verified leftover: a polite signal where the OS has one, then a kill after `grace` seconds.
+def end_one(record: dict, *, grace: float = 8.0, settle: float = 10.0, say=None) -> str:
+    """End one verified leftover: a polite signal where the OS has one, then a kill after `grace` seconds, then up to
+    `settle` seconds for it to be gone (#4051: an engine holding tens of GB of pinned memory can take longer than the
+    10 s that were enough for a small process; `say` is told every ~10 s while it is waited for).
     Returns 'gone' (already not running), 'ended' or 'stuck'."""
     pid = record["pid"]
     if not is_same(pid, record["token"], record["cmd"]):
@@ -322,23 +324,29 @@ def end_one(record: dict, *, grace: float = 8.0) -> str:
     else:
         if is_same(pid, record["token"], record["cmd"]):
             subprocess.run(["taskkill", "/F", "/PID", str(pid)], capture_output=True)      # one PID, verified just above
-    deadline = time.monotonic() + 10
+    started = time.monotonic()
+    deadline = started + settle
+    told = 0
     while time.monotonic() < deadline and running():
         time.sleep(0.1)
+        waited = int((time.monotonic() - started) // 10)
+        if say is not None and waited > told:
+            told = waited
+            say(f"[strata] still waiting for process {pid} to give its memory back ({told * 10} s)")
     if running():
         return "stuck"
     drop_record(pid)
     return "ended"
 
 
-def end_leftovers(profile: str | None = None, *, grace: float = 8.0, say=None) -> dict:
+def end_leftovers(profile: str | None = None, *, grace: float = 8.0, settle: float = 10.0, say=None) -> dict:
     """End every verified leftover (children first: a record whose parent is itself a leftover goes last)."""
     to_end, unverifiable = leftovers(profile)
     children_first = sorted(to_end, key=lambda r: r["written_at"], reverse=True)
     result = {"ended": [], "gone": [], "stuck": [], "unverifiable": [
         {"pid": r["pid"], "role": r.get("role")} for r in unverifiable]}
     for record in children_first:
-        outcome = end_one(record, grace=grace)
+        outcome = end_one(record, grace=grace, settle=settle, say=say)
         result[outcome].append({"pid": record["pid"], "role": record.get("role"), "profile": record.get("profile")})
         if say is not None and outcome != "gone":
             say(f"[strata] leftover {record.get('role', 'process')} {record['pid']} {outcome}")
@@ -388,12 +396,12 @@ def orphans_of(engine_exe: str) -> list[Ident]:
     return found
 
 
-def end_orphans(engine_exe: str, *, grace: float = 8.0, say=None) -> list[dict]:
+def end_orphans(engine_exe: str, *, grace: float = 8.0, settle: float = 10.0, say=None) -> list[dict]:
     """End the orphans of `engine_exe` (see `orphans_of`), re-verified one by one."""
     ended = []
     for ident in orphans_of(engine_exe):
         record = {"pid": ident.pid, "token": ident.token, "cmd": list(ident.cmd), "role": "engine", "written_at": 0}
-        outcome = end_one(record, grace=grace)
+        outcome = end_one(record, grace=grace, settle=settle, say=say)
         ended.append({"pid": ident.pid, "outcome": outcome})
         if say is not None and outcome != "gone":
             say(f"[strata] an engine left behind by an earlier run (pid {ident.pid}) {outcome}")

tests/terminal/test_procs.py

@@ -189,6 +189,33 @@ class SubreaperOrphans(unittest.TestCase):
             self.assertEqual(procs.end_one(record, grace=5), "ended")
         self.assertGreaterEqual(same.call_count, 1)
 
+    def slow_exit(self, exits_after, **kw):
+        """end_one of a process that is still there `exits_after` virtual seconds after the kill (a fake clock)."""
+        record = {"pid": 4243, "token": "t1", "cmd": ["engine", "--serve"]}
+        now = [0.0]
+        said = []
+
+        def sleep(s):
+            now[0] += s
+
+        def token(pid):
+            return "t1" if now[0] < exits_after else None
+        with mock.patch.object(procs, "WINDOWS", False), mock.patch.object(procs, "is_same", side_effect=lambda *a: token(0) is not None), \
+                mock.patch.object(procs, "_signal_verified"), mock.patch.object(procs, "start_token", side_effect=token), \
+                mock.patch.object(procs, "drop_record"), mock.patch.object(procs.signal, "SIGKILL", 9, create=True), \
+                mock.patch.object(procs.time, "sleep", side_effect=sleep), \
+                mock.patch.object(procs.time, "monotonic", side_effect=lambda: now[0]):
+            return procs.end_one(record, say=said.append, **kw), said
+
+    def test_an_engine_that_needs_longer_than_ten_seconds_to_give_its_memory_back_is_waited_for(self):
+        # #4051: 50 GiB of pinned memory is released by the kernel after the signal; the next engine started beside it
+        outcome, _ = self.slow_exit(25.0)                              # the old limit: 8 s grace + 10 s
+        self.assertEqual(outcome, "stuck")
+        outcome, said = self.slow_exit(40.0, settle=120.0)
+        self.assertEqual(outcome, "ended")
+        self.assertTrue(said and "still waiting for process 4243" in said[0], said)
+        self.assertEqual(self.slow_exit(40.0, settle=10.0)[0], "stuck")
+
 
 if __name__ == "__main__":
     unittest.main()