serversdown
f3ecf8ffe4
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>
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""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
|