v43: isolate interactive inference budgets, expose active chat and stop timeout retry loops

Author: Mantiee

CHAT_GUI_PACING_FIX.md

@@ -7,3 +7,11 @@
 This fixes optional vision admission, not every possible source of latency. An admitted image request still uses the configured remote timeout. A chat request can still show `queued` while executing. It is not proof that the dashboard was edited or that the latest layout passed host validation.
 
 Verification: 123 remote-helper/campaign tests passed, including deferred cooldown, thread contention, preserved screenshot and unchanged idle reservation. Ruff and installer shell syntax checks passed. V100/RTX deployment requires the operator upgrade; no real hardware result is claimed here.
+
+## v43: interactive budgets and visible processing
+
+Research policy no longer overrides chat thinking/output settings. Chat uses no thinking, at most 2048 output tokens and at most 60 seconds per HTTP operation. A shared monotonic 180-second deadline follows the client's copies across template/token counting, tool routing, final generation and helper pacing. This is an inference/pacing budget, not a hard wall-clock guarantee for CPU model startup or arbitrary guest tools. Timeout ends the request with a visible failure instead of endlessly retrying the same timeout.
+
+`chat-status` includes a `processing` record for the request currently handled by the live mission PID, including start time and time budget. The persisted state remains `queued` for compatibility while execution is active. The record is cleared when handling finishes. Completion updates only still-queued requests, preserving operator cancellation.
+
+The dashboard request still requires actual execution and host layout validation. These changes do not claim that an HTML edit or profitable training has occurred.

pyproject.toml

@@ -3,7 +3,7 @@ name = "rlms"
 authors = [
     {name = "Alex Zhang", email = "altzhang@mit.edu"},
 ]
-version = "0.1.3+v100.42"
+version = "0.1.3+v100.43"
 description = "Recursive Language Models."
 readme = "README.md"
 license = "MIT"

rlm/clients/llamacpp.py

@@ -114,14 +114,25 @@ def request(self, endpoint: str, data: dict[str, Any] | None = None) -> dict[str
         )
         return result
 
+    def remaining_timeout(self) -> float:
+        deadline = getattr(self, "request_deadline", None)
+        if deadline is None:
+            return self.timeout
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise requests.Timeout("Chat request exceeded its total time budget")
+        return min(self.timeout, remaining)
+
     def http_request(self, endpoint: str, data: dict[str, Any] | None = None) -> dict[str, Any]:
         # No environment proxy routing for local model traffic.
         with requests.Session() as session:
             session.trust_env = False
             response = (
-                session.get(self.base_url + endpoint, timeout=self.timeout)
+                session.get(self.base_url + endpoint, timeout=self.remaining_timeout())
                 if data is None
-                else session.post(self.base_url + endpoint, json=data, timeout=self.timeout)
+                else session.post(
+                    self.base_url + endpoint, json=data, timeout=self.remaining_timeout()
+                )
             )
             response.raise_for_status()
             return response.json()

rlm/v100/mission_chat.py

@@ -6,6 +6,7 @@
 
 import copy
 import json
+import os
 import sqlite3
 import threading
 import time
@@ -60,6 +61,15 @@ def inspect(root: Path, identity: str) -> dict:
     result = dict(row)
     if result["response"]:
         result["response"] = json.loads(result["response"])
+    active_path = root / "research/state/chat-active.json"
+    if active_path.exists():
+        active = json.loads(active_path.read_text())
+        if active.get("id") == identity and result["state"] == "queued":
+            from rlm.v100.mission import status
+
+            mission = status(root)
+            if mission["running"] and mission.get("pid") == active.get("pid"):
+                result["processing"] = active
     return result
 
 
@@ -316,6 +326,12 @@ def respond(root: Path, directory: Path, request: dict, accepted_cpu=None) -> di
             client.enable_thinking = False
             client.activity_actor = "chat-delegate"
             delegated = True
