fix(stream): fence lifecycle ownership, strict done filter, narrower terminal

- onResumeStreamReplay no longer blindly clears the fence in finally.
  It stops deleting the active request_id on entry and instead checks,
  in the finally, whether it is still the active request before
  calling clearResumeFence. A newer request arriving during replay
  sets its own id; the older handler now leaves that state alone and
  lets the newer request's lifecycle drive cleanup.

- requestResumeForAllInProgress filters by `done === false` instead of
  `!done`. Legacy messages with a missing `done` field used to fan
  out as spurious resume requests on every reconnect/load; now only
  explicitly in-progress assistants get targeted.

- Terminal-event detection for TTL shortening no longer treats any
  event with `data.error` present as terminal. Restricted to
  `type === 'chat:completion' AND data.error`, matching the exact
  shape middleware.py emits when the provider actually errors out.
  A transient warning on some other event type won't shorten the log
  TTL prematurely anymore.
This commit is contained in:
Claude 2026-04-15 07:46:21 +00:00
parent 630f452e93
commit 418ef25dc4
No known key found for this signature in database
2 changed files with 20 additions and 5 deletions

View file

@ -1112,10 +1112,17 @@ async def get_event_emitter(request_info, update_db=True):
# will no longer replay against.
outer_type = event_data.get('type') if isinstance(event_data, dict) else None
inner = event_data.get('data') if isinstance(event_data, dict) else None
# Only the specific chat:completion-with-error shape is terminal;
# a generic `error` field on some other event type could be a
# transient warning and shouldn't age the keys out early.
is_terminal = (
(isinstance(inner, dict) and inner.get('done') is True)
or outer_type == 'chat:tasks:cancel'
or (isinstance(inner, dict) and inner.get('error'))
or (
outer_type == 'chat:completion'
and isinstance(inner, dict)
and inner.get('error')
)
)
if is_terminal and REDIS is not None and user_id and message_id:
try:

View file

@ -724,7 +724,11 @@
if (expected && payload?.request_id !== expected) {
return;
}
resumeActiveRequestIdByMessageId.delete(messageId);
// Don't delete the active request_id here: a NEWER request could
// start during replay and set its own id. The finally below only
// clears the fence if we're still the active request; if a newer
// request has taken over, let its own lifecycle handle cleanup.
const ownRequestId = payload?.request_id;
const envelopes = Array.isArray(payload?.envelopes) ? payload.envelopes : [];
try {
for (const envelope of envelopes) {
@ -733,8 +737,10 @@
await chatEventHandler(envelope);
}
} finally {
// Finally-clear so a mid-replay throw can't freeze live updates.
await clearResumeFence(messageId);
const current = resumeActiveRequestIdByMessageId.get(messageId);
if (current === ownRequestId || current == null) {
await clearResumeFence(messageId);
}
}
};
@ -745,10 +751,12 @@
};
// Iterate all in-flight assistants so arena siblings aren't missed.
// Strict `done === false` so legacy messages with a missing `done`
// field don't get fanned out as unnecessary resume requests.
const requestResumeForAllInProgress = () => {
if (!history?.messages) return;
for (const message of Object.values(history.messages)) {
if (message && message.role === 'assistant' && !message.done) {
if (message && message.role === 'assistant' && message.done === false) {
requestResumeForMessage(message);
}
}