fix(core): keep abnormally ended text streams from rethrowing globally

[anzemur]

What

When a participant disconnects while one of its text streams is still open, livekit-client errors the open stream's controller with a DataStreamError (AbnormalEnd) — by design, in validateParticipantHasNoActiveDataStreams.

setupTextStream reads each incoming stream with from(reader).pipe(scan(...)).subscribe(next) — no error handler. RxJS then reports the unhandled observable error by rethrowing it globally from a timer (reportUnhandledError → setTimeout(() => { throw err })), so every abnormally ended transcription stream surfaces as an uncaught DataStreamError: Participant agent-… unexpectedly disconnected in the middle of sending data that no application code can catch.

For voice-agent apps this fires whenever the agent leaves the room mid-segment (interrupted farewell, agent process exit, crash). In our production app it is the #1 error by users impacted (~940 users / week) while being functionally benign — the transcript had already been delivered chunk by chunk and the session is over.

The chat handler's streamObservable.subscribe({ next }) has the same gap.

Fix

Give both subscribes an error callback: keep the already-accumulated text (every chunk was emitted through next; on abnormal end there is nothing left to deliver) and log the error at debug level. For chat, finalize already cleans up attachment state on error.

Tests

New test drives setupTextStream with a reader that yields chunks and then throws the abnormal-end error: the accumulated text survives, the error is logged, and nothing rethrows globally (without the fix, the test run fails with the uncaught error). Full @livekit/components-core suite passes (99 tests).

Live verification

Reproduced end to end against a real livekit-server (Docker, --dev) with real Chromium on both sides (Playwright): a sender participant streams four chunks on lk.transcription without closing the writer, then goes away; the receiver page runs setupTextStream and records all uncaught errors plus emissions. 8/8 deterministic runs across 2 iterations x {unfixed 0.12.15, this branch} x {tab hard-killed, clean room.disconnect() with the stream open}:

• Unfixed: exactly one uncaught page-level DataStreamError: Participant … unexpectedly disconnected in the middle of sending data (surfaced via window.onerror / Playwright pageerror — no application code can catch it).
• Fixed: zero uncaught errors, and the accumulated text still ends complete ("Hello from the agent"), so nothing is lost by handling the error.

The only delta between the two receiver bundles is @livekit/components-core; every other dependency is alias-pinned identically. The live test drives textStream.ts; the identical handler added in chat.ts shares the same mechanism but was not separately driven.

Update: attachment (byte) streams

Review feedback pointed out the fix did not cover attachment streams: a disconnect mid-attachment left the for await rejection unhandled, the attachment's Future pending forever (it was typed Future<…, never> and could not reject), the message pipeline hanging, and the streamIdToAttachments entry leaked.

Addressed by settling the future at the source: the byte stream handler now catches the failed read and rejects the attachment's Future (error type widened to Error). The rejection propagates through the existing message pipeline, so the subscribe error callback records it and finalize cleans up the attachment state — no hang, no leak, no unhandled rejection, and the pipeline stays healthy for subsequent messages. The error callback is also reason-aware now: DataStreamErrorReason.AbnormalEnd logs at debug (expected disconnect churn) while any other failure logs a visible warning.

New tests cover both: an attachment ending abnormally (no unhandled rejection, message dropped, a later message with a completing attachment still delivers) and a non-disconnect attachment failure hitting the warn path.

Update 2: early rejections and attachments that never start

Two more gaps from review, both fixed:

• Rejection before anyone listens: concatMap subscribes to the attachment futures one at a time, so a future rejecting before its turn (a later attachment failing first, or any attachment failing before the first text chunk) surfaced as an unhandled rejection. Each future now pre-attaches a no-op catch at creation (as suggested), keeping the rejection handled while concatMap's later subscription still observes it.
• Attachment byte stream never starts: if the sender drops before opening the byte stream, no controller exists for livekit-client to error, so nothing settled the future and the pipeline hung. The message pipeline now ends with takeUntil on the sender's RoomEvent.ParticipantDisconnected in that room, and on the room's own disconnect, since a local disconnect clears stream controllers without erroring them. finalize then removes the attachment entry.

On the behavior question: the message is dropped, terminally rather than delivered text-only - consistent with a mid-transfer attachment failure, so partial failure behaves uniformly (error callback logs, finalize cleans up, later messages unaffected). Happy to switch to text-only delivery if that's preferred.

Tests added for both: a later attachment rejecting while the first is still in flight (no unhandled rejection), and a never-starting attachment settled by the disconnect (terminal drop, pipeline healthy for the next message).

Update 3: takeUntil instead of a local error