+    client.request_deadline = request.get("deadline", time.monotonic() + 180)
+    client.timeout = min(client.timeout, 60)
+    client.enable_thinking = False
+    client.sampling_args = dict(client.sampling_args)
+    client.sampling_args["max_tokens"] = min(2048, client.sampling_args.get("max_tokens", 2048))
+    client.activity_context = {**client.activity_context, "chat_request_id": request.get("id")}
     with connect(root) as db:
         history = db.execute(
             "SELECT message,response FROM requests WHERE state='completed' ORDER BY rowid DESC LIMIT 3"
@@ -440,6 +456,17 @@ def service_loop(root: Path, directory: Path, stop: threading.Event, accepted_cp
         if row is None:
             continue
         request = dict(row)
+        request["deadline"] = time.monotonic() + 180
+        atomic_json(
+            root / "research/state/chat-active.json",
+            {
+                "id": request["id"],
+                "pid": os.getpid(),
+                "started": time.time(),
+                "time_budget_seconds": 180,
+                "phase": "processing",
+            },
+        )
         try:
             response = (
                 respond(root, directory, request, accepted_cpu)
@@ -448,17 +475,23 @@ def service_loop(root: Path, directory: Path, stop: threading.Event, accepted_cp
             )
             with connect(root) as db:
                 db.execute(
-                    "UPDATE requests SET state='completed',response=?,error=NULL,updated=? WHERE id=?",
+                    "UPDATE requests SET state='completed',response=?,error=NULL,updated=? WHERE id=? AND state='queued'",
                     (json.dumps(response, ensure_ascii=False), time.time(), request["id"]),
                 )
             ActivityLog(root, "controller", "chat").write(
                 "decisions", "user-command-completed", {"id": request["id"], **response}
             )
         except (requests.ConnectionError, requests.Timeout) as error:
+            expired = time.monotonic() >= request["deadline"] or isinstance(error, requests.Timeout)
             with connect(root) as db:
                 db.execute(
-                    "UPDATE requests SET attempts=attempts+1,error=?,updated=? WHERE id=?",
-                    (str(error)[:300], time.time(), request["id"]),
+                    "UPDATE requests SET state=?,attempts=attempts+1,error=?,updated=? WHERE id=? AND state='queued'",
+                    (
+                        "failed" if expired else "queued",
+                        str(error)[:300],
+                        time.time(),
+                        request["id"],
+                    ),
                 )
         except Exception as error:
             with connect(root) as db:
@@ -469,6 +502,8 @@ def service_loop(root: Path, directory: Path, stop: threading.Event, accepted_cp
             ActivityLog(root, "controller", "chat").write(
                 "errors", "user-command-failed", {"id": request["id"], "error": str(error)[:500]}
             )
+        finally:
+            atomic_json(root / "research/state/chat-active.json", {"phase": "idle"})
 
 
 @contextmanager

rlm/v100/remote_helper.py

@@ -180,7 +180,13 @@ def workload_slot(self, wait: bool = True):
         if type(wait) is not bool:
             raise ValueError("Helper wait must be boolean")
         with ExitStack() as scope:
-            if not HELPER_WORKLOAD_LOCK.acquire(blocking=wait):
+            deadline_bound = getattr(self, "request_deadline", None)
+            acquired = (
+                HELPER_WORKLOAD_LOCK.acquire(timeout=self.remaining_timeout())
+                if wait and deadline_bound is not None
+                else HELPER_WORKLOAD_LOCK.acquire(blocking=wait)
+            )
+            if not acquired:
                 raise TimeoutError("Helper busy; optional work deferred")
             scope.callback(HELPER_WORKLOAD_LOCK.release)
             path = None
@@ -194,7 +200,18 @@ def workload_slot(self, wait: bool = True):
                 path.parent.mkdir(parents=True, exist_ok=True)
                 lease = scope.enter_context(path.with_suffix(".lock").open("a"))
                 try:
-                    fcntl.flock(lease, fcntl.LOCK_EX | (0 if wait else fcntl.LOCK_NB))
+                    while True:
+                        try:
+                            fcntl.flock(
+                                lease,
+                                fcntl.LOCK_EX
+                                | (fcntl.LOCK_NB if not wait or deadline_bound else 0),
+                            )
+                            break
+                        except BlockingIOError:
+                            if not wait:
+                                raise
+                            time.sleep(min(0.1, self.remaining_timeout()))
                 except BlockingIOError as error:
                     raise TimeoutError("Helper lease busy; optional work deferred") from error
                 state = json.loads(path.read_text()) if path.exists() else {}
@@ -213,6 +230,7 @@ def workload_slot(self, wait: bool = True):
             if not wait and started < deadline:
                 raise TimeoutError("Helper cooling down; optional work deferred")
             while time.monotonic() < deadline:
+                self.remaining_timeout()
                 time.sleep(max(0, min(1, deadline - time.monotonic())))
             quota = {
                 "path": path,
@@ -255,7 +273,7 @@ def remote_request(self, endpoint: str, data: dict | None = None) -> dict:
             arguments = {} if data is None else {"json": data}
             with method(
                 self.base_url + endpoint,
-                timeout=(5, self.timeout),
+                timeout=(min(5, self.remaining_timeout()), self.remaining_timeout()),
                 allow_redirects=False,
                 stream=True,
                 **arguments,

rlm/v100/research_policy.py

@@ -50,6 +50,9 @@ def choose(root: Path, target: str, thinking: bool, max_tokens: int, batch_token
 
 def apply(client, root: Path):
     result = copy.copy(client)
+    if getattr(client, "activity_actor", "").startswith("chat"):
+        # Interactive latency limits must not inherit long research thinking/output budgets.
+        return result
     if getattr(client, "research_device", None) == "cpu":
         return result
     target = "helper" if hasattr(result, "helper_batch_tokens") else "master"

tests/test_v100_chat_backend.py

@@ -113,6 +113,9 @@ def test_wrong_cpu_endpoint_is_not_reused(tmp_path, monkeypatch):
 
 def test_chat_selects_same_accepted_master_during_gpu_training(tmp_path, monkeypatch):
     accepted, cpu, _ = admitted(tmp_path, monkeypatch)
+    cpu.timeout = 600
+    cpu.sampling_args = {"max_tokens": 1024}
+    cpu.activity_context = {}
     atomic_json(tmp_path / "serving-active.json", accepted)
     monkeypatch.setattr(
         mission,

tests/test_v100_mission_evidence.py

@@ -147,7 +147,13 @@ def test_ordinary_chat_gets_host_evidence_before_model_can_guess(tmp_path, monke
     monkeypatch.setattr(
         mission_chat, "load_profile", lambda *args: {"runtime": {}, "server": {"model": str(model)}}
     )
-    client = SimpleNamespace(request=lambda *args: {"model_path": str(model)}, model_name="v100")
+    client = SimpleNamespace(
+        request=lambda *args: {"model_path": str(model)},
+        model_name="v100",
+        timeout=120,
+        sampling_args={"max_tokens": 2048},
+        activity_context={},
+    )
     monkeypatch.setattr(competition, "helper_client", lambda *args: client)
     seen = []
 

tests/test_v100_remote_helper.py

@@ -834,3 +834,40 @@ def get(self, url, **kwargs):
         client().remote_request("/api/tags")
     with pytest.raises(ValueError, match="management"):
         client().remote_request("/api/pull", {"model": remote_helper.MODEL})
+
+
+def test_chat_deadline_expires_before_any_http_request(monkeypatch):
+    instance = LlamaCppClient(timeout=600)
+    instance.request_deadline = 10
+    monkeypatch.setattr(remote_helper.time, "monotonic", lambda: 11)
+    with pytest.raises(requests.Timeout, match="total time budget"):
+        instance.http_request("/props")
+
+
+def test_chat_deadline_bounds_timeout_and_helper_cooldown(transport, tmp_path, monkeypatch):
+    instance = client(activity_root=str(tmp_path))
+    clock = {"now": 0.0}
+    monkeypatch.setattr(remote_helper.time, "monotonic", lambda: clock["now"])
+    monkeypatch.setattr(
+        remote_helper.time, "sleep", lambda seconds: clock.update(now=clock["now"] + seconds)
+    )
+    with instance.workload_slot() as quota:
+        instance.reserve_helper_idle(quota, 600)
+    instance.request_deadline = 3
+    assert instance.remaining_timeout() == min(instance.timeout, 3)
+    with pytest.raises(requests.Timeout, match="total time budget"):
+        with instance.workload_slot():
+            pytest.fail("Cooldown outlived the chat deadline")
+    assert clock["now"] == 3
+    assert not transport[0]
+
+
+def test_chat_does_not_inherit_research_thinking_budget(tmp_path):
+    from rlm.v100.research_policy import apply, choose
+
+    choose(tmp_path, "master", True, 8192, 512)
+    instance = LlamaCppClient(sampling_args={"max_tokens": 2048}, enable_thinking=False)
+    instance.activity_actor = "chat"
+    actual = apply(instance, tmp_path)
+    assert actual.enable_thinking is False
+    assert actual.sampling_args["max_tokens"] == 2048

tests/test_v100_upgrade_audit.py

@@ -565,3 +565,26 @@ def server(*args):
     )
     assert verdict["passed"] is improves
     assert (selected is source) is not improves
+
+
+def test_chat_timeout_is_terminal_and_clears_processing_marker(tmp_path, monkeypatch):
+    import requests
+
+    from rlm.v100 import mission_chat
+
+    identity = mission_chat.submit(tmp_path, "Edit the dashboard")
+    stop = threading.Event()
+
+    def reply(root, directory, request):
+        active = json.loads((root / "research/state/chat-active.json").read_text())
+        assert active["id"] == identity and active["phase"] == "processing"
+        stop.set()
+        raise requests.Timeout("Read timed out")
+
+    monkeypatch.setattr(mission_chat, "respond", reply)
+    mission_chat.service(tmp_path, tmp_path, stop)
+    actual = mission_chat.inspect(tmp_path, identity)
+    assert actual["state"] == "failed" and actual["attempts"] == 1
+    assert json.loads((tmp_path / "research/state/chat-active.json").read_text()) == {
+        "phase": "idle"
+    }