fix(stream): gate resume on REALTIME mode, chat ownership check, strict request_id match

- resume_stream now short-circuits when ENABLE_REALTIME_CHAT_SAVE is
  on, mirroring the write-side gate in _stream_seq_allocate. Without
  this, stale logs written before the flag was flipped (or by a
  mixed-version peer) could still replay on top of DB-backed content
  and double-apply.

- Added optional chat-ownership check: if the client sends chat_id,
  we validate via Chats.get_chat_by_id_and_user_id and reject on
  failure. Defense-in-depth alongside the user-scoped key. Fails
  OPEN when the chat lookup errors or chat_id is absent, so we
  don't regress the "stub not persisted in DB" refresh scenario
  that motivated removing the earlier stricter check. Frontend
  updated to include chat_id in the resume payload.

- Client now requires an exact request_id match when expected is
  set. A reply without a request_id used to fall through and clear
  the active fence, which could prematurely flush a newer in-flight
  request's buffer. Now rejects both missing and mismatched.

Deferred: suggestion to prune resumeSeqByMessageId on per-message
terminal events. That was intentionally removed two rounds ago
because the same pattern caused continuation-reuses-message_id
to replay duplicate content. The memory footprint is int-per-
message bounded by chat size and is cleared at chat/navigation
boundaries — this is a deliberate trade-off, not an oversight.
This commit is contained in:
Claude 2026-04-15 07:01:02 +00:00
parent fb212156f3
commit 4cc28eabff
No known key found for this signature in database
2 changed files with 27 additions and 5 deletions

View file

@ -719,12 +719,31 @@ async def resume_stream(sid, data):
envelopes = []
user = SESSION_POOL.get(sid)
user_id = user.get('id') if user else None
if user_id and REDIS is not None:
# Gate read on REALTIME mode to match the write-side gate. Otherwise
# stale logs written before the flag was flipped (or by a mixed-
# version peer) would still get replayed on top of DB-backed content
# and double-apply.
if user_id and REDIS is not None and not ENABLE_REALTIME_CHAT_SAVE:
try:
last_seq = int(data.get('last_seq') or 0)
except (TypeError, ValueError):
last_seq = 0
envelopes = await _stream_log_read(user_id, message_id, last_seq)
# Defense-in-depth: validate chat ownership even though the log
# key is already user-scoped. Rejects only when we can *prove*
# ownership fails — a missing/unknown chat_id falls through to
# the user-scoped key guarantee so we don't regress the "stub
# not yet persisted in DB" refresh case.
chat_id = data.get('chat_id')
chat_ok = True
if chat_id:
try:
chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id)
chat_ok = chat is not None
except Exception as e:
log.warning(f'resume-stream chat ownership check failed: {e}')
chat_ok = True # fail open to not regress legitimate callers
if chat_ok:
envelopes = await _stream_log_read(user_id, message_id, last_seq)
await sio.emit(
'resume-stream:replay',

View file

@ -699,6 +699,7 @@
: `${message.id}-${Date.now()}-${Math.random()}`;
resumeActiveRequestIdByMessageId.set(message.id, requestId);
$socket.emit('resume-stream', {
chat_id: $chatId,
message_id: message.id,
request_id: requestId,
last_seq: resumeSeqByMessageId.get(message.id) ?? 0
@ -708,10 +709,12 @@
const onResumeStreamReplay = async (payload) => {
const messageId = payload?.message_id;
if (!messageId) return;
// Drop stale replies from superseded requests so they can't
// clear a fence that belongs to a newer in-flight request.
// Drop stale replies. If an active request_id is set for this
// message, require an exact match — a reply without a request_id,
// or with a mismatched one, belongs to a superseded request and
// must not clear the active fence.
const expected = resumeActiveRequestIdByMessageId.get(messageId);
if (expected && payload?.request_id && payload.request_id !== expected) {
if (expected && payload?.request_id !== expected) {
return;
}
resumeActiveRequestIdByMessageId.delete(messageId);