fix(tui_gateway): prevent resume stalls during submit and teardown

contributors/emails/theunathi@gmail.com

@@ -0,0 +1,2 @@
+UnathiCodex
+# PR contribution (tui_gateway: keep busy submits resume-safe)

tests/test_tui_gateway_queue_on_busy.py

@@ -10,6 +10,7 @@
 """
 
 import threading
+import time
 import types
 
 from tui_gateway import server
@@ -52,11 +53,14 @@ def test_busy_interrupt_mode_interrupts_and_queues(monkeypatch):
     monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
     calls = {"interrupt": 0}
     agent = types.SimpleNamespace(interrupt=lambda *a, **k: calls.__setitem__("interrupt", calls["interrupt"] + 1))
-    session = _session(agent=agent)
+    session = _session(agent=agent, running=True)
 
     resp = server._handle_busy_submit("r1", "sid", session, "redirect", "ws-1")
 
     assert resp["result"]["status"] == "queued"
+    deadline = time.monotonic() + 1
+    while calls["interrupt"] != 1 and time.monotonic() < deadline:
+        time.sleep(0.01)
     assert calls["interrupt"] == 1
     assert session["queued_prompt"]["text"] == "redirect"
 
@@ -65,7 +69,7 @@ def test_busy_queue_mode_queues_without_interrupting(monkeypatch):
     monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "queue")
     calls = {"interrupt": 0}
     agent = types.SimpleNamespace(interrupt=lambda *a, **k: calls.__setitem__("interrupt", calls["interrupt"] + 1))
-    session = _session(agent=agent)
+    session = _session(agent=agent, running=True)
 
     resp = server._handle_busy_submit("r1", "sid", session, "later", "ws-1")
 
@@ -77,7 +81,7 @@ def test_busy_queue_mode_queues_without_interrupting(monkeypatch):
 def test_busy_steer_mode_injects_when_accepted(monkeypatch):
     monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "steer")
     agent = types.SimpleNamespace(steer=lambda text: True, interrupt=lambda *a, **k: None)
-    session = _session(agent=agent)
+    session = _session(agent=agent, running=True)
 
     resp = server._handle_busy_submit("r1", "sid", session, "nudge", "ws-1")
 
@@ -88,14 +92,48 @@ def test_busy_steer_mode_injects_when_accepted(monkeypatch):
 def test_busy_steer_mode_falls_back_to_queue_when_rejected(monkeypatch):
     monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "steer")
     agent = types.SimpleNamespace(steer=lambda text: False, interrupt=lambda *a, **k: None)
-    session = _session(agent=agent)
+    session = _session(agent=agent, running=True)
 
     resp = server._handle_busy_submit("r1", "sid", session, "nudge", "ws-1")
 
     assert resp["result"]["status"] == "queued"
     assert session["queued_prompt"]["text"] == "nudge"
 
 
+def test_busy_interrupt_does_not_hold_history_lock_or_delay_queue(monkeypatch):
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    interrupt_started = threading.Event()
+    release_interrupt = threading.Event()
+
+    def blocking_interrupt():
+        interrupt_started.set()
+        release_interrupt.wait(timeout=2)
+
+    session = _session(
+        agent=types.SimpleNamespace(interrupt=blocking_interrupt),
+        running=True,
+    )
+
+    started = time.monotonic()
+    resp = server._handle_busy_submit("r1", "sid", session, "keep this", "ws-1")
+
+    assert resp["result"]["status"] == "queued"
+    assert time.monotonic() - started < 0.25
+    assert session["queued_prompt"]["text"] == "keep this"
+    assert interrupt_started.wait(timeout=1)
+    assert session["history_lock"].acquire(timeout=0.25)
+    session["history_lock"].release()
+    release_interrupt.set()
+
+
+def test_busy_helper_retries_when_turn_finished(monkeypatch):
+    monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
+    session = _session(running=False)
+
+    assert server._handle_busy_submit("r1", "sid", session, "run now", "ws-1") is None
+    assert session.get("queued_prompt") is None
+
+
 # ── _drain_queued_prompt ───────────────────────────────────────────────────
 
 def test_drain_fires_queued_prompt_and_claims_running(monkeypatch):

tests/test_tui_gateway_server.py

@@ -2278,6 +2278,50 @@ def test_session_close_commits_memory_and_fires_finalize_hook(monkeypatch):
         server._sessions.pop("sid", None)
 
 
+def test_session_close_releases_resume_lock_before_slow_teardown(monkeypatch):
+    """One slow session finalizer must not stall unrelated session.resume RPCs."""
+    teardown_started = threading.Event()
+    release_teardown = threading.Event()
+    response = {}
+
+    def _slow_teardown(_session, *, end_reason="tui_close"):
+        assert end_reason == "tui_close"
+        teardown_started.set()
+        assert release_teardown.wait(timeout=2.0)
+
+    monkeypatch.setattr(server, "_teardown_session", _slow_teardown)
+    server._sessions["slow-close"] = _session()
+
+    def _close():
+        response.update(
+            server.handle_request(
+                {
+                    "id": "close",
+                    "method": "session.close",
+                    "params": {"session_id": "slow-close"},
+                }
+            )
+        )
+
+    thread = threading.Thread(target=_close)
+    thread.start()
+    acquired = False
+    try:
+        assert teardown_started.wait(timeout=1.0)
+        assert "slow-close" not in server._sessions
+        acquired = server._session_resume_lock.acquire(timeout=0.2)
+        assert acquired, "slow teardown kept the global resume lock held"
+    finally:
+        if acquired:
+            server._session_resume_lock.release()
+        release_teardown.set()
+        thread.join(timeout=2.0)
+        server._sessions.pop("slow-close", None)
+
+    assert not thread.is_alive()
+    assert response["result"] == {"closed": True}
+
+
 def test_ws_orphan_reap_closes_worker_when_session_stays_detached(monkeypatch):
     """A detached WS session past its grace window has its slash_worker closed.
 
