From f3ecf8ffe4753049527d809299d32ba01d2f3777 Mon Sep 17 00:00:00 2001 From: serversdown Date: Sat, 11 Jul 2026 00:17:48 +0000 Subject: [PATCH] fix(chat): de-duplicate the double turn execution at its source MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The UI POSTs the SSE stream and, when nothing streams to the browser (iOS can't read a fetch-stream body → the fetch throws in ~1s), falls back to the blocking endpoint. But the server-side stream runs to completion regardless, so BOTH turns executed — double-persisting the message and (once logging became guaranteed) double-logging the hand. Make a turn idempotent instead of chasing why the client bails: the first request for a (session, message) owns it; a concurrent duplicate waits on the owner's Event and reuses its reply rather than running a second full turn. respond and respond_stream both claim/await; a finally always releases waiters. Short window so a genuine later resend still runs fresh. Verified with a threaded race: two simultaneous calls, body runs once, both get the same reply. Also fixes the duplicate user-message persistence (the same double-execution) that was polluting reconstructed history. 6 dedup tests + concurrency check; suite 232. Co-Authored-By: Claude Opus 4.8 (1M context) --- lyra/chat.py | 239 +++++++++++++++++++++++++-------------- tests/test_hand_dedup.py | 4 +- tests/test_turn_dedup.py | 69 +++++++++++ 3 files changed, 225 insertions(+), 87 deletions(-) create mode 100644 tests/test_turn_dedup.py diff --git a/lyra/chat.py b/lyra/chat.py index 87937b9..b9a08bc 100644 --- a/lyra/chat.py +++ b/lyra/chat.py @@ -10,11 +10,55 @@ deliberate) and hands back a ready message list + the active mode. Then: """ from __future__ import annotations +import threading +import time + from lyra import config, llm, logbus, memory, mind, modes, poker_prompts, summary from lyra import tools as toolkit from lyra.llm import Backend 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 +_turn_lock = threading.Lock() +_turns: dict[tuple, dict] = {} # (session_id, msg) -> {event, reply, ts} + + +def _claim_turn(session_id: str, user_msg: str): + """(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()) + now = time.monotonic() + with _turn_lock: + for k in [k for k, r in _turns.items() if now - r["ts"] > _TURN_TTL]: + del _turns[k] + rec = _turns.get(key) + if rec is not None: + return False, rec + rec = {"event": threading.Event(), "reply": None, "ts": now} + _turns[key] = rec + return True, rec + + +def _finish_turn(rec: dict, reply: str) -> None: + rec["reply"] = reply + rec["ts"] = time.monotonic() + rec["event"].set() + + +def _await_duplicate(rec: dict) -> str: + rec["event"].wait(timeout=_TURN_TTL) + 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 # when launched with --jinja + a tool-capable model, else it 500s on the tools @@ -133,26 +177,37 @@ def respond(session_id: str, user_msg: str, backend: Backend = "cloud", logbus.log("info", "chat request", session=session_id, backend=backend, model=model, embed=cfg.embed_backend) - turn = mind.assemble(session_id, user_msg, backend, model) - messages = turn.messages - tool_specs = toolkit.specs(turn.mode.tools) if backend in cfg.tool_backends else None - ctx = {"session_id": session_id, "backend": 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) + if not is_owner: + logbus.log("info", "duplicate turn deduped", session=session_id, path="respond") + return _await_duplicate(rec) - # Persist the user turn before the tool loop so its timestamp precedes any - # tool events fired mid-turn (keeps the transcript export in true order). - memory.remember(session_id, "user", user_msg) - reply, tools_run = _mind_loop(messages, backend, model, tool_specs, ctx, session_id) - _ensure_hand_logged(messages, user_msg, turn.msg_type, tools_run, backend, model, ctx, session_id) - mouth = _mouth_target(cfg, backend, model) - if mouth and reply: - reply = _voice_pass(messages, reply, *mouth) - if not reply: - reply = _TANGLED - logbus.log("info", "reply", session=session_id, chars=len(reply), voiced=bool(mouth)) + reply = _TANGLED + try: + turn = mind.assemble(session_id, user_msg, backend, model) + messages = turn.messages + tool_specs = toolkit.specs(turn.mode.tools) if backend in cfg.tool_backends else None + ctx = {"session_id": session_id, "backend": backend} - memory.remember(session_id, "assistant", reply) - summary.maybe_summarize_async(session_id) # compact once enough new turns pile up - return reply + # Persist the user turn before the tool loop so its timestamp precedes any + # tool events fired mid-turn (keeps the transcript export in true order). + memory.remember(session_id, "user", user_msg) + reply, tools_run = _mind_loop(messages, backend, model, tool_specs, ctx, session_id) + _ensure_hand_logged(messages, user_msg, turn.msg_type, tools_run, backend, model, ctx, session_id) + mouth = _mouth_target(cfg, backend, model) + if mouth and reply: + reply = _voice_pass(messages, reply, *mouth) + if not reply: + reply = _TANGLED + logbus.log("info", "reply", session=session_id, chars=len(reply), voiced=bool(mouth)) + + memory.remember(session_id, "assistant", reply) + summary.maybe_summarize_async(session_id) # compact once enough new turns pile up + return reply + finally: + _finish_turn(rec, reply) def respond_stream(session_id: str, user_msg: str, backend: Backend = "cloud", @@ -164,73 +219,87 @@ 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) - turn = mind.assemble(session_id, user_msg, backend, model) - messages = turn.messages - tool_specs = toolkit.specs(turn.mode.tools) if backend in cfg.tool_backends else None - ctx = {"session_id": session_id, "backend": backend} - mouth = _mouth_target(cfg, backend, model) + # 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) + if not is_owner: + logbus.log("info", "duplicate turn deduped", session=session_id, path="stream") + reply = _await_duplicate(rec) + yield ("delta", reply) + yield ("done", reply) + return - # Persist the user turn up front (see respond): keeps tool events, which fire - # mid-turn, chronologically after the user message in the exported transcript. - memory.remember(session_id, "user", user_msg) + reply = _TANGLED + try: + turn = mind.assemble(session_id, user_msg, backend, model) + messages = turn.messages + tool_specs = toolkit.specs(turn.mode.tools) if backend in cfg.tool_backends else None + ctx = {"session_id": session_id, "backend": backend} + mouth = _mouth_target(cfg, backend, model) - if mouth is None: - # No separate voice: stream the mind directly (the original path, unchanged). - parts: list[str] = [] - tools_run: list[str] = [] - for _ in range(MAX_TOOL_ROUNDS): - assistant_msg = None - tool_calls = None - for ev, payload in llm.chat_call_stream( - messages, backend=backend, model=model, tools=tool_specs - ): - if ev == "delta": - parts.append(payload) - yield ("delta", payload) - elif ev == "message": - assistant_msg = payload - elif ev == "tool_calls": - tool_calls = payload - if not tool_calls: - break - messages.append(assistant_msg) - for tc in tool_calls: - result = toolkit.dispatch(tc["name"], tc["arguments"], ctx) - memory.add_tool_event(session_id, tc["name"], tc["arguments"], result) - logbus.log("info", "tool call", session=session_id, tool=tc["name"], result=result[:80]) - messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result}) - _maybe_switch_mode(session_id, tc["name"]) - tools_run.append(tc["name"]) - yield ("tool", tc["name"]) - for name in _ensure_hand_logged(messages, user_msg, turn.msg_type, tools_run, - backend, model, ctx, session_id): - yield ("tool", name) - reply = "".join(parts) - if not reply: - reply = _TANGLED - yield ("delta", reply) - else: - # Mind decides + runs tools (non-streamed); mouth re-voices, streamed. - draft, tools_run = _mind_loop(messages, backend, model, tool_specs, ctx, session_id) - tools_run += _ensure_hand_logged(messages, user_msg, turn.msg_type, tools_run, - backend, model, ctx, session_id) - for name in tools_run: - yield ("tool", name) - parts = [] - try: - for ev, payload in llm.chat_call_stream( - mind.voice_messages(messages, draft), backend=mouth[0], model=mouth[1], tools=None - ): - if ev == "delta": - parts.append(payload) - yield ("delta", payload) - except Exception as exc: - logbus.log("error", "voice stream failed", error=str(exc)[:160]) - reply = "".join(parts).strip() or draft or _TANGLED - if not parts: - yield ("delta", reply) + # Persist the user turn up front (see respond): keeps tool events, which fire + # mid-turn, chronologically after the user message in the exported transcript. + memory.remember(session_id, "user", user_msg) - logbus.log("info", "reply", session=session_id, chars=len(reply), voiced=bool(mouth)) - memory.remember(session_id, "assistant", reply) - summary.maybe_summarize_async(session_id) - yield ("done", reply) + if mouth is None: + # No separate voice: stream the mind directly (the original path, unchanged). + parts: list[str] = [] + tools_run: list[str] = [] + for _ in range(MAX_TOOL_ROUNDS): + assistant_msg = None + tool_calls = None + for ev, payload in llm.chat_call_stream( + messages, backend=backend, model=model, tools=tool_specs + ): + if ev == "delta": + parts.append(payload) + yield ("delta", payload) + elif ev == "message": + assistant_msg = payload + elif ev == "tool_calls": + tool_calls = payload + if not tool_calls: + break + messages.append(assistant_msg) + for tc in tool_calls: + result = toolkit.dispatch(tc["name"], tc["arguments"], ctx) + memory.add_tool_event(session_id, tc["name"], tc["arguments"], result) + logbus.log("info", "tool call", session=session_id, tool=tc["name"], result=result[:80]) + messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result}) + _maybe_switch_mode(session_id, tc["name"]) + tools_run.append(tc["name"]) + yield ("tool", tc["name"]) + for name in _ensure_hand_logged(messages, user_msg, turn.msg_type, tools_run, + backend, model, ctx, session_id): + yield ("tool", name) + reply = "".join(parts) + if not reply: + reply = _TANGLED + yield ("delta", reply) + else: + # Mind decides + runs tools (non-streamed); mouth re-voices, streamed. + draft, tools_run = _mind_loop(messages, backend, model, tool_specs, ctx, session_id) + tools_run += _ensure_hand_logged(messages, user_msg, turn.msg_type, tools_run, + backend, model, ctx, session_id) + for name in tools_run: + yield ("tool", name) + parts = [] + try: + for ev, payload in llm.chat_call_stream( + mind.voice_messages(messages, draft), backend=mouth[0], model=mouth[1], tools=None + ): + if ev == "delta": + parts.append(payload) + yield ("delta", payload) + except Exception as exc: + logbus.log("error", "voice stream failed", error=str(exc)[:160]) + reply = "".join(parts).strip() or draft or _TANGLED + if not parts: + yield ("delta", reply) + + logbus.log("info", "reply", session=session_id, chars=len(reply), voiced=bool(mouth)) + memory.remember(session_id, "assistant", reply) + summary.maybe_summarize_async(session_id) + yield ("done", reply) + finally: + _finish_turn(rec, reply) diff --git a/tests/test_hand_dedup.py b/tests/test_hand_dedup.py index 70d5027..f80d7a2 100644 --- a/tests/test_hand_dedup.py +++ b/tests/test_hand_dedup.py @@ -70,7 +70,7 @@ def test_parse_prompt_records_straddles(): # --- hero stack auto-fill from the last logged stack ---------------------- def test_hero_stack_filled_from_last_stack_log(poker, monkeypatch): - sid = poker.start_session(venue="Meadows", stakes="1/3", buy_in=400) + poker.start_session(venue="Meadows", stakes="1/3", buy_in=400) poker.log_stack(275) # his last reported stack monkeypatch.setattr(poker, "parse_hand", lambda *a, **k: {"game": "NLH", "hero_involved": True, @@ -84,7 +84,7 @@ def test_hero_stack_filled_from_last_stack_log(poker, monkeypatch): def test_stated_stack_is_never_overridden(poker, monkeypatch): - sid = poker.start_session(venue="Meadows", buy_in=400) + poker.start_session(venue="Meadows", buy_in=400) poker.log_stack(275) monkeypatch.setattr(poker, "parse_hand", lambda *a, **k: {"game": "NLH", "hero_involved": True, diff --git a/tests/test_turn_dedup.py b/tests/test_turn_dedup.py new file mode 100644 index 0000000..3f09961 --- /dev/null +++ b/tests/test_turn_dedup.py @@ -0,0 +1,69 @@ +"""Turn de-duplication: the UI hits two endpoints for one message (SSE stream + +blocking fallback). Only the first should execute; the duplicate reuses its result.""" +from __future__ import annotations + +import pytest + +from lyra import chat + + +@pytest.fixture(autouse=True) +def _clean_turns(): + chat._turns.clear() + yield + chat._turns.clear() + + +def test_claim_is_owner_once_per_key(): + o1, r1 = chat._claim_turn("s1", "flopped a set") + o2, r2 = chat._claim_turn("s1", "flopped a set") + assert o1 is True and o2 is False + assert r1 is r2 # the duplicate waits on the SAME record + + +def test_different_messages_each_own(): + o1, _ = chat._claim_turn("s1", "hand A") + o2, _ = chat._claim_turn("s1", "hand B") + o3, _ = chat._claim_turn("s2", "hand A") # different session + assert o1 and o2 and o3 + + +def test_await_returns_owner_reply(): + _, rec = chat._claim_turn("s1", "msg") + chat._finish_turn(rec, "the answer") + assert chat._await_duplicate(rec) == "the answer" + + +def test_respond_duplicate_reuses_result_without_running_turn(monkeypatch): + # owner already ran and cached its reply + _, rec = chat._claim_turn("s1", "same hand") + chat._finish_turn(rec, "owner reply") + + def boom(*a, **k): + raise AssertionError("duplicate must NOT execute the turn") + monkeypatch.setattr(chat.mind, "assemble", boom) + + out = chat.respond("s1", "same hand", "cloud") + assert out == "owner reply" + + +def test_respond_stream_duplicate_yields_cached_reply(monkeypatch): + _, rec = chat._claim_turn("s1", "same hand") + chat._finish_turn(rec, "owner reply") + + def boom(*a, **k): + raise AssertionError("duplicate must NOT execute the turn") + monkeypatch.setattr(chat.mind, "assemble", boom) + + events = list(chat.respond_stream("s1", "same hand", "cloud")) + assert ("delta", "owner reply") in events + assert ("done", "owner reply") in events + + +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 + o2, _ = chat._claim_turn("s1", "later resend") + assert o1 and o2 # a genuine later resend runs fresh