fix(chat): de-duplicate the double turn execution at its source

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) <noreply@anthropic.com>

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)
+    # A concurrent duplicate (the UI's stream + blocking fallback for one message)
-    messages = turn.messages
+    # reuses the owner's result instead of running a second full turn.
-    tool_specs = toolkit.specs(turn.mode.tools) if backend in cfg.tool_backends else None
+    is_owner, rec = _claim_turn(session_id, user_msg)
-    ctx = {"session_id": session_id, "backend": backend}
+    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
+    reply = _TANGLED
-    # tool events fired mid-turn (keeps the transcript export in true order).
+    try:
-    memory.remember(session_id, "user", user_msg)
+        turn = mind.assemble(session_id, user_msg, backend, model)
-    reply, tools_run = _mind_loop(messages, backend, model, tool_specs, ctx, session_id)
+        messages = turn.messages
-    _ensure_hand_logged(messages, user_msg, turn.msg_type, tools_run, backend, model, ctx, session_id)
+        tool_specs = toolkit.specs(turn.mode.tools) if backend in cfg.tool_backends else None
-    mouth = _mouth_target(cfg, backend, model)
+        ctx = {"session_id": session_id, "backend": backend}
-    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)
+        # Persist the user turn before the tool loop so its timestamp precedes any
-    summary.maybe_summarize_async(session_id)  # compact once enough new turns pile up
+        # tool events fired mid-turn (keeps the transcript export in true order).
-    return reply
+        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)
+    # A concurrent duplicate (this stream + the UI's blocking fallback for one
-    messages = turn.messages
+    # message) reuses the owner's result instead of running a second full turn.
-    tool_specs = toolkit.specs(turn.mode.tools) if backend in cfg.tool_backends else None
+    is_owner, rec = _claim_turn(session_id, user_msg)
-    ctx = {"session_id": session_id, "backend": backend}
+    if not is_owner:
-    mouth = _mouth_target(cfg, backend, model)
+        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
+    reply = _TANGLED
-    # mid-turn, chronologically after the user message in the exported transcript.
+    try:
-    memory.remember(session_id, "user", user_msg)
+        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:
+        # Persist the user turn up front (see respond): keeps tool events, which fire
-        # No separate voice: stream the mind directly (the original path, unchanged).
+        # mid-turn, chronologically after the user message in the exported transcript.
-        parts: list[str] = []
+        memory.remember(session_id, "user", user_msg)
-        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))
+        if mouth is None:
-    memory.remember(session_id, "assistant", reply)
+            # No separate voice: stream the mind directly (the original path, unchanged).
-    summary.maybe_summarize_async(session_id)
+            parts: list[str] = []
-    yield ("done", reply)
+            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)

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,

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