fix(chat): stop losing a user message that arrived mid-turn - #4795
fix(chat): stop losing a user message that arrived mid-turn#4795ericallam wants to merge 3 commits into
Conversation
🦋 Changeset detectedLatest commit: 3dd60c2 The changes in this PR will be included in the next version bump. This PR includes changesets to release 27 packages
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 |
WalkthroughThe change introduces durable session-stream records with stable IDs and sequence numbers. A route-based dispatcher separates messages, stops, and handovers, while replay checkpoints preserve unconsumed messages across suspension and recovery. Chat agents now expose mailbox inspection and record-consumption APIs. Steering messages remain queued when injection is declined. Tests cover routing, cursor handling, replay, recovery, waitpoint races, and warm-process reuse. Documentation and changesets describe the updated behavior. Merge Risk: 🟠 High · up to The PR changes chat message queueing and resume behavior, but the current head still permits queued user messages to be cleared before resumption, which can permanently skip them; a recovery path is also missing a required method and a retry edge case can redeliver a control event. Merge should wait for these correctness issues to be fixed. 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Full details: Description checkExplanation The description is detailed and directly related to the changes. It explains the root causes, design, testing, limitations, and affected behavior. It does not use every template heading or include a Closes issue reference, checklist state, changelog heading, or screenshots, but the core required information is present. Full details: Docstring CoverageExplanation Docstring coverage is 47.50% which is insufficient. The required threshold is 80.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 40 functions across 24 files. (7 skipped: 7 unsupported.)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/core/src/v3/test/test-session-stream-manager.ts (1)
242-257: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftClamp the test cursor behind buffered records.
If record 50 remains buffered and a handler consumes record 51,
#advanceLastDispatched()stores 51 and this method returns 51. A resumed test flow then skips record 50. Track unconsumed sequence numbers and apply the same clamp asStandardSessionStreamManager.lastDispatchedSeqNum().
🧹 Nitpick comments (2)
packages/trigger-sdk/src/v3/ai.ts (1)
215-227: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider collapsing the two
.outcursor scans into one.
findLatestSessionInCursorandfindLatestSessionInCheckpointscan the same.outrecords for the sameturn-completecontrol record.findLatestSessionInCheckpointreads one extra header. On a resuming boot where the snapshot carries nolastInEventId, the boot phase callsfindLatestSessionInCursor(Line 5935) andinstallChatInputRouterthen callsfindLatestSessionInCheckpoint, so the run performs two fullreadSessionStreamRecords(chatId, "out")round trips.
installChatInputRouteralready treatsfallbackResumeFromas a fallback for the boundary value. You can restrict the boot phase to the snapshot field and deletefindLatestSessionInCursor, or have the boot phase reuse the checkpoint scan result.Also applies to: 1904-1922
packages/trigger-sdk/test/mid-turn-resume-floor.test.ts (1)
76-87: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUpdate the docblock to describe the behavior the test now asserts.
The docblock states in the present tense that a mid-turn message "is delivered to the turn's push handler and parked in the in-memory wire buffer" and that "the turn boundary then publishes a cursor past a message that exists only in this process's memory". This PR removes that wire buffer and holds the resume floor behind the queued record, so the text describes the pre-fix behavior while the assertion below verifies the fix. Rewrite it as the invariant under test, and keep the pre-fix behavior in the past tense if you want the history.
Separately, the
SeqReadercast at Lines 104, 106 and 120 appears unnecessary.ai.tscallssessionStreams.lastSeqNum(payload.chatId, "in")directly without a cast.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 00bff6ce-95a6-4671-8531-67ff4acd9ec8
📒 Files selected for processing (31)
.changeset/quiet-floors-hold.md.changeset/spry-steers-defer.md.changeset/tidy-mailboxes-wait.mddocs/ai-chat/client-protocol.mdxdocs/ai-chat/custom-agents.mdxdocs/ai-chat/pending-messages.mdxdocs/ai-chat/reference.mdxpackages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/apiClient/runStream.tspackages/core/src/v3/session-streams-api.tspackages/core/src/v3/sessionStreams/index.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/manager.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/core/src/v3/test/mock-task-context.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/trigger-sdk/src/v3/ai.tspackages/trigger-sdk/src/v3/sessions.tspackages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/test/replay-session-in.test.ts
Included review availability: Your plan provides up to 10 included reviews per hour; 9 remain after this review.
📜 Review details
⏰ Context from checks skipped due to timeout. (28)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (20, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (5, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (19, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (21, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (11, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (2, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (6, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (8, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (18, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (7, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (13, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (10, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (15, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (9, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (16, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (17, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (14, 24)
- GitHub Check: webapp / 🧪 Unit Tests: Webapp (12, 24)
- GitHub Check: e2e / 🧪 CLI v3 tests (warp-windows-latest-x64-8x - npm)
- GitHub Check: sdk-compat / Deno Runtime
- GitHub Check: e2e / 🧪 CLI v3 tests (warp-windows-latest-x64-8x - pnpm)
- GitHub Check: packages / 🧪 Unit Tests: Packages (2, 3)
- GitHub Check: e2e-webapp / 🧪 E2E Tests: Webapp (2, 2)
- GitHub Check: packages / 🧪 Unit Tests: Packages (1, 3)
- GitHub Check: packages / 🧪 Unit Tests: Packages (3, 3)
- GitHub Check: e2e-webapp / 🧪 E2E Tests: Webapp (1, 2)
- GitHub Check: internal / 🧪 Unit Tests: Internal
- GitHub Check: Analyze (javascript-typescript)
🧰 Additional context used
📓 Path-based instructions (14)
Always import from `@trigger.dev/sdk`. Never use `@trigger.dev/sdk/v3` or deprecated `client.defineJob`.
📄 CodeRabbit inference engine (AGENTS.md)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/trigger-sdk/src/v3/ai.ts
We use vitest exclusively. **Never mock anything** - use testcontainers instead.
📄 CodeRabbit inference engine (AGENTS.md)
Files:
packages/core/src/v3/apiClient/runStream.test.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/router.test.ts
**Import subpaths only** (never root).
📄 CodeRabbit inference engine (AGENTS.md)
Files:
packages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/session-streams-api.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/test/mock-task-context.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/core/src/v3/apiClient/runStream.tspackages/core/src/v3/sessionStreams/index.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/core/src/v3/sessionStreams/manager.ts
**Prefer static imports over dynamic imports.** Only use dynamic `import()` when:
📄 CodeRabbit inference engine (AGENTS.md)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/session-streams-api.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/test/mock-task-context.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/core/src/v3/apiClient/runStream.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/core/src/v3/sessionStreams/index.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/trigger-sdk/src/v3/ai.tspackages/core/src/v3/sessionStreams/manager.ts
Add crumbs as you write code — not just when debugging. Mark lines with
📄 CodeRabbit inference engine (AGENTS.md)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/core/src/v3/apiClient/runStream.test.tsdocs/ai-chat/pending-messages.mdxpackages/core/src/v3/session-streams-api.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/core/src/v3/sessionStreams/noopManager.tsdocs/ai-chat/reference.mdxdocs/ai-chat/custom-agents.mdxpackages/core/src/v3/test/mock-task-context.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/core/src/v3/apiClient/runStream.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/core/src/v3/sessionStreams/index.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tsdocs/ai-chat/client-protocol.mdxpackages/trigger-sdk/src/v3/ai.tspackages/core/src/v3/sessionStreams/manager.ts
Always import from `@trigger.dev/sdk`. Never use `@trigger.dev/sdk/v3` (deprecated path alias)
📄 CodeRabbit inference engine (packages/trigger-sdk/CLAUDE.md)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/trigger-sdk/src/v3/ai.ts
Use zod for validation in packages/core and apps/webapp
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Files:
packages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/session-streams-api.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/test/mock-task-context.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/core/src/v3/apiClient/runStream.tspackages/core/src/v3/sessionStreams/index.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/core/src/v3/sessionStreams/manager.ts
In the Trigger.dev SDK (packages/trigger-sdk), prefer isomorphic code like fetch and ReadableStream instead of Node.js-specific code
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/trigger-sdk/src/v3/ai.ts
Never import the root package (`@trigger.dev/core`). Always use subpath imports such as `@trigger.dev/core/v3`, `@trigger.dev/core/v3/utils`, `@trigger.dev/core/logger`, or `@trigger.dev/core/schemas`
📄 CodeRabbit inference engine (packages/core/CLAUDE.md)
Files:
packages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/session-streams-api.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/test/mock-task-context.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/core/src/v3/apiClient/runStream.tspackages/core/src/v3/sessionStreams/index.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/core/src/v3/sessionStreams/manager.ts
Use vitest for all tests in the Trigger.dev repository
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Files:
packages/core/src/v3/apiClient/runStream.test.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/router.test.ts
Use function declarations instead of default exports
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/session-streams-api.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/test/mock-task-context.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/core/src/v3/apiClient/runStream.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/core/src/v3/sessionStreams/index.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/trigger-sdk/src/v3/ai.tspackages/core/src/v3/sessionStreams/manager.ts
MDX documentation pages must include frontmatter with title (required), description (required), and sidebarTitle (optional) in YAML format
📄 CodeRabbit inference engine (docs/CLAUDE.md)
Files:
docs/ai-chat/pending-messages.mdxdocs/ai-chat/reference.mdxdocs/ai-chat/custom-agents.mdxdocs/ai-chat/client-protocol.mdx
Use types over interfaces for TypeScript
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/session-streams-api.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/test/mock-task-context.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/core/src/v3/apiClient/runStream.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/core/src/v3/sessionStreams/index.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/trigger-sdk/src/v3/ai.tspackages/core/src/v3/sessionStreams/manager.ts
When creating or editing OTEL metrics (counters, histograms, gauges), ensure metric attributes have low cardinality by using only enums, booleans, bounded error codes, or bounded shard IDs
📄 CodeRabbit inference engine (.cursor/rules/otel-metrics.mdc)
Files:
packages/trigger-sdk/src/v3/test/mock-chat-agent.tspackages/core/src/v3/apiClient/runStream.test.tspackages/core/src/v3/session-streams-api.tspackages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/core/src/v3/sessionStreams/noopManager.tspackages/core/src/v3/test/mock-task-context.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/mockChatAgent.test.tspackages/core/src/v3/sessionStreams/wireProtocol.tspackages/trigger-sdk/test/replay-session-in.test.tspackages/core/src/v3/apiClient/runStream.tspackages/trigger-sdk/test/chat-messages-mailbox.test.tspackages/trigger-sdk/test/pending-message-drain.test.tspackages/trigger-sdk/src/v3/sessions.tspackages/core/src/v3/sessionStreams/index.tspackages/trigger-sdk/src/v3/test/test-session-handle.tspackages/core/src/v3/sessionStreams/manager.test.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/test/session-waitpoint-backend.tspackages/core/src/v3/test/test-session-stream-manager.tspackages/core/src/v3/sessionStreams/router.test.tspackages/core/src/v3/sessionStreams/router.tspackages/trigger-sdk/src/v3/ai.tspackages/core/src/v3/sessionStreams/manager.ts
🧠 Learnings (3)
📚 Learning: 2026-08-16T18:36:58.179Z
Learnt from: ericallam
Repo: triggerdotdev/trigger.dev PR: 4537
File: packages/trigger-sdk/test/normalizeKeyString.test.ts:1-2
Timestamp: 2026-08-16T18:36:58.179Z
Learning: For related SDK `chat.agent` tests in the Trigger.dev repository—including chat channels, handover, snapshot, and transport-event coverage—keep new test files under `packages/trigger-sdk/test/` rather than colocating them with the `packages/trigger-sdk/src/v3/` source files.
Applied to files:
packages/trigger-sdk/test/chat-warm-process-reuse.test.tspackages/trigger-sdk/test/mid-turn-resume-floor.test.tspackages/trigger-sdk/test/chat-messages-mailbox.test.ts
📚 Learning: 2026-04-30T20:30:29.458Z
Learnt from: ericallam
Repo: triggerdotdev/trigger.dev PR: 3226
File: docs/ai-chat/quick-start.mdx:13-13
Timestamp: 2026-04-30T20:30:29.458Z
Learning: In this repo’s documentation MDX files (`docs/**/*.mdx`), use `ts` and `tsx` (not `typescript`) as the code-fence language tags for TypeScript/TSX snippets. Do not flag `ts`/`tsx` code-fence language tags as incorrect in any docs MDX file, since this is the site-wide Mintlify-compatible convention.
Applied to files:
docs/ai-chat/custom-agents.mdx
📚 Learning: 2026-06-04T18:16:35.386Z
Learnt from: nicktrn
Repo: triggerdotdev/trigger.dev PR: 3836
File: apps/supervisor/src/backpressure/backpressureMonitor.ts:3-5
Timestamp: 2026-06-04T18:16:35.386Z
Learning: When reviewing TypeScript in this repo, apply the rule “prefer type aliases over interfaces” only to data/object shapes and union/intersection type modeling. If an interface is being used as a behavioral contract for collaborators to implement (e.g., method-shape interfaces that define required behavior, such as `BackpressureLogger` / `BackpressureSignalSource` in `apps/supervisor/src/backpressure/backpressureMonitor.ts`), keep it as an `interface` and do not flag it as a type-alias-vs-interface violation.
Applied to files:
packages/core/src/v3/apiClient/runStream.tspackages/core/src/v3/sessionStreams/types.tspackages/core/src/v3/sessionStreams/router.ts
🪛 LanguageTool
.changeset/quiet-floors-hold.md
[style] ~7-~7: Consider using “who” when you are referring to a person instead of an object.
Context: ... rather than only present in the worker that received it.
(THAT_WHO)
.changeset/spry-steers-defer.md
[style] ~20-~20: For conciseness, consider replacing this expression with an adverb.
Context: ...rsation up. An injected one is consumed at the moment it is injected, so it is never also ans...
(AT_THE_MOMENT)
docs/ai-chat/custom-agents.mdx
[style] ~263-~263: Three successive sentences begin with the same word. Consider rewording the sentence or use a thesaurus to find a synonym.
Context: ...is Session's .in channel. - payload is the existing ChatTaskWirePayload deli...
(ENGLISH_WORD_REPEAT_BEGINNING_RULE)
🔇 Additional comments (16)
packages/trigger-sdk/src/v3/ai.ts (9)
28-42: LGTM!Also applies to: 1503-1521
1616-1700: LGTM!Also applies to: 1702-1765
1834-1847: LGTM!Also applies to: 1864-1880
1981-2030: LGTM!Also applies to: 2032-2069
3010-3020: LGTM!Also applies to: 3663-3675
7946-7946: LGTM!Also applies to: 8271-8271, 9134-9141, 9158-9158
9798-9800: LGTM!Also applies to: 9866-9888, 9915-9948
11007-11017: LGTM!Also applies to: 11032-11040
6738-6739: 🩺 Stability & AvailabilityKeep the
observe()cleanup unchanged.SessionChannelRouter.observe()returns{ off: () => void }, andoff()removes the observer.packages/trigger-sdk/src/v3/test/mock-chat-agent.ts (1)
7-7: LGTM!Also applies to: 394-395
packages/trigger-sdk/src/v3/test/test-session-handle.ts (1)
7-7: LGTM!Also applies to: 32-53, 262-265
packages/trigger-sdk/test/chat-messages-mailbox.test.ts (1)
36-103: LGTM!Also applies to: 126-190, 192-255, 257-304, 306-353
packages/trigger-sdk/test/chat-warm-process-reuse.test.ts (1)
23-60: LGTM!packages/trigger-sdk/test/mockChatAgent.test.ts (1)
1881-1884: LGTM!packages/trigger-sdk/test/replay-session-in.test.ts (1)
7-7: LGTM!Also applies to: 170-193, 214-229, 238-239
packages/trigger-sdk/src/v3/sessions.ts (1)
792-812: 🩺 Stability & AvailabilityNo issue:
#dispatchadvanceslastDispatchedSeqNumwhen it consumes a record, andawaitWakeguarantees that the woken record is readable beforeonceRecord()runs.
| By default (without `pendingMessages`), a message sent while the agent is responding never interrupts the in-flight response: it's buffered and processed as its own turn once the current turn completes, with multiple messages running sequentially in arrival order. | ||
|
|
||
| The `pendingMessages` option enables steering instead, injecting user messages between tool-call steps via the AI SDK's `prepareStep`. Messages that arrive during streaming are queued and injected at the next step boundary. If there are no more step boundaries (single-step response or final text generation), the message becomes the next turn automatically. | ||
| The `pendingMessages` option enables steering instead, injecting user messages between tool-call steps via the AI SDK's `prepareStep`. Messages that arrive during streaming are queued and injected at the next step boundary. A message that is not injected becomes the next turn instead, whether that is because `shouldInject` returned `false` or because there were no more step boundaries (single-step response or final text generation). Nothing is lost either way, and the backend handles it, so no client-side re-send is involved. |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Document the required chat.toStreamTextOptions() integration.
This sentence says that the backend handles every non-injected message. The PR scope retains a case where pendingMessages is configured without spreading chat.toStreamTextOptions(), and no queue consumer exists. State this requirement or fix the flow before promising that no message is lost.
| function repeatingApiClient(record: { | ||
| id: string; | ||
| recordId?: string; | ||
| chunk: unknown; | ||
| timestamp: number; | ||
| }): ApiClient { | ||
| return { | ||
| async subscribeToSessionStream<T>( | ||
| _sessionIdOrExternalId: string, | ||
| _io: "out" | "in", | ||
| options?: { onPart?: (part: SSEStreamPart<T>) => void; signal?: AbortSignal } | ||
| ) { | ||
| options?.onPart?.(record as SSEStreamPart<T>); | ||
| const signal = options?.signal; | ||
| // eslint-disable-next-line require-yield | ||
| return (async function* () { | ||
| if (signal?.aborted) return; | ||
| await new Promise<void>((resolve) => { | ||
| signal?.addEventListener("abort", () => resolve(), { once: true }); | ||
| }); | ||
| })() as unknown as Awaited<ReturnType<ApiClient["subscribeToSessionStream"]>>; | ||
| }, | ||
| } as unknown as ApiClient; | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Use the repository's container-backed stream fixture instead of mocking the client.
These tests replace the stream client with in-memory mocks. Use the container-backed interaction path so waitpoint registration, delivery, and stream behavior are exercised through the real test fixture. The same remediation applies to the pending-message drain tests.
📍 Affects 2 files
packages/core/src/v3/sessionStreams/manager.test.ts#L48-L71(this comment)packages/trigger-sdk/test/pending-message-drain.test.ts#L373-L390
Source: Coding guidelines
| clearRoute(name: string): void { | ||
| const state = this.#stateOrThrow(name); | ||
| state.queue.length = 0; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🔎 Supported by static analysis
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the router API and every invocation. Expect every call to target a
# non-replayable route such as "handover".
ast-grep outline packages/core/src/v3/sessionStreams/router.ts --items all
ast-grep run --lang ts --pattern '$OBJ.clearRoute($ROUTE)' packages
rg -n -C 3 '\.clearRoute\(' packagesRepository: triggerdotdev/trigger.dev
Length of output: 3785
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- router implementation ---'
sed -n '20,145p' packages/core/src/v3/sessionStreams/router.ts
sed -n '390,510p' packages/core/src/v3/sessionStreams/router.ts
printf '%s\n' '--- route constants and configuration ---'
rg -n -C 8 'CHAT_ROUTE_HANDOVER|replayable|handover|messages' packages/trigger-sdk/src/v3/ai.ts packages/core/src/v3/sessionStreamsRepository: triggerdotdev/trigger.dev
Length of output: 50384
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- router route types and state ---'
sed -n '20,145p' packages/core/src/v3/sessionStreams/router.ts
printf '%s\n' '--- router floor and clearRoute ---'
sed -n '430,505p' packages/core/src/v3/sessionStreams/router.ts
printf '%s\n' '--- handover references ---'
rg -n -C 6 'CHAT_ROUTE_HANDOVER' packages/trigger-sdk/src/v3/ai.ts
printf '%s\n' '--- router construction references ---'
rg -n -C 8 'SessionChannelRouter|chatInputRouter|sessionStreams|replayable:' packages/trigger-sdk/src/v3/ai.ts packages/core/src/v3/sessionStreams --glob '*.ts'Repository: triggerdotdev/trigger.dev
Length of output: 50383
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- resume and checkpoint implementation ---'
rg -n -C 14 'resumeFloor\(\)|checkpoint\(\)|earliestUnrecovered|queue\.push|state\.queue' packages/core/src/v3/sessionStreams/router.tsRepository: triggerdotdev/trigger.dev
Length of output: 8599
Reject clearing replayable routes. clearRoute("messages") removes queued user messages. Because resumeFloor() then sees no pending replayable record, a later boot can resume past those messages. Throw when state.route.replayable is true.
| disconnectStream(_sessionId: string, _io: SessionChannelIO): void { | ||
| // no-op — no real SSE tail in tests | ||
| // The production manager keeps buffered records reachable across a | ||
| // waitpoint suspension. The exact waitpoint record is removed on resume. | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Implement reconnectStream() in the test manager.
TestSessionStreamManager implements SessionStreamManager but does not define reconnectStream(). packages/trigger-sdk/src/v3/sessions.ts calls this method. Mock-context session recovery can call an undefined method and fail. Add the no-op implementation used by NoopSessionStreamManager.
| await installChatInputRouter(payload.chatId, { | ||
| resuming: Boolean(payload.continuation), | ||
| }); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🔎 Supported by static analysis
🏁 Script executed:
#!/bin/bash
# Confirm the router drops non-replayable records at or below appliedThrough,
# and that ctx.attempt.number is the field used elsewhere in ai.ts.
rg -n -C6 'replayed|appliedThrough' packages/core/src/v3/sessionStreams/router.ts
rg -n -C3 'attempt\.number' packages/trigger-sdk/src/v3/ai.ts
rg -n -C4 'installChatInputRouter\(' packages/trigger-sdk/src/v3/ai.tsRepository: triggerdotdev/trigger.dev
Length of output: 7826
🏁 Script executed:
#!/bin/bash
# Inspect the custom-agent caller, the router installation path, and the route
# table/checkpoint handling that determine whether a retry replays `stop`.
sed -n '1950,2075p' packages/trigger-sdk/src/v3/ai.ts
sed -n '5585,5650p' packages/trigger-sdk/src/v3/ai.ts
rg -n -C8 'CHAT_INPUT_ROUTES|stop.*replay|replayable.*stop|resumeFrom|appliedThrough' packages/trigger-sdk/src/v3/ai.ts packages/core/src/v3/sessionStreams/router.tsRepository: triggerdotdev/trigger.dev
Length of output: 30103
🏁 Script executed:
#!/bin/bash
# Resolve how the replay-window scan behaves without a checkpoint and how
# SessionChannelRouter handles an already-applied `stop` record.
sed -n '1924,1978p' packages/trigger-sdk/src/v3/ai.ts
sed -n '204,315p' packages/core/src/v3/sessionStreams/router.ts
rg -n -C5 'function chatAgent|const chatAgent|attempt.*number|retry|OOM|out of memory' packages/trigger-sdk/src/v3/ai.tsRepository: triggerdotdev/trigger.dev
Length of output: 23344
Include the attempt number in the custom-agent resuming flag.
When a retry has no prior turn-complete checkpoint, chatCustomAgent does not scan the replay window. The router can then deliver a previously applied non-replayable stop record to the retried turn.
| void harness.sendMessage(userMessage("m2", "u-2")); | ||
| await waitFor( | ||
| () => (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in") !== undefined | ||
| ); | ||
| const m2Seq = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in")!; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Wait for the sequence to advance past m1, not merely to exist.
lastSeqNum(chatId, "in") is already defined at this point, because m1 was received before the turn started. The waitFor predicate therefore returns true on its first check, possibly before the m2 send lands, and m2Seq can capture m1's sequence instead of m2's. harness.sendMessage is invoked with void, so nothing guarantees the append completed.
Capture the sequence before the send and wait for a strictly greater value.
🔧 Proposed fix
+ const seqBeforeM2 = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in");
void harness.sendMessage(userMessage("m2", "u-2"));
await waitFor(
- () => (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in") !== undefined
+ () => {
+ const seq = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in");
+ return seq !== undefined && (seqBeforeM2 === undefined || seq > seqBeforeM2);
+ }
);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| void harness.sendMessage(userMessage("m2", "u-2")); | |
| await waitFor( | |
| () => (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in") !== undefined | |
| ); | |
| const m2Seq = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in")!; | |
| const seqBeforeM2 = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in"); | |
| void harness.sendMessage(userMessage("m2", "u-2")); | |
| await waitFor( | |
| () => { | |
| const seq = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in"); | |
| return seq !== undefined && (seqBeforeM2 === undefined || seq > seqBeforeM2); | |
| } | |
| ); | |
| const m2Seq = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in")!; |
A message arriving while a turn was streaming was handed to the turn's push handler and parked in an in-memory array. The router counts a record handed to a handler as terminally decided, so it stopped holding the resume floor behind it and the turn boundary published a cursor past a message that existed only in this process. A crash before the next turn lost it silently. The handler is now attached only when there is a steering config to feed. Without one the record stays queued on the router, which holds the floor until a turn takes it, and both in-memory wire buffers go away. The wait path already takes from the queue before suspending, so a message that arrived mid-turn is still picked up as the next turn without a round trip. The floor also doubles as the wake cursor, so an over-advanced floor parked a waitpoint nothing would complete. It is now recorded on the wait span to make that diagnosable from a trace. Not addressed here: with a steering config, a declined message is still dropped rather than left queued. That path depends on an unresolved question about what declining should mean.
A message arriving mid-turn with a `pendingMessages` config was routed into a turn-local steering queue. If the batch was declined for injection it was discarded with the turn: never injected, never written to the wire buffer, never answered, and nothing raised at either end. Declining is also the default, since a config without `shouldInject` declines every batch, so the documented default behaviour was the losing one. Notification and consumption are now separate. `observe` on the router tells a consumer a record arrived without taking it, so the record stays queued and keeps holding the resume floor, and injection is the point of consumption: `take` removes exactly the records that were injected. A declined batch never reaches that line, so its records stay queued and become later turns, which is what the docs have always promised. `observe` is rejected on an at-arrival route. An observer there would either have to count as a listener, which would stop an unconsumed stop being discarded and bring back the wedged mailbox, or watch records it cannot affect.
The docs described a mid-turn message becoming the next turn only when there were no more step boundaries, and the client-side lifecycle credited the frontend with auto-sending it. Neither matched the behaviour: a message the agent declines to inject is now held on the backend and answered as the next turn, with no client re-send involved, and that covers an explicit `shouldInject: false` as well as a turn that never reaches a boundary. Also spells out that a declined message keeps its place in the queue, so it survives a crash rather than living only in the worker that received it.
4d2027c to
3dd60c2
Compare
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
1 participant
Follow-up to #4644, now rebased onto main so the diff is just these three commits.
Summary
Two ways a chat could lose a user message, both pre-existing and both raised while reviewing #4644.
A message arriving while a turn was streaming was handed to that turn's push handler and parked in an in-memory array. The router counts a record handed to a handler as terminally decided, so it stopped holding the resume floor behind it, and the turn boundary published a cursor past a message that existed only in that process. A crash before the next turn lost it, silently. Measured: with the message at sequence 1, the boundary published
session-in-event-id: 1, so a resume skipped it.Separately, a message the agent declined to inject was discarded with the turn. Never injected, never written to the wire buffer, never answered. That was also the documented default, since a
pendingMessagesconfig withoutshouldInjectdeclines every batch.Design
Notification and consumption are now separate concerns on the router.
observereports that a record arrived without taking it, so the record stays queued and keeps holding the floor. It is rejected on anat-arrivalroute: an observer there would either have to count as a listener, which would stop an unconsumed stop being discarded and bring back a wedged mailbox, or watch records it cannot affect.takeremoves exactly one queued record.The managed loop and the
chat.createSession()iterator now only subscribe when there is a steering config to feed, and injection is the point of consumption. A declined batch never reaches the take, so its records stay queued and become later turns. Both in-memory wire buffers are gone, so a message waiting for its turn is durable rather than living in whichever worker received it.The floor doubles as the wake cursor:
awaitWakeregisters with it and the server completes the waitpoint immediately if anything sits after that sequence. An over-advanced floor was therefore also a missed wake. It is now recorded on the wait span so a run that never woke can be diagnosed from its trace.Verification
Both fixes have a red and green pair, each checked against the unmodified source rather than only observed to pass:
Also 8 new router tests for
observeandtake. Suites green at 385 for the SDK and 886 for core.Not addressed
A
pendingMessagesconfig with nochat.toStreamTextOptions()spread still swallows messages, because nothing drains the queue at all. Same shape, different trigger, tracked separately.