strip out old streams when they are no longer present

[ianmacartney]

Cleans up streaming messages from the list when no new deltas arrive but a stream is no longer present.

Fixes #185
Fixes #187

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

Summary by CodeRabbit

• Refactor
  • Improved stream message processing to avoid work when no new updates are present, reduce redundant state changes, and minimize churn when streams are removed—resulting in more efficient and stable stream handling.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[pkg-pr-new]

Open in StackBlitz

npm i https://pkg.pr.new/get-convex/agent/@convex-dev/agent@186

commit: 00d3861

[coderabbitai]

Walkthrough

Reworks delta stream handling in useDeltaStreams.ts: moves per-delta processing into a guarded block when deltas exist, computes newCursors and a cursorsChanged flag, and only calls setCursors when cursors actually differ (including detection of removed streams), reducing unnecessary cursor updates and identity churn.

Changes

Cohort / File(s)	Summary
Delta stream & cursor handling src/react/useDeltaStreams.ts	Restructures guard logic to proceed on streamMessages rather than requiring newDeltas.length; encloses per-delta processing in an inner length check; computes newCursors and cursorsChanged, detects removed streams by comparing existing cursors to newCursors, and invokes setCursors only when an actual change is detected.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

• Verify correctness of the cursors equality/comparison logic and removal detection.
• Confirm the new guard preserves previous behavior when newDeltas are present and avoids unnecessary processing when absent.
• Check interactions with stream completion and that deferred setCursors does not leave stale "streaming" entries (relevant to linked issue #185).
• Review potential edge cases where cursors should change but comparison misses subtle mutations (deep vs shallow equality).

Poem

🐇 I hop through deltas, keen and spry,

New cursors counted with careful eye.
Only when changed will I rearrange,
No extra churn, just tidy change.
✨

Pre-merge checks and finishing touches

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 14.29% which is insufficient. The required threshold is 80.00%.	You can run @coderabbitai generate docstrings to improve docstring coverage.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately describes the main change: guarding cursor updates and removing old streams when they are no longer present.
Linked Issues check	✅ Passed	The PR addresses #185 by detecting and preventing stale stream deltas (via cursorsChanged flag and newCursors comparison) and #187 indirectly by optimizing state update logic.
Out of Scope Changes check	✅ Passed	All changes are scoped to useDeltaStreams.ts and directly address the linked issues without introducing unrelated modifications.

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch ian/debug-uimessages

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[ianmacartney]

@coderabbitai review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Actionable comments posted: 0

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

example/convex/workflows/chaining.ts (1)

58-82: Based on my verification, I've found critical issues with the review comment. The web search and codebase analysis reveal:

1. API change direction is backwards: Version 0.3.2 changed the parameter FROM ctx TO step (not the reverse as stated)
2. Codebase incompatibility: The entire codebase uniformly uses the OLD ctx API, but the package.json specifies @convex-dev/[email protected], which requires the NEW step API
3. No handlers use the correct API: I found zero occurrences of handlers using the step parameter

The review comment is factually incorrect and the actual issue is more severe. Here is the rewritten review comment:

---

Update workflow handlers to use step parameter for @convex-dev/[email protected] compatibility.

Version 0.3.2 of @convex-dev/workflow introduces a breaking API change requiring handlers to use a step parameter instead of ctx. The code currently uses the pre-0.3.2 API (e.g., ctx.runAction() at line 67-70), which is incompatible with the installed version. All handlers must be migrated: change async (ctx, { ... }) to async (step, { ... }) and replace ctx.runAction() calls with step.runAction().

🧹 Nitpick comments (3)

src/client/streamText.ts (3)

77-77: Unnecessary nullish coalescing on required parameter.

The options parameter is required in the function signature, so the ?? {} fallback is never reached.

-  const { threadId } = options ?? {};
+  const { threadId } = options;

---

100-100: Inconsistent optional chaining on required parameter.

options is a required parameter, so optional chaining is unnecessary here.

-            agentName: options?.agentName,
+            agentName: options.agentName,

---

129-133: Address or document the commented-out streamer metadata update.

This commented-out code suggests an incomplete feature for updating streamer metadata when the model changes during multi-step execution. Consider either implementing this or adding a TODO comment explaining why it's deferred.

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between b48f51c and 50d5d69.

⛔ Files ignored due to path filters (1)

• package-lock.json is excluded by !**/package-lock.json

📒 Files selected for processing (10)

• CHANGELOG.md (1 hunks)
• example/convex/workflows/chaining.ts (1 hunks)
• package.json (2 hunks)
• src/client/index.ts (4 hunks)
• src/client/saveInputMessages.ts (1 hunks)
• src/client/start.ts (3 hunks)
• src/client/streamText.ts (1 hunks)
• src/client/utils.ts (1 hunks)
• src/deltas.ts (1 hunks)
• src/react/useDeltaStreams.ts (1 hunks)

🧰 Additional context used

🧬 Code graph analysis (4)

example/convex/workflows/chaining.ts (1)

src/client/index.ts (2)

• saveMessage (139-139)
• saveMessage (694-723)

src/client/streamText.ts (5)

src/client/types.ts (3)

• AgentPrompt (41-81)
• Options (588-609)
• GenerationOutputMetadata (232-253)

src/client/streaming.ts (2)

• StreamingOptions (134-156)
• mergeTransforms (171-192)

src/client/start.ts (1)

• startGeneration (30-310)

src/shared.ts (2)

• getModelName (94-104)
• getProviderName (106-111)

src/client/utils.ts (1)

• errorToString (22-27)

src/client/saveInputMessages.ts (1)

playground/bin/agent-playground.cjs (1)

• args (15-15)

src/client/index.ts (1)

src/client/streamText.ts (1)

• streamText (33-163)

🔇 Additional comments (15)

package.json (1)

10-10: LGTM! Version bumps are consistent.

The package version bump to 0.3.1 aligns with the CHANGELOG entries, and the workflow dependency update to 0.3.2 matches the API changes in the workflow example file.

Also applies to: 90-90

src/deltas.ts (1)

80-83: Excellent fix for the performance issue!

This change directly addresses issue #187 by removing the eager JSON.stringify from the assertion message. The simplified message still conveys the intent without the expensive serialization that was causing UI slowdowns during streaming.

src/client/saveInputMessages.ts (1)

82-93: LGTM! Correctly propagates promptMessageId.

This change ensures that promptMessageId is preserved when saving messages, which aligns with the CHANGELOG entry about preventing messages from being saved as newest when promptMessageId wasn't latest.

CHANGELOG.md (1)

3-9: LGTM! Changelog accurately describes the changes.

The 0.3.1 entry clearly documents both the performance fix (avoiding JSON.stringify in render path) and the promptMessageId preservation enhancement.

src/react/useDeltaStreams.ts (1)

106-150: Excellent fix for issue #185!

This refactor properly handles the cleanup of streaming messages when streams are no longer present, which directly addresses the duplicate message problem described in issue #185. The key improvements are:

1. Stream removal detection (lines 142-147): Detects when streams disappear from the active set and marks state as changed, ensuring stale stream deltas are removed from the UI.
2. Conditional state updates (lines 148-150): Only calls setCursors when actual changes occur, reducing unnecessary re-renders.
3. Proper guard ordering (line 106, 108): Processes delta updates when streamMessages exist, with nested guard for actual deltas.

The logic correctly handles the lifecycle: when a stream completes and the backend no longer returns it in the stream list, this code will detect the removal and trigger a state update to clean it up.

src/client/start.ts (1)

93-93: LGTM! Good use of optional parameter with runtime validation.

Making languageModel optional in the API while asserting at runtime (line 158) provides flexibility while maintaining safety. The model is resolved from either args.model or opts.languageModel, with a clear error if neither is provided.

Also applies to: 157-158

src/client/utils.ts (2)

22-27: LGTM! Simple and robust error conversion.

The errorToString utility correctly handles both Error instances and other types, providing a reliable way to convert any thrown value to a string.

---

3-20: I'll help you verify the concerns about the willContinue function's continuation logic. Let me start by searching for usage patterns and understanding the intended behavior.
<function_calls>


#!/bin/bash

Search for usage patterns of willContinue to understand expected behavior

rg -n -C3 'willContinue\s*(' --type=ts


</function_calls>

Now let me also search for any test files or documentation related to this function:
<function_calls>


#!/bin/bash

Search for test files that might test willContinue

fd -e test.ts -e spec.ts -e test.js | head -20

Also search for any documentation or comments about stopWhen

rg -n 'stopWhen' --type=ts -C2 | head -50


</function_calls>

<function_calls>


#!/bin/bash

Check the type definitions for StopCondition to understand the contract

rg -n 'StopCondition' --type=ts -B2 -A5


</function_calls>

<function_calls>


#!/bin/bash

Look at the full context of the utils.ts file to understand the function better

wc -l src/client/utils.ts


</function_calls>

<function_calls>


#!/bin/bash

Read the full utils.ts file to get complete context

cat -n src/client/utils.ts


</function_calls>

src/client/streamText.ts (3)

1-24: LGTM!

Imports are well-organized, pulling types and utilities from the appropriate modules. The delegation to startGeneration, utility functions (errorToString, willContinue), and streaming constructs aligns with the modular design described in the PR.

---

26-76: LGTM!

The function signature and documentation are clear. The type parameters and option types properly support the streaming use case with optional delta persistence.

---

145-162: LGTM!

The stream consumption logic correctly handles the returnImmediately option and builds the appropriate metadata. The conditional await pattern ensures the stream is fully consumed when required.

src/client/index.ts (4)

19-19: LGTM!

The streamText import from "ai" is correctly removed as the local wrapper is now used instead.

---

66-66: LGTM!

Import correctly updated to pull only what's needed from the streaming module.

---

90-91: LGTM!

The imports for the extracted streamText function and utilities (errorToString, willContinue) correctly support the refactored code structure.

---

543-562: LGTM!

The delegation to the extracted streamText function is clean. The argument merging correctly prioritizes explicit streamTextArgs values over agent defaults, and the options spread order is appropriate.

[coderabbitai]

Actionable comments posted: 0

🧹 Nitpick comments (1)

src/react/useDeltaStreams.ts (1)

115-118: Consider lazy error message construction for hot-path assertions.

While issue #187 specifically mentions updateFromUIMessageChunks in deltas.ts, these assertions also construct error messages eagerly (using template literals) on every delta, even when the assertion passes. During active streaming, this could contribute to unnecessary string allocations.

🔎 Optional refactor to lazy error messages

-        assert(
-          previousEnd === delta.start,
-          `Gap found in deltas for ${delta.streamId} jumping to ${delta.start} from ${previousEnd}`,
-        );
+        assert(
+          previousEnd === delta.start,
+          () => `Gap found in deltas for ${delta.streamId} jumping to ${delta.start} from ${previousEnd}`,
+        );

-        assert(
-          !oldCursor || oldCursor === delta.start,
-          `Gap found - first delta after ${oldCursor} is ${delta.start} for stream ${delta.streamId}`,
-        );
+        assert(
+          !oldCursor || oldCursor === delta.start,
+          () => `Gap found - first delta after ${oldCursor} is ${delta.start} for stream ${delta.streamId}`,
+        );

This requires that the assert function from convex-helpers supports lazy message evaluation (callback form). If not, this optimization may not be applicable.

Also applies to: 121-124

📜 Review details

Configuration used: defaults

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 00d3861 and f9dac1e.

📒 Files selected for processing (1)

• src/react/useDeltaStreams.ts (1 hunks)

🔇 Additional comments (2)

src/react/useDeltaStreams.ts (2)

106-128: Good fix for detecting removed streams without new deltas.

The broadened guard at line 106 (changed from newDeltas?.length && streamMessages to just streamMessages) is the key change that allows removal detection even when no new deltas arrive. This directly addresses issue #185 where streams can get stuck in a "streaming" state after completion.

The inner guard at line 108 correctly preserves the delta processing logic only when deltas are present.

Please verify that this change successfully resolves the duplicate message issue described in #185 by testing the tool approval flow with saveStreamDeltas: true.

---

129-150: Excellent removal detection and optimization.

The addition of the cursorsChanged flag and the removal detection loop (lines 141-147) are well-implemented:

1. Removal detection: Iterating over existing cursors to find streams no longer present in newCursors correctly identifies completed/removed streams.
2. State update optimization: Only calling setCursors when cursors actually changed reduces unnecessary re-renders.

This combination ensures that when streams complete and are no longer in streamMessages, their cursors are properly cleaned up, preventing the stuck streaming state described in issue #185.