Coach relay: phone writes go through an ordered outbox; 280 s request cap

The RTDB mirror wrote synchronously on the lane's one worker, so a hanging
Firebase (15 s connect timeouts) would have stalled every desktop request.
PhoneLink now only enqueues; relay.outbox writes on its own thread in the
contract's order (chat reset/lines, state, heartbeat, then the result),
keeps only the latest state and heartbeat, never drops a result or a chat
line, and retries a failed write first, with backoff.

The whole-request 280 s cap is enforced again: Coach.handle sets a
deadline and a coach turn gets what is left of it (at most 240 s), so an
ambiguous message's code check plus the turn cannot outlive the phone's
300 s "No reply" wait. The contract's caps line is back to that wording.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PjvF5DZG9JduSBYKxqyysi

leetcode_coach/leetcode_coach/relay/app.py

@@ -104,6 +104,7 @@ class Relay:  # pylint: disable=too-many-instance-attributes
         )
         phone.attach()
         self.phone = phone
+        self._spawn("rtdb-write", lambda: phone.outbox.run(self.stop_event))
         self._spawn("rtdb-listen", lambda: phone.listen(self.stop_event))
         self._spawn(
             "rtdb-tend", lambda: phone.tend(self.stop_event, self.stop_event.wait)
@@ -133,6 +134,8 @@ class Relay:  # pylint: disable=too-many-instance-attributes
         """Stop taking requests, let the running one finish, close the port."""
         self.stop_event.set()
         self.lane.stop()
+        if self.phone is not None:
+            self.phone.outbox.wake()
         if any(t.name == "http" for t in self._threads):
             self.server.shutdown()  # blocks unless serve_forever runs
         self.server.server_close()

leetcode_coach/leetcode_coach/relay/chat.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 import logging
 import os
 import re
+import time
 from typing import TYPE_CHECKING, Any, Final
 import urllib.error
 
@@ -33,6 +34,9 @@ if TYPE_CHECKING:
 _logger: Final = logging.getLogger(__name__)
 
 TURN_TIMEOUT_S: Final = 240.0
+REQUEST_CAP_S: Final = 280.0
+"""A whole request; a coach turn gets what is left of it (at most 240 s)."""
+MIN_TURN_S: Final = 15.0
 RESETS_RE: Final = re.compile(r"resets\s+([^\n]+?)\s*$", re.IGNORECASE | re.MULTILINE)
 TurnError = CoachError
 """A coach turn produced no reply (``model_error`` / ``claude_unavailable``)."""
@@ -61,9 +65,20 @@ class CoachChat:
         """Bind the claude call and the local model (both injectable)."""
         self._ask = ask
         self._local_chat = local_chat
+        self.monotonic: Callable[[], float] = time.monotonic
+        self.deadline = float("inf")
+        """When the running request hits `REQUEST_CAP_S` (set by `Coach.handle`)."""
+
+    def budget(self) -> float:
+        """Seconds this turn may take; `model_error` when too few are left."""
+        left = self.deadline - self.monotonic()
+        if left < MIN_TURN_S:
+            msg = "The PC ran out of time for this turn; send it again."
+            raise TurnError(msg, code="model_error", retry=True)
+        return min(TURN_TIMEOUT_S, left)
 
     def _bounded(self, messages: list[dict[str, str]]) -> local_backend.LocalReply:
-        return self._local_chat(messages, timeout=TURN_TIMEOUT_S)
+        return self._local_chat(messages, timeout=self.budget())
 
     def open(self, session: Session) -> str:
         """The coach's opening reply for a fresh session."""
@@ -79,7 +94,7 @@ class CoachChat:
         args = ["--model", model()]
         if resume and session.claude_id:
             args += ["--resume", session.claude_id]
-        reply = self._ask(args, text, timeout=TURN_TIMEOUT_S, env=_env(session))
+        reply = self._ask(args, text, timeout=self.budget(), env=_env(session))
         if claude_cli.is_limit_hit(reply):
             reason = str(reply.get("result") or "usage limit")
             _logger.warning("claude usage limit (%s): coaching on locally", reason)

leetcode_coach/leetcode_coach/relay/coach.py

@@ -18,7 +18,7 @@ from typing import TYPE_CHECKING, Any, Final
 
 from leetcode_coach import answer_key, claude_cli, local_backend, toolbox
 from leetcode_coach.relay import config, grading, protocol, runcode, steps, turns
-from leetcode_coach.relay.chat import CoachChat
+from leetcode_coach.relay.chat import REQUEST_CAP_S, CoachChat
 from leetcode_coach.relay.errors import CoachError
 from leetcode_coach.relay.probe import ACCEPTED, AcceptedProbe
 from leetcode_coach.relay.session import Session, grade_line, restore, save
@@ -170,6 +170,7 @@ class Coach:
     def handle(self, req: protocol.Request) -> dict[str, Any]:
         """Run one request; its result (section 2)."""
         with self.lock:
+            self.chat.deadline = self.chat.monotonic() + REQUEST_CAP_S
             try:
                 data = self._dispatch(req)
             except CoachError as exc:

leetcode_coach/leetcode_coach/relay/outbox.py

@@ -0,0 +1,105 @@
+"""The phone mirror's writes, in order, on their own thread.
+
+A hanging Firebase must never stall the desktop: the lane's one worker only
+*enqueues* (chat lines, ``state``, the heartbeat, a result) and this outbox
+writes them. Order is the contract's: chat reset / new lines, then
+``state``, then the heartbeat, then the request's result -- the phone
+refetches chat when ``heartbeat.state_rev`` moves, so a heartbeat landing
+before its chat lines would lose one. ``state`` and the heartbeat keep only
+their latest value (an older one still waiting is dropped, the newer goes to
+the back); results and chat lines are never dropped. A failed write stays at
+the front and is retried after a backoff; nothing behind it overtakes it.
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass
+import logging
+import threading
+from typing import TYPE_CHECKING, Final
+
+from leetcode_coach.relay.rtdb import RtdbError
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+_logger: Final = logging.getLogger(__name__)
+
+BACKOFF_S: Final = (1.0, 2.0, 5.0, 10.0, 30.0)
+IDLE_S: Final = 0.5
+
+
+@dataclass(frozen=True)
+class Write:
+    """One pending write: what it is (for the log) and how to do it."""
+
+    what: str
+    run: Callable[[], None]
+    latest: str | None = None
+    """A key whose older pending writes this one replaces (state, heartbeat)."""
+
+
+class Outbox:
+    """An ordered queue of RTDB writes and the loop that drains it."""
+
+    def __init__(self) -> None:
+        """Start empty; `run` drains on a thread, `drain` in the caller."""
+        self._cv = threading.Condition()
+        self._writes: deque[Write] = deque()
+
+    def put(
+        self, what: str, run: Callable[[], None], latest: str | None = None
+    ) -> None:
+        """Queue one write (replacing a waiting one with the same `latest`)."""
+        with self._cv:
+            if latest is not None:
+                self._writes = deque(w for w in self._writes if w.latest != latest)
+            self._writes.append(Write(what, run, latest))
+            self._cv.notify_all()
+
+    def pending(self) -> list[str]:
+        """What is still waiting, in order."""
+        with self._cv:
+            return [w.what for w in self._writes]
+
+    def drain(self) -> bool:
+        """Write everything waiting; False at the first failure (it stays first)."""
+        while True:
+            with self._cv:
+                if not self._writes:
+                    return True
+                write = self._writes.popleft()
+            try:
+                write.run()
+            except RtdbError as exc:
+                _logger.warning("RTDB %s failed: %s", write.what, exc)
+                with self._cv:
+                    newer = write.latest is not None and any(
+                        w.latest == write.latest for w in self._writes
+                    )
+                    if not newer:
+                        self._writes.appendleft(write)
+                return False
+
+    def run(
+        self, stop: threading.Event, wait: Callable[[float], object] | None = None
+    ) -> None:
+        """Drain until `stop`, backing off after a failed write."""
+        pause = stop.wait if wait is None else wait
+        failures = 0
+        while not stop.is_set():
+            with self._cv:
+                if not self._writes:
+                    self._cv.wait(IDLE_S)
+                    continue
+            if self.drain():
+                failures = 0
+                continue
+            pause(BACKOFF_S[min(failures, len(BACKOFF_S) - 1)])
+            failures += 1
+
+    def wake(self) -> None:
+        """Let a waiting `run` notice its stop event at once."""
+        with self._cv:
+            self._cv.notify_all()

leetcode_coach/leetcode_coach/relay/phone.py

@@ -8,8 +8,9 @@ on the shared lane. Results, ``state`` (a JSON string leaf) and the chat
 lines are written as they change; the heartbeat every `HEARTBEAT_S` and on
 every ``rev`` or busy change. ``requests/{rid}`` goes 24 h after its result.
 
-Firebase being down never stops the relay: every write failure is logged
-and the next change or beat tries again; the desktop keeps being served.
+Writes only ever *enqueue* (`outbox`): the lane's worker never waits on
+Firebase, so a hanging or down database never slows the desktop. A failed
+write is logged and retried in order; the relay never exits over it.
 """
 
 from __future__ import annotations
@@ -22,6 +23,7 @@ from typing import TYPE_CHECKING, Any, Final
 
 from leetcode_coach.relay import protocol, sse
 from leetcode_coach.relay.done import DAY_MS
+from leetcode_coach.relay.outbox import Outbox
 from leetcode_coach.relay.rtdb import SERVER_TIME, RtdbError
 
 if TYPE_CHECKING:
@@ -74,19 +76,11 @@ class PhoneLink:
         self._clock = clock_ms
         self._beat_every = heartbeat_s
         self._rescan_every = rescan_s
-        self._write_lock = threading.Lock()
+        self.outbox = Outbox()
+        self._mirror_lock = threading.Lock()
         self._chat_sid: str | None = None
         self._sent: set[str] = set()
 
-    def _write(self, what: str, call: Callable[[], None]) -> bool:
-        with self._write_lock:
-            try:
-                call()
-            except RtdbError as exc:
-                _logger.warning("RTDB %s failed: %s", what, exc)
-                return False
-        return True
-
     def attach(self) -> None:
         """Listen to the lane and the backend; then mirror everything once."""
         self.lane.on_result.append(self.on_result)
@@ -98,12 +92,14 @@ class PhoneLink:
     def beat(self) -> None:
         """PUT the heartbeat (``beat`` as a server timestamp)."""
         beat = {**self._heartbeat(), "beat": SERVER_TIME}
-        self._write("heartbeat", lambda: self.rtdb.put("heartbeat", beat))
+        put = functools.partial(self.rtdb.put, "heartbeat", beat)
+        self.outbox.put("heartbeat", put, latest="heartbeat")
 
     def put_result(self, rid: str, result: dict[str, Any]) -> None:
         """``requests/{rid}/result`` as a JSON string leaf."""
         text = json.dumps(result, ensure_ascii=False)
-        self._write("result", lambda: self.rtdb.put(f"requests/{rid}/result", text))
+        put = functools.partial(self.rtdb.put, f"requests/{rid}/result", text)
+        self.outbox.put("result", put)
 
     def on_result(self, req: protocol.Request, result: dict[str, Any]) -> None:
         """A finished request: answer it if it came from the phone."""
@@ -111,21 +107,25 @@ class PhoneLink:
             self.put_result(req.rid, result)
 
     def on_state(self, state: dict[str, Any], session: Session | None) -> None:
-        """Mirror ``state``, the new chat lines and the heartbeat."""
+        """Queue the chat lines, then ``state``, then the heartbeat."""
         text = json.dumps(state, ensure_ascii=False)
-        self._write("state", lambda: self.rtdb.put("state", text))
-        sid = None if session is None else session.sid
-        if sid != self._chat_sid and self._write(
-            "chat reset", lambda: self.rtdb.delete("chat")
-        ):
-            self._chat_sid, self._sent = sid, set()
-        if session is not None and sid == self._chat_sid:
+        with self._mirror_lock:
+            sid = None if session is None else session.sid
+            if sid != self._chat_sid:
+                self.outbox.put(
+                    "chat reset", functools.partial(self.rtdb.delete, "chat")
+                )
+                self._chat_sid, self._sent = sid, set()
+            lines = {} if session is None else dict(session.chat)
             new: dict[str, object] = {
-                k: v for k, v in session.chat.items() if k not in self._sent
+                k: v for k, v in lines.items() if k not in self._sent
             }
-            if new and self._write("chat", lambda: self.rtdb.patch("chat", new)):
+            if new:
+                self.outbox.put("chat", functools.partial(self.rtdb.patch, "chat", new))
                 self._sent |= set(new)
-        self.beat()
+            put = functools.partial(self.rtdb.put, "state", text)
+            self.outbox.put("state", put, latest="state")
+            self.beat()
 
     def consider(self, rid: str, node: object) -> None:
         """One ``requests/{rid}`` node: answer, queue, or leave it."""
@@ -183,7 +183,7 @@ class PhoneLink:
             done = self.lane.done.get(str(rid))
             if done is None or now - done.at > DAY_MS:
                 drop = functools.partial(self.rtdb.delete, f"requests/{rid}")
-                self._write("cleanup", drop)
+                self.outbox.put("cleanup", drop)
 
     def listen(self, stop: threading.Event) -> None:
         """The SSE loop; a cancelled stream leaves the rescans to carry on."""

leetcode_coach/leetcode_coach/tests/test_relay_outbox.py

@@ -0,0 +1,123 @@
+"""A hanging Firebase never stalls the desktop; writes land in contract order."""
+
+from __future__ import annotations
+
+import threading
+import time
+from typing import TYPE_CHECKING
+
+from leetcode_coach.relay import outbox as outbox_mod
+from leetcode_coach.relay.done import DoneMap
+from leetcode_coach.relay.lane import Lane
+from leetcode_coach.relay.phone import PhoneLink
+from leetcode_coach.relay.rtdb import RtdbError
+from leetcode_coach.tests._relay_direct import FakeAsk, make_coach, req
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+class Stuck:
+    """An RTDB whose writes hang until released, recording their order."""
+
+    def __init__(self) -> None:
+        self.release = threading.Event()
+        self.calls: list[tuple[str, str]] = []
+
+    def _hang(self, method: str, path: str) -> None:
+        self.release.wait(10)
+        self.calls.append((method, path))
+
+    def put(self, path: str, _value: object) -> None:
+        self._hang("put", path)
+
+    def patch(self, path: str, _value: object) -> None:
+        self._hang("patch", path)
+
+    def delete(self, path: str) -> None:
+        self._hang("delete", path)
+
+
+def test_the_desktop_is_served_while_rtdb_hangs(tmp_path: Path) -> None:
+    coach = make_coach(tmp_path, FakeAsk([{"result": "Hi?"}]))
+    lane = Lane(
+        coach.handle, DoneMap(tmp_path / "done.jsonl", lambda: 1), clock_ms=lambda: 1
+    )
+    rtdb = Stuck()
+    link = PhoneLink(rtdb, lane, coach, lambda: {"version": 1}, clock_ms=lambda: 1)
+    stop = threading.Event()
+    writer = threading.Thread(target=link.outbox.run, args=(stop,))
+    writer.start()
+    link.attach()
+    t0 = time.monotonic()
+    started = coach.handle(req("session_start", slug="missing-number"))
+    phone_req = req("hint", transport="phone", session=started["data"]["session"])
+    link.on_result(phone_req, {"ok": True})
+    assert started["ok"]
+    assert time.monotonic() - t0 < 2  # never waited on the hanging writes
+    rtdb.release.set()
+    deadline = time.monotonic() + 5
+    while link.outbox.pending() and time.monotonic() < deadline:
+        time.sleep(0.01)
+    stop.set()
+    link.outbox.wake()
+    writer.join(timeout=5)
+    tail = rtdb.calls[-5:]
+    assert tail == [
+        ("delete", "chat"),
+        ("patch", "chat"),
+        ("put", "state"),
+        ("put", "heartbeat"),
+        ("put", f"requests/{phone_req.rid}/result"),
+    ]
+
+
+def test_a_failing_write_backs_off_and_a_stale_one_is_dropped() -> None:
+    box = outbox_mod.Outbox()
+    tries: list[str] = []
+
+    def fail() -> None:
+        tries.append("state-1")
+        msg = "PUT state: HTTP 503"
+        raise RtdbError(msg)
+
+    box.put("state", fail, latest="state")
+    assert not box.drain()
+    assert box.pending() == ["state"]
+    stop = threading.Event()
+    pauses: list[float] = []
+
+    def recovered() -> None:
+        tries.append("state-2")
+        stop.set()
+
+    def pause(seconds: float) -> None:
+        pauses.append(seconds)
+        box.put("state", recovered, latest="state")
+
+    box.run(stop, pause)
+    assert pauses == [1.0]
+    assert tries == ["state-1", "state-1", "state-2"]
+    assert box.pending() == []
+
+    def fail_then_newer() -> None:
+        box.put("state", lambda: tries.append("state-3"), latest="state")
+        msg = "PUT state: HTTP 503"
+        raise RtdbError(msg)
+
+    box.put("state", fail_then_newer, latest="state")
+    assert not box.drain()
+    assert box.pending() == ["state"]  # the failed, older one was dropped
+    assert box.drain()
+    assert tries[-1] == "state-3"
+
+
+def test_an_idle_outbox_stops_on_wake() -> None:
+    box = outbox_mod.Outbox()
+    stop = threading.Event()
+    runner = threading.Thread(target=box.run, args=(stop,))
+    runner.start()
+    stop.set()
+    box.wake()
+    runner.join(timeout=2)
+    assert not runner.is_alive()

leetcode_coach/leetcode_coach/tests/test_relay_phone.py

@@ -90,6 +90,7 @@ def test_events_at_every_depth_reach_the_lane(link: Any) -> None:
     made.consider(rid(), {"req": _req()})
     limited = rid()
     made.consider(limited, {"req": _req()})
+    assert made.outbox.drain()
     assert "rate_limited" in str(rec.calls[-1][2])
     assert rec.calls[-1][1] == f"requests/{limited}/result"
     assert rec.paths("put").count("heartbeat") == 1
@@ -101,6 +102,7 @@ def test_results_go_only_to_the_phone(link: Any) -> None:
     made.on_result(desktop, {"ok": True})
     phone_req = type("R", (), {"transport": "phone", "rid": rid()})()
     made.on_result(phone_req, {"ok": True})
+    assert made.outbox.drain()
     puts = rec.paths("put")
     assert f"requests/{phone_req.rid}/result" in puts
     assert f"requests/{desktop.rid}/result" not in puts
@@ -111,18 +113,26 @@ def test_write_failures_are_logged_and_retried(
 ) -> None:
     made, rec, _lane = link
     coach = made.coach
+    assert made.outbox.drain()
+    rec.calls.clear()
     rec.fail = {"delete"}
     with caplog.at_level(logging.WARNING):
         started(coach)
+        assert not made.outbox.drain()
     assert "RTDB chat reset failed" in caplog.text
-    assert rec.paths("patch") == []
-    rec.fail = {"patch"}
-    made.on_state(coach.state(), coach.session)
-    assert made._sent == set()
+    assert rec.calls == []  # nothing overtook the failed reset
+    assert made.outbox.pending() == ["chat reset", "chat", "state", "heartbeat"]
+    made.on_state(coach.state(), coach.session)  # newer state/beat replace waiting
+    assert made.outbox.pending() == ["chat reset", "chat", "state", "heartbeat"]
     rec.fail = set()
-    made.on_state(coach.state(), coach.session)
-    assert rec.calls[-2][0] == "patch"
-    assert json.loads(str(rec.calls[0][2]))["schema"] == 1
+    assert made.outbox.drain()
+    assert [(m, p) for m, p, _v in rec.calls] == [
+        ("delete", "chat"),
+        ("patch", "chat"),
+        ("put", "state"),
+        ("put", "heartbeat"),
+    ]
+    assert json.loads(str(rec.calls[2][2]))["schema"] == 1
 
 
 def test_rescan_answers_and_cleans_up(
@@ -149,6 +159,7 @@ def test_rescan_answers_and_cleans_up(
         "junk": 5,
     }
     made.rescan()
+    assert made.outbox.drain()
     assert sorted(rec.paths("delete")) == sorted(
         [f"requests/{old}", f"requests/{lost}"]
     )

leetcode_coach/leetcode_coach/tests/test_relay_units.py

@@ -169,3 +169,17 @@ def test_env_and_config_seams(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -
     monkeypatch.delenv(config.PORT_ENV)
     assert config.state_dir() == config.DEFAULT_DIR
     assert config.port() == config.DEFAULT_PORT
+
+
+def test_a_turn_gets_what_is_left_of_the_request(tmp_path: Path) -> None:
+    ask = FakeAsk([{"result": "Hi?"}, {"result": "Ok?"}])
+    coach = make_coach(tmp_path, ask)
+    sid = started(coach)
+    assert ask.calls[0]["timeout"] == chat.TURN_TIMEOUT_S
+    ticks = iter([0.0, 100.0, 0.0, 270.0])
+    coach.chat.monotonic = lambda: next(ticks)
+    assert coach.handle(req("chat", session=sid, text="so what then?"))["ok"]
+    assert ask.calls[-1]["timeout"] == 180.0
+    late = coach.handle(req("chat", session=sid, text="and now?"))
+    assert late["code"] == "model_error"
+    assert late["retry"] is True