Following review, the never-started attachment case no longer rejects with an error built in components-js. The pipeline completes via takeUntil(senderLeft$), placed before finalize, and via takeUntil on the room's disconnect. The attachment map no longer stores the Room. New tests: a message still waiting on an attachment ends on a local room disconnect, and chat text already delivered is kept when the sender leaves mid-stream.

[changeset-bot]

🦋 Changeset detected

Latest commit: 9219315

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 7 packages

Name	Type
@livekit/components-core	Patch
@livekit/components-react	Patch
@livekit/agents-ui	Patch
@livekit/component-example-next	Patch
@livekit/components-js-docs	Patch
@livekit/component-docs-storybook	Patch
@livekit/components-docs-gen	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[vercel]

@anzemur is attempting to deploy a commit to the LiveKit Team on Vercel.

A member of the Team first needs to authorize it.

[1egoman]

@anzemur Thanks for the contribution! There's a few issues though which I think would need to be addressed prior to being merged:

1. The change does not correct the problem for data stream attachments.

For context, the way data streams works, the sender transmits each attachment as a separate data stream forming a "tree". The attachedStreamIds field links each attachment stream to its parent text stream, and setupChat makes one Future for each attachment. The chat message pipeline waits for all of these Future objects before resolving.

If the sender disconnects during an attachment stream, the for await loop in the byte stream handler throws an error. The byte stream handler is an async function, and livekit-client does not catch the rejection. Because of this, the error becomes an unhandled rejection, and the new error callback does not receive this error, because the error is not in the text stream.

This means the attachment Future does not resolve or reject, it stays pending indefinitely. This means the pipeline does not complete, finalize does not run, and the key stays in streamIdToAttachments permanently. The user also does not receive the chat message.

So, I don't think a top-level catch is sufficient. Since no error bubbles up to the top level, you first need to settle the attachment Future, then, the top-level handler can record the error in one place.

2. Errors in the text stream are safe, but errors in an attachment stream are not

If the text stream fails during a read, finalize runs for all error types and deletes all state in any persistent maps. This is correct today.

However, if an attachment stream fails during a read, the state stays. This is true for all error types, not only for a disconnection. So I think you would need to make this case disconenction error specific, which you can do by reading the DataStreamErrorReason enum on the error.

General thoughts on these issues? I think as part of this change it might be worthwhile to introduce at least one more (if not more) test(s) with a data stream that has an attachment and validate that errors propegate properly in the above situations.

[1egoman]

Thanks for the continued progress on this! I think you are on the right track, but there's a few small bugs with with what you have now which I think would need to be addressed prior to being merged.

[1egoman]

This new error type would I think be the first instance of a components-js introduced error type which livekit-client doesn't expose. I'm not 100% opposed to it but I think my strong preference would be to fix this issue at the livekit-client level and then let components-js inherit that fix.

Alternatively, if there isn't a way to nicely encapsulate this into livekit-client, maybe something like the below could be an option:

• Build a per-sender "participant left" observable from RoomEvent.ParticipantDisconnected, filtered to this room and participantInfo.identity.
• Add takeUntil(senderLeft$) to streamObservable, placed before finalize.

The pipeline would then complete instead of erroring, finalize still cleans up, and the message is dropped just as it is now. If the text stream was still mid-flight it errors on its own anyway, so takeUntil changes nothing there. This would remove the class, the reject loop and the instanceof AttachmentNotReceivedError branch.

[anzemur]

Went with your takeUntil alternative in 9219315. The text stream handler builds senderLeft$ from RoomEvent.ParticipantDisconnected on this room, filtered to participantInfo.identity, and the pipeline has takeUntil(senderLeft$) right before finalize. The class, the reject loop and the instanceof branch are removed; the message is still dropped, and finalize cleans up.

One nuance on the mid-flight case: livekit-client errors the open controller and emits ParticipantDisconnected in the same tick, so takeUntil completes the pipeline before the read error reaches it, and RxJS drops that late error on the closed subscriber without rethrowing. The result is the same, the text already emitted stays and nothing is uncaught; only the debug log is skipped. Covered by "keeps the text already delivered when the sender leaves mid-stream", which mirrors that ordering.

If you'd still prefer the never-opened case to live in livekit-client, I'm happy to open that as a follow-up there and drop senderLeft$ here once it lands.

[1egoman]

I think I'd really prefer this be fixed upstream in livekit-client. The additional suggestion I provided was more in case there plain isn't a way to upstream this, but I would like to avoid a situation where a user runs into a variation of the same issue on livekit-client and we have to figure out how to reconcile the state between the two layers and "back out" hacks at this level.