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 26 commits into
Conversation
`chat.messages.next()` and `hasPending()` only inspect the head of the `.in` buffer. A control record whose kind has no consumer on this boot therefore parked at the head forever, and every message queued behind it became undeliverable with no error: `hasPending()` stayed false and `next()` timed out on every call. Records of a kind nothing on the run consumes are now discarded at dispatch instead. Consuming at dispatch keeps the resume cursor exact, since the record never enters the buffer and so leaves no unconsumed barrier for `lastDispatchedSeqNum()` to clamp behind. `message` is always claimed, and handover kinds are claimed for the window in which a handover-prepare boot is actually waiting for them. The mixed-kinds test now builds its blocked-head state on a handover-prepare boot, where the handover kind is claimed and so stays buffered for the raw read it asserts.
… kinds
Removes a JSDoc block and a `{@link}` reference to a claim-kinds helper that
is not part of this change, which also left `chat.createStopSignal` carrying
two doc comments.
The drain also warned on every record it discarded, including a stop that the
stop facade had already handled: all handlers are invoked for a record
regardless of whether an earlier one consumed it, so a stop with an active
stop signal is both aborted and drained. A known kind with no active consumer
is an expected state, so the warning is now limited to kinds this SDK version
does not recognise, which is the case that indicates a newer server.
Also corrects the docs and changeset, which claimed `next()` returning
`undefined` always means the mailbox is idle. A control record that does have
its own consumer can sit at the head while `next()` times out.
…payload `session.in.wait()` needed the exact sequence of the record it returned, so that record could be acknowledged rather than guessed at. It got that by having the server attach the sequence to the waitpoint output, which meant a new versioned wire format, a capability flag on the waitpoint request, a second Redis key so a mid-deploy instance could still drain, and a payload-matching fallback for a server that does not send the envelope. That fallback gave up whenever two records on a channel shared a payload, and gave up by returning the record to the caller without acknowledging it, so the reconnecting tail delivered it a second time. None of that is necessary. The append route commits the record to the channel before it drains any waitpoint, so once the run wakes, the record is durably readable from the channel with its real sequence. The waitpoint is now treated as a wake signal only: its output is discarded, the tail re-attaches, and the record is read back through the normal buffer path, which acknowledges it and advances the cursor exactly. This removes the wire format, the capability flag, the format key, the fallback, and every webapp change, leaving no cross-service surface and no mixed-version behaviour to reason about. It also fixes the case the fallback could not: a duplicate append whose idempotency claim was lost wakes the run, the channel has nothing new, and the run waits instead of answering a stale record twice. Adds a regression test that the delivered record is acknowledged when identical payloads repeat on a channel, and one that a message queued behind a control record nothing consumes is still delivered. Drops the three tests that only described the removed envelope and its fallback.
The `session-in-event-id` header has two consumers with opposite needs. A fresh boot reads it back as the `.in` resume cursor and needs it conservative, while a client compares it against the append sequence of its own send to recognise a turn boundary that predates that send, which needs it exact. Holding the cursor behind every unconsumed record served neither: an unconsumed control record pushed the header below the sequence of the message the turn had just answered, so a client discarded its own turn-complete and stayed streaming. The cursor is now held only behind records whose loss would matter, which for chat means messages. Replaying a stop or a handover on the next boot is benign, and a handover for a turn that never ran is discarded, so control records no longer need to hold the cursor. The manager takes the rule as a per-channel predicate and defaults to holding behind everything, so a missing or throwing predicate can only make the cursor more conservative. This also ends the case where one never-consumed record pinned the cursor for the rest of the run. Two further gaps in the same machinery: The unclaimed-kind drain and the cursor rule were installed only for `chat.customAgent`. `chat.agent` builds its task directly and got neither, so the managed agent, which is the common surface, kept accumulating barriers mid-turn. Both are now installed for both surfaces, with the drain attached after each one's resume cursor is seeded so it cannot open the subscribe at seq 0. A handover-prepare boot claims the handover kinds so a signal arriving before `waitForHandover` attaches is not drained, but the claim was released only inside `waitForHandover`. A loop that never called it held the claim for the life of the run, leaving a handover record parked at the head of the channel where it wedged `chat.messages.next()` permanently. The claim is now also released at the first turn boundary, by which point the handover window has closed either way.
…ndary The release sat in `chat.writeTurnComplete`, which only hand-rolled loops call. The managed agent reaches a turn boundary through the internal chunk writer, so a handover-prepare boot there kept the claim for the life of the run and a handover record stayed parked at the head of the channel, wedging `chat.messages.next()`. Moved to `writeTurnCompleteChunk`, which every surface goes through.
…d API docs The changeset only mentioned the new mailbox helpers, and led with them. The change a user is most likely to care about is that a chat could silently lose a message, which affected the managed agent too, so the release note now leads with that and with the retried-send duplicate. `chat.writeTurnComplete()` also promised that `sessionInEventId` identified the exact input record the turn acknowledged. It does not: it is the cursor that is safe to resume from, held back behind any message still waiting to be handled, so a value below the record just handled is expected rather than a fault.
…gain on resume Holding the resume cursor behind a message still waiting to be handled means resuming from it necessarily re-delivers every record after that point, including records the previous run had already handled. For a message that is the entire point. For a control record it is a fault: a stop carries no record of which turn it belonged to, so on redelivery it aborts whichever turn happens to be live, which is usually the turn answering the very message the cursor was held back to protect. The user got their answer cut off instead of never arriving, which is better but still wrong. Each turn boundary now also reports the highest sequence that run had actually consumed, unclamped, alongside the cursor that is safe to resume from. On boot a run reads it back and drops control records at or below it before any consumer sees them. Messages are never dropped, so recovery is unchanged. A chat whose turns predate the header reports nothing, drops nothing, and behaves as before. Driven on a stack: a message set aside, a stop consumed, a turn boundary, a SIGKILL, then a continuation. Before, the continuation consumed the message and was immediately aborted by the replayed stop. Now the stop is dropped and the turn proceeds, while a stop that arrives live still aborts its turn.
…line Adds the session channel router: one reader that classifies every record once and gives it exactly one destination, instead of several kind-filtered facades each taking records off a shared buffer. A route declares two independent things: whether it queues a record when no consumer is ready, and whether a record it never handled has to survive into the next boot. Those two make the resume floor, the replay window and the discard-the-unowned behaviour derived properties rather than three predicates installed by hand. Not wired to anything yet. The chat layer keeps its current mechanisms until the reproduction catalog is green on the router.
…for every consumer `session.in` carries records for several consumers whose delivery needs differ: a user message must be delivered eventually and so can lag arbitrarily, while a stop only means anything to the turn that is live when it lands. One scalar cursor cannot describe both, and each fix so far has been a workaround for that: a clamp so the cursor retreats below a queued message, a barrier predicate so control records do not make it retreat, a drop predicate and a second header so the retreat does not re-apply control records that were already applied, and a drain so a record with no consumer does not park at the head of a shared buffer. The router replaces all of it. Every record is classified once and handed to one route; each route declares whether it queues when no consumer is ready and whether an unhandled record must survive into the next boot. The resume floor, the replay window and the discard-the-unowned behaviour then fall out of those declarations, so `hasPending` and `next` read their own route's queue rather than the head of a buffer shared with three other kinds, and the checkpoint is read and the subscription opened in one call, closing the window where a listener could attach before the resume cursor was seeded. No wire change: both turn-boundary headers keep their meanings, so existing sessions resume exactly as before and no webapp change is involved. Also converges the test double on production semantics. It decided consumption after awaiting handlers, which left a window where a handler registered mid-dispatch saw a buffer the record had not been added to yet, and it carried its own copy of the cursor clamp. It now decides synchronously like production and stubs only the network boundary, so the wait path runs its real implementation in tests.
…lder SDK A turn boundary written before the replay window was published carries only the resume cursor. Resuming one, the run had no way to tell a control record it was re-reading from one arriving live, so a stop that a previous run had already applied was applied a second time, aborting the turn answering the very message the cursor was held back to protect. Queued messages were never at risk. Boundaries that predate the window now resolve it from the channel: everything already on `.in` at boot is by definition not arriving live on this run, so it is the end of that run's replay window. Bounded by `afterEventId`, so the read covers the replay window rather than the whole conversation, and only on the first turn after an upgrade. The trade is deliberate. A stop that landed legitimately in the moments before boot is now dropped along with the replayed ones. Missing a stop leaves the user able to press stop again; applying a stale one kills an answer they are waiting for. Verified on a deployed test-cloud project across two versions: a session driven to a crash on a build that writes the old boundary format, then continued on this build. Before, the continuation reported the stop aborting its turn; after, it declines the replayed stop and the turn answering the recovered message completes.
The mailbox section described head-of-line behaviour that no longer happens: it said a control record arriving before a message keeps `hasPending()` false and makes `next()` wait, when a message behind a stop or a handover is now reported and returned normally. Docs that state the opposite of what ships are worse than none, so those claims are replaced with what the delivery guarantee actually is. Also documents two things that had no coverage. A stop applies only to the turn that was live when it arrived, so a recovered message's turn is not aborted by a stop from before the crash. And `sessionInEventId` is a lower bound rather than the sequence of the record a turn answered, which is the mistake a client makes when it tries to match a turn boundary to its own send.
…turn boundary A stop applied by a run that then died was still being applied a second time on recovery whenever no turn boundary recorded it. A boundary is written when a turn ends, so a stop that arrives after the last one is not covered by it, and the previous fix only helped when the boundary carried no window at all. The result was a recovered turn aborted by a stop the user had pressed against a turn that no longer exists. Everything already on the channel when a run boots is, by definition, not arriving live on that run, so the window is now taken from the channel's own tail and the boundary's value is only a floor for it. Queued messages are unaffected, since they are replayable and never dropped. Found by extending the reproductions to the managed `chat.agent` surface, which reaches this case naturally: the managed loop writes its boundary at the end of each turn, so a stop arriving mid-turn lands after it.
… resuming The previous commit took the replay window from the channel's tail so a stop that no turn boundary covered could not be applied twice. On a first boot that is wrong: nothing has been applied by anyone yet, so anything already on the channel was treated as replayed and dropped. That broke head starts. The client can signal a handover before the agent run has booted, which is the whole point of the flow, and the signal was discarded as already-applied. The agent then waited out its idle window and lost the warm partial. A replay window now only exists for a run that is resuming, which is either a run whose predecessor left a turn boundary or one the wire marks as a continuation or a retried attempt. A first boot has no window and applies what it finds.
A worker process is reused across runs, and the executor tears the channel subscription down at the end of each one. The router was cached per chat, so a second run of the same chat in the same process skipped the attach and then had no subscription feeding it: no messages arrived, and the conversation hung with no error raised. Caching it per run instead means every run attaches its own subscription, while a nested `chat.createSession` inside one run still shares the router. Reported by review, reproduced first as a failing test that drives two runs through one process with the executor's teardown in between.
The pendingMessages tests covered the wire-buffer path, where no steering config exists and messages become later turns, but nothing covered shouldInject or onReceived. That left the whole steering delivery path uncovered, and its failure mode is silent: a steer appends with a 2xx, the agent never sees it, and nothing is raised anywhere. Asserts the per-turn handler receives a message that lands while the turn is streaming, which is the first thing any steering integration depends on.
The recovered answer is persisted correctly, but a chat page open across the crash keeps showing the partial it already received. Say so in the changeset rather than letting the release note imply the recovery is visible without reloading.
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.
🦋 Changeset detectedLatest commit: 4d2027c 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 |
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")!; |
Stacked on #4644 and should merge after it. That PR's head is on a fork, so GitHub cannot base this one on it directly; until it merges, the diff here also shows its commits. The three commits that belong to this PR are the top three.
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.