From 7910d266db6e482187bb47da4b8255f019a105df Mon Sep 17 00:00:00 2001 From: serversdown Date: Sun, 12 Jul 2026 01:28:10 +0000 Subject: [PATCH] feat(chat): client turn-id makes fire-and-forget bulletproof MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brian fires a quick message then locks his phone to go play the hand — which drops the SSE stream, and on wake the UI re-fires via the blocking fallback. The prior (session, message) + 20s window caught the near-simultaneous case but not a re-fire minutes later. Now the UI stamps each send with a unique turnId (crypto.randomUUID) and carries the SAME id on both the stream and the fallback; the server dedupes on it. Bulletproof regardless of how long he's away — lock for an hour, come back, still exactly one execution and one log — and a genuinely new send gets a fresh id so nothing legit is swallowed. Id-keyed turns keep a long (1h) window; requests without an id keep the short (session, msg) window for near-simultaneous dupes. Verified end-to-end: two POSTs with the same turnId → one reply, one persisted exchange pair (the duplicate reused the owner's result). 9 dedup tests; suite 235. Co-Authored-By: Claude Opus 4.8 (1M context) --- lyra/chat.py | 61 +++++++++++++++++++++++--------------- lyra/web/server.py | 8 +++-- lyra/web/static/index.html | 9 +++++- tests/test_turn_dedup.py | 26 +++++++++++++++- 4 files changed, 76 insertions(+), 28 deletions(-) diff --git a/lyra/chat.py b/lyra/chat.py index b9a08bc..bfeb911 100644 --- a/lyra/chat.py +++ b/lyra/chat.py @@ -21,31 +21,41 @@ MAX_TOOL_ROUNDS = 5 # cap tool-call iterations per turn # --- turn de-duplication -------------------------------------------------- # The web UI hits TWO endpoints for one message: it POSTs the SSE stream, and if -# nothing streams to the browser (iOS can't read a fetch stream body → the fetch -# throws in ~1s) it falls back to the blocking endpoint. But the server-side stream -# runs to completion regardless, so BOTH turns execute — double-persisting the -# message and (once logging became guaranteed) double-logging the hand. This guard -# makes a turn idempotent: the first request for a given (session, message) owns it; -# a duplicate that arrives while it's in flight waits for and reuses that result -# instead of running a second full turn. Window is short so a genuine re-send later -# still runs fresh. -_TURN_TTL = 20.0 +# nothing streams to the browser (a dropped connection — most often because Brian +# locks his phone to go play the hand) it falls back to the blocking endpoint. But +# the server-side stream runs to completion regardless, so BOTH turns execute — +# double-persisting the message and double-logging the hand. This guard makes a turn +# idempotent: the first request owns it; a duplicate reuses the owner's result +# instead of running a second full turn. +# +# The UI stamps each send with a unique turn_id and passes the SAME id on the stream +# AND the fallback, so we dedupe on that — bulletproof no matter how long he's away +# (a genuine new message gets a fresh id, so nothing legit is ever swallowed). Requests +# with no id fall back to a short (session, message) window for near-simultaneous dupes. +_TURN_TTL_ID = 3600.0 # id-keyed: unique per send, so keep it long for fire-and-forget +_TURN_TTL_MSG = 20.0 # (session, msg) keyed: short — only near-simultaneous dupes _turn_lock = threading.Lock() -_turns: dict[tuple, dict] = {} # (session_id, msg) -> {event, reply, ts} +_turns: dict[tuple, dict] = {} # key -> {event, reply, ts, ttl} -def _claim_turn(session_id: str, user_msg: str): +def _turn_key(session_id: str, user_msg: str, turn_id: str | None): + if turn_id: + return ("tid", turn_id), _TURN_TTL_ID + return (session_id, (user_msg or "").strip()), _TURN_TTL_MSG + + +def _claim_turn(session_id: str, user_msg: str, turn_id: str | None = None): """(is_owner, rec). Owner executes the turn then calls _finish_turn; a non-owner - (the near-simultaneous duplicate) waits on rec['event'] and reuses rec['reply'].""" - key = (session_id, (user_msg or "").strip()) + (a duplicate of the same send) waits on rec['event'] and reuses rec['reply'].""" + key, ttl = _turn_key(session_id, user_msg, turn_id) now = time.monotonic() with _turn_lock: - for k in [k for k, r in _turns.items() if now - r["ts"] > _TURN_TTL]: + for k in [k for k, r in _turns.items() if now - r["ts"] > r["ttl"]]: del _turns[k] rec = _turns.get(key) if rec is not None: return False, rec - rec = {"event": threading.Event(), "reply": None, "ts": now} + rec = {"event": threading.Event(), "reply": None, "ts": now, "ttl": ttl} _turns[key] = rec return True, rec @@ -56,8 +66,11 @@ def _finish_turn(rec: dict, reply: str) -> None: rec["event"].set() +_AWAIT_TIMEOUT = 120.0 # a duplicate waits at most this long for the owner to finish + + def _await_duplicate(rec: dict) -> str: - rec["event"].wait(timeout=_TURN_TTL) + rec["event"].wait(timeout=_AWAIT_TIMEOUT) return rec["reply"] or _TANGLED # Which backends get function-calling tools is config-driven (cfg.tool_backends, # env TOOL_BACKENDS, default "cloud"). The MI50's llama.cpp server only does tools @@ -170,16 +183,16 @@ def _voice_pass(messages, draft: str, backend: Backend, model: str | None) -> st def respond(session_id: str, user_msg: str, backend: Backend = "cloud", - model_override: str | None = None) -> str: + model_override: str | None = None, turn_id: str | None = None) -> str: """Produce Lyra's reply to a single user message and persist the exchange.""" cfg = config.load() model = _resolve_model(backend, model_override, cfg) logbus.log("info", "chat request", session=session_id, backend=backend, model=model, embed=cfg.embed_backend) - # A concurrent duplicate (the UI's stream + blocking fallback for one message) - # reuses the owner's result instead of running a second full turn. - is_owner, rec = _claim_turn(session_id, user_msg) + # A duplicate of the same send (the UI's stream + blocking fallback) reuses the + # owner's result instead of running a second full turn. + is_owner, rec = _claim_turn(session_id, user_msg, turn_id) if not is_owner: logbus.log("info", "duplicate turn deduped", session=session_id, path="respond") return _await_duplicate(rec) @@ -211,7 +224,7 @@ def respond(session_id: str, user_msg: str, backend: Backend = "cloud", def respond_stream(session_id: str, user_msg: str, backend: Backend = "cloud", - model_override: str | None = None): + model_override: str | None = None, turn_id: str | None = None): """Streaming generator version of `respond`. Yields ("delta", text), ("tool", name), and a final ("done", reply). Same side effects as `respond`.""" cfg = config.load() @@ -219,9 +232,9 @@ def respond_stream(session_id: str, user_msg: str, backend: Backend = "cloud", logbus.log("info", "chat request (stream)", session=session_id, backend=backend, model=model, embed=cfg.embed_backend) - # A concurrent duplicate (this stream + the UI's blocking fallback for one - # message) reuses the owner's result instead of running a second full turn. - is_owner, rec = _claim_turn(session_id, user_msg) + # A duplicate of the same send (this stream + the UI's blocking fallback) reuses + # the owner's result instead of running a second full turn. + is_owner, rec = _claim_turn(session_id, user_msg, turn_id) if not is_owner: logbus.log("info", "duplicate turn deduped", session=session_id, path="stream") reply = _await_duplicate(rec) diff --git a/lyra/web/server.py b/lyra/web/server.py index c991764..b8afdd0 100644 --- a/lyra/web/server.py +++ b/lyra/web/server.py @@ -273,11 +273,13 @@ def create_app() -> FastAPI: user_msg = _last_user_message(body.get("messages", [])) model_override = body.get("model") or None + turn_id = body.get("turnId") or None memory.ensure_session(session_id) if body.get("mode"): memory.set_session_mode(session_id, body["mode"]) try: - reply = await asyncio.to_thread(chat.respond, session_id, user_msg, backend, model_override) + reply = await asyncio.to_thread(chat.respond, session_id, user_msg, backend, + model_override, turn_id) except Exception as exc: logbus.log("error", "chat failed", session=session_id, error=str(exc)) reply = f"[error] {exc}" @@ -305,6 +307,7 @@ def create_app() -> FastAPI: backend = _backend_for(body.get("backend")) user_msg = _last_user_message(body.get("messages", [])) model_override = body.get("model") or None + turn_id = body.get("turnId") or None memory.ensure_session(session_id) if body.get("mode"): memory.set_session_mode(session_id, body["mode"]) @@ -316,7 +319,8 @@ def create_app() -> FastAPI: def produce(): try: - for event in chat.respond_stream(session_id, user_msg, backend, model_override): + for event in chat.respond_stream(session_id, user_msg, backend, + model_override, turn_id): loop.call_soon_threadsafe(q.put_nowait, event) except Exception as exc: # surface to the client stream, don't hang logbus.log("error", "chat stream failed", session=session_id, error=str(exc)) diff --git a/lyra/web/static/index.html b/lyra/web/static/index.html index a0db8ab..4826679 100644 --- a/lyra/web/static/index.html +++ b/lyra/web/static/index.html @@ -398,10 +398,17 @@ // live poker session forces the cloud backend regardless of the saved pick. if (mode === "poker_cash") backend = "cloud"; + // One id per send, carried on BOTH the stream and the blocking fallback so the + // server runs this turn exactly once even if you lock your phone and it re-fires. + const turnId = (window.crypto && crypto.randomUUID) + ? crypto.randomUUID() + : String(Date.now()) + "-" + Math.random().toString(36).slice(2); + const body = { mode: mode, messages: history, - sessionId: currentSession + sessionId: currentSession, + turnId: turnId }; // Only add backend if in standard mode diff --git a/tests/test_turn_dedup.py b/tests/test_turn_dedup.py index 3f09961..0a46128 100644 --- a/tests/test_turn_dedup.py +++ b/tests/test_turn_dedup.py @@ -64,6 +64,30 @@ def test_fresh_message_after_window_runs_again(): # a completed turn lingers only briefly; simulate expiry and confirm re-ownership o1, rec = chat._claim_turn("s1", "later resend") chat._finish_turn(rec, "first") - rec["ts"] -= chat._TURN_TTL + 1 # age it past the window + rec["ts"] -= chat._TURN_TTL_MSG + 1 # age it past the (session,msg) window o2, _ = chat._claim_turn("s1", "later resend") assert o1 and o2 # a genuine later resend runs fresh + + +# --- client turn-id keying (the fire-and-forget guarantee) ---------------- + +def test_same_turn_id_dedupes_regardless_of_message(): + # the fallback may resend the SAME id; dedupe on the id, not the text + o1, r1 = chat._claim_turn("s1", "a hand", turn_id="tid-123") + o2, r2 = chat._claim_turn("s1", "a hand", turn_id="tid-123") + assert o1 is True and o2 is False and r1 is r2 + + +def test_different_turn_ids_each_own(): + o1, _ = chat._claim_turn("s1", "same text", turn_id="tid-1") + o2, _ = chat._claim_turn("s1", "same text", turn_id="tid-2") + assert o1 and o2 # a genuinely new send never gets swallowed + + +def test_turn_id_window_survives_long_after_the_msg_window(): + # locked-phone case: the re-fire can arrive minutes later and must still dedupe + o1, rec = chat._claim_turn("s1", "big hand", turn_id="tid-lock") + chat._finish_turn(rec, "cached") + rec["ts"] -= chat._TURN_TTL_MSG + 60 # well past the short window, but not the id window + o2, r2 = chat._claim_turn("s1", "big hand", turn_id="tid-lock") + assert o1 and o2 is False and r2["reply"] == "cached"