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 efficiency by reducing unnecessary state updates and optimizing stream state management.

_{✏️ 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

This change optimizes cursor state updates in useDeltaStreams.ts by adding change detection logic that defers cursor state updates only when actual changes occur, and restructures delta processing to handle cases where deltas may be absent while preventing unnecessary stream removal-related churn.

Changes

Cohort / File(s)	Summary
Cursor state optimization src/react/useDeltaStreams.ts	Adds cursorsChanged flag and change-detection logic to guard setCursors invocation; detects stream removal by comparing existing cursors against computed newCursors; reduces unnecessary state updates and identity churn when streams are removed

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

• Cursor change detection logic: Verify that the comparison logic correctly identifies when cursors have actually changed and when streams have been removed
• Guard condition modification: Ensure the restructured guard (from newDeltas.length to only requiring streamMessages) maintains correct behavior and doesn't skip necessary processing
• State update deference: Confirm that deferring setCursors behind the change check reduces duplicates without inadvertently preventing needed updates in edge cases (e.g., stream completion scenarios)
• Integration with linked issue #185: Validate that this change addresses the duplicate message problem when saveStreamDeltas: true is used after tool approval

Poem

🐰 A rabbit hops through streams so fine,
Cursors changed? A clever sign!
No duplicate deltas shall we see,
Just flowing updates, pure and free! ✨

Pre-merge checks and finishing touches

❌ Failed checks (1 warning, 1 inconclusive)

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.
Linked Issues check	❓ Inconclusive	The PR addresses #185 by cleaning up stale stream deltas and #187's root cause (the guard on newDeltas prevents unnecessary iterations); however, #187's specific JSON.stringify performance fix is not implemented in the code changes shown.	Verify whether #187's lazy error message optimization in deltas.ts is also included in this PR, or if it requires a separate fix for the performance hotspot identified.

✅ Passed checks (3 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: removing/stripping out old streams when they are no longer present, which directly addresses the core issue of duplicate streaming messages.
Out of Scope Changes check	✅ Passed	All changes focus on cursor management and stream cleanup in useDeltaStreams.ts, which are directly related to removing stale streams and preventing duplicate messages in the linked issues.

✨ 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

---

📜 Recent review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

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

📒 Files selected for processing (1)

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

🚧 Files skipped from review as they are similar to previous changes (1)

• src/react/useDeltaStreams.ts

---

_{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/workflow@0.3.2, 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/workflow@0.3.2 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.