@@ -2308,6 +2352,51 @@ def close(self):
         server._sessions.pop("orphan-sid", None)
 
 
+def test_ws_orphan_reap_releases_resume_lock_before_slow_teardown(monkeypatch):
+    """Grace reaping claims under the lock but finalizes after releasing it."""
+    scheduled = {}
+    teardown_started = threading.Event()
+    release_teardown = threading.Event()
+
+    class _Timer:
+        def __init__(self, _delay, callback):
+            scheduled["callback"] = callback
+
+        def start(self):
+            return None
+
+    def _slow_teardown(_session, *, end_reason="tui_close"):
+        assert end_reason == "ws_orphan_reap"
+        teardown_started.set()
+        assert release_teardown.wait(timeout=2.0)
+
+    monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 0.01)
+    monkeypatch.setattr(server.threading, "Timer", _Timer)
+    monkeypatch.setattr(server, "_teardown_session", _slow_teardown)
+    server._sessions["slow-orphan"] = _session(
+        transport=server._detached_ws_transport,
+        running=False,
+    )
+
+    server._schedule_ws_orphan_reap("slow-orphan")
+    thread = threading.Thread(target=scheduled["callback"])
+    thread.start()
+    acquired = False
+    try:
+        assert teardown_started.wait(timeout=1.0)
+        assert "slow-orphan" not in server._sessions
+        acquired = server._session_resume_lock.acquire(timeout=0.2)
+        assert acquired, "orphan teardown kept the global resume lock held"
+    finally:
+        if acquired:
+            server._session_resume_lock.release()
+        release_teardown.set()
+        thread.join(timeout=2.0)
+        server._sessions.pop("slow-orphan", None)
+
+    assert not thread.is_alive()
+
+
 def test_finalize_session_closes_slash_worker(monkeypatch):
     """_finalize_session closes the slash_worker subprocess itself.
 
@@ -9671,17 +9760,20 @@ def close(self):
         server._sessions.pop("live-restart", None)
 
 
-def test_session_close_rpc_delegates_to_close_session_by_id(monkeypatch):
+def test_session_close_rpc_claims_then_tears_down(monkeypatch):
     seen = []
+    claimed = {"session_key": "k"}
+    monkeypatch.setattr(server, "_pop_session_by_id", lambda sid: seen.append(sid) or claimed)
     monkeypatch.setattr(
-        server, "_close_session_by_id",
-        lambda sid, *, end_reason: bool(seen.append((sid, end_reason))) or True,
+        server,
+        "_teardown_popped_session",
+        lambda session, *, end_reason: seen.append((session, end_reason)) or True,
     )
     resp = server.handle_request(
         {"id": "1", "method": "session.close", "params": {"session_id": "s9"}}
     )
     assert resp["result"] == {"closed": True}
-    assert seen == [("s9", "tui_close")]
+    assert seen == ["s9", (claimed, "tui_close")]
 
 
 def test_close_sessions_for_transport_closes_flagged_repoints_rest(monkeypatch):