serversdown/project-lyra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(chat): client turn-id makes fire-and-forget bulletproof

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
serversdown
2026-07-12 01:28:10 +00:00
parent 80519d20b1
commit 7910d266db
4 changed files with 76 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lyra/chat.py
+37 -24
View File
@@ -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)