feat(orchestrator): queue follow-up messages during active turns, with explicit steer#4245
feat(orchestrator): queue follow-up messages during active turns, with explicit steer#4245t3dotgg wants to merge 7 commits into
Conversation
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
github-actions
Bot
added
vouch:trusted
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
ApprovabilityVerdict: Needs human review This PR introduces a substantial new feature (message queueing with steer capability) including a new database table, persistence layer, orchestration commands/events, complex decider logic, provider protocol integration, and new UI components. The scope and complexity warrant human review, and unresolved comments identify potential data integrity and stability issues. You can customize Macroscope's approvability policy. Learn more. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/web/src/components/ChatView.tsx (1)
3682-3717: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winRevoke previews acknowledged only by the queue.
Queued messages are absent from
displayServerMessages, so their blob URLs remain inattachmentPreviewHandoffByMessageId. Removing such a queued image never promotes the handoff and leaks the object URLs until unmount.Proposed fix
- const serverIds = new Set([ - ...activeThread.messages.map((message) => message.id), - ...activeThread.queuedMessages.map((message) => message.messageId), - ]); + const persistedMessageIds = new Set(activeThread.messages.map((message) => message.id)); + const serverIds = new Set([ + ...persistedMessageIds, + ...activeThread.queuedMessages.map((message) => message.messageId), + ]); @@ - if (previewUrls.length > 0) { + if (previewUrls.length > 0 && persistedMessageIds.has(removedMessage.id)) { handoffAttachmentPreviews(removedMessage.id, previewUrls); continue; } revokeUserMessagePreviewUrls(removedMessage);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/web/src/components/ChatView.tsx` around lines 3682 - 3717, Update the optimistic-message cleanup effect around activeThread.queuedMessages so previews for queued-only acknowledgements are also removed from attachmentPreviewHandoffByMessageId. Ensure the queue acknowledgment path promotes or revokes the handed-off blob URLs when the queued image is removed, while preserving existing cleanup for messages present in activeThread.messages.
🧹 Nitpick comments (2)
packages/client-runtime/src/state/threadReducer.ts (1)
278-316: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover queued-message reducer lifecycle.
Add reducer cases for queueing, replacing an existing
messageId, and removing it. The changed fixture only validates the expanded shape, not these new state transitions. Verify withvp test run packages/client-runtime/src/state/threadReducer.test.ts.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/client-runtime/src/state/threadReducer.ts` around lines 278 - 316, The threadReducer lifecycle coverage is incomplete for queued messages. Update the reducer handling around the “thread.message-queued” and “thread.queued-message-removed” cases to support adding a queued message, replacing an existing entry with the same messageId, and removing it while preserving updatedAt and the full queued-message shape; add or update tests in threadReducer.test.ts and run the specified test command.Source: Coding guidelines
apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts (1)
19-24: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSingle-source the queued-message DB row schema instead of duplicating it. Both files define an identical
ProjectionQueuedMessageDbRowSchematransform; drift between them would go unnoticed at compile time since each is a separate local const.
apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts#L19-L24: export this schema (or move it intoServices/ProjectionQueuedMessages.ts) as the canonical definition.apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts#L80-L85: import the canonical schema instead of redefining it locally.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts` around lines 19 - 24, Make ProjectionQueuedMessageDbRowSchema in apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts the exported canonical schema. In apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts, import and reuse that exported schema instead of defining a local duplicate; preserve the existing transform fields and behavior at both sites.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts`:
- Around line 1450-1462: Update getCommandReadModel’s queuedMessageRows loop to
advance updatedAt with each row’s queuedAt timestamp, using the same maxIso
logic as getSnapshot. Keep the existing queuedMessagesByThread grouping and skip
behavior unchanged.
In `@apps/server/src/provider/Layers/CodexSessionRuntime.ts`:
- Around line 1290-1326: Restrict the error handling in the turn/steer request
within the activeSession branch to the specific not-steerable/precondition
failure, allowing transient RPC or protocol errors to propagate instead of
triggering turn/start. Preserve the existing fallback behavior for that matched
case and keep successful steer responses unchanged.
---
Outside diff comments:
In `@apps/web/src/components/ChatView.tsx`:
- Around line 3682-3717: Update the optimistic-message cleanup effect around
activeThread.queuedMessages so previews for queued-only acknowledgements are
also removed from attachmentPreviewHandoffByMessageId. Ensure the queue
acknowledgment path promotes or revokes the handed-off blob URLs when the queued
image is removed, while preserving existing cleanup for messages present in
activeThread.messages.
---
Nitpick comments:
In `@apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts`:
- Around line 19-24: Make ProjectionQueuedMessageDbRowSchema in
apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts the exported
canonical schema. In
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts, import and
reuse that exported schema instead of defining a local duplicate; preserve the
existing transform fields and behavior at both sites.
In `@packages/client-runtime/src/state/threadReducer.ts`:
- Around line 278-316: The threadReducer lifecycle coverage is incomplete for
queued messages. Update the reducer handling around the “thread.message-queued”
and “thread.queued-message-removed” cases to support adding a queued message,
replacing an existing entry with the same messageId, and removing it while
preserving updatedAt and the full queued-message shape; add or update tests in
threadReducer.test.ts and run the specified test command.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: f7596e4b-fb86-4489-8567-5937c39efcea
📒 Files selected for processing (36)
apps/mobile/src/lib/threadActivity.test.tsapps/server/src/orchestration/Layers/OrchestrationEngine.test.tsapps/server/src/orchestration/Layers/ProjectionPipeline.tsapps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.tsapps/server/src/orchestration/Layers/ProjectionSnapshotQuery.tsapps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.tsapps/server/src/orchestration/Layers/ProviderRuntimeIngestion.tsapps/server/src/orchestration/Schemas.tsapps/server/src/orchestration/commandInvariants.test.tsapps/server/src/orchestration/decider.queue.test.tsapps/server/src/orchestration/decider.tsapps/server/src/orchestration/projector.test.tsapps/server/src/orchestration/projector.tsapps/server/src/persistence/Layers/ProjectionQueuedMessages.tsapps/server/src/persistence/Migrations.tsapps/server/src/persistence/Migrations/033_ProjectionQueuedMessages.tsapps/server/src/persistence/Services/ProjectionQueuedMessages.tsapps/server/src/provider/Layers/CodexSessionRuntime.tsapps/server/src/provider/Layers/ProviderSessionReaper.test.tsapps/server/src/server.test.tsapps/server/src/ws.tsapps/web/src/components/ChatView.logic.test.tsapps/web/src/components/ChatView.logic.tsapps/web/src/components/ChatView.tsxapps/web/src/components/CommandPalette.logic.test.tsapps/web/src/components/Sidebar.logic.test.tsapps/web/src/components/chat/QueuedMessageChips.tsxapps/web/src/lib/threadSort.test.tsapps/web/src/worktreeCleanup.test.tspackages/client-runtime/src/operations/commands.tspackages/client-runtime/src/state/entities.test.tspackages/client-runtime/src/state/threadCommands.tspackages/client-runtime/src/state/threadReducer.test.tspackages/client-runtime/src/state/threadReducer.tspackages/client-runtime/src/state/threads-sync.test.tspackages/contracts/src/orchestration.ts
| const queuedMessagesByThread = new Map<string, Array<OrchestrationQueuedMessage>>(); | ||
| const sessionByThread = new Map<string, OrchestrationSession>(); | ||
|
|
||
| for (let index = 0; index < queuedMessageRows.length; index += 1) { | ||
| const row = queuedMessageRows[index]; | ||
| if (!row) { | ||
| continue; | ||
| } | ||
| const threadQueuedMessages = queuedMessagesByThread.get(row.threadId) ?? []; | ||
| threadQueuedMessages.push(mapQueuedMessageRow(row)); | ||
| queuedMessagesByThread.set(row.threadId, threadQueuedMessages); | ||
| } | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
getCommandReadModel doesn't advance updatedAt from queued-message timestamps.
getSnapshot's queued-message loop does updatedAt = maxIso(updatedAt, row.queuedAt) (line 1154), but the equivalent loop here (1453-1461) only builds queuedMessagesByThread without touching updatedAt. None of the earlier updatedAt-tracking loops (1382-1439) cover queuedMessageRows either, so a thread whose only new activity is a queued message won't bump the command read model's updatedAt.
🐛 Proposed fix
for (let index = 0; index < queuedMessageRows.length; index += 1) {
const row = queuedMessageRows[index];
if (!row) {
continue;
}
+ updatedAt = maxIso(updatedAt, row.queuedAt);
const threadQueuedMessages = queuedMessagesByThread.get(row.threadId) ?? [];
threadQueuedMessages.push(mapQueuedMessageRow(row));
queuedMessagesByThread.set(row.threadId, threadQueuedMessages);
}📝 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.
| const queuedMessagesByThread = new Map<string, Array<OrchestrationQueuedMessage>>(); | |
| const sessionByThread = new Map<string, OrchestrationSession>(); | |
| for (let index = 0; index < queuedMessageRows.length; index += 1) { | |
| const row = queuedMessageRows[index]; | |
| if (!row) { | |
| continue; | |
| } | |
| const threadQueuedMessages = queuedMessagesByThread.get(row.threadId) ?? []; | |
| threadQueuedMessages.push(mapQueuedMessageRow(row)); | |
| queuedMessagesByThread.set(row.threadId, threadQueuedMessages); | |
| } | |
| const queuedMessagesByThread = new Map<string, Array<OrchestrationQueuedMessage>>(); | |
| const sessionByThread = new Map<string, OrchestrationSession>(); | |
| for (let index = 0; index < queuedMessageRows.length; index += 1) { | |
| const row = queuedMessageRows[index]; | |
| if (!row) { | |
| continue; | |
| } | |
| updatedAt = maxIso(updatedAt, row.queuedAt); | |
| const threadQueuedMessages = queuedMessagesByThread.get(row.threadId) ?? []; | |
| threadQueuedMessages.push(mapQueuedMessageRow(row)); | |
| queuedMessagesByThread.set(row.threadId, threadQueuedMessages); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts` around lines
1450 - 1462, Update getCommandReadModel’s queuedMessageRows loop to advance
updatedAt with each row’s queuedAt timestamp, using the same maxIso logic as
getSnapshot. Keep the existing queuedMessagesByThread grouping and skip behavior
unchanged.
|
|
||
| // Steer the active turn via the protocol-native `turn/steer` — | ||
| // appending input mid-turn instead of opening a second provider | ||
| // turn while the first is still settling. The `expectedTurnId` | ||
| // precondition fails when the turn just ended or is not steerable | ||
| // (e.g. review/compact); fall back to a fresh `turn/start`. | ||
| const activeSession = yield* Ref.get(sessionRef); | ||
| if (activeSession.status === "running" && activeSession.activeTurnId !== undefined) { | ||
| const steeredTurnId = yield* client | ||
| .request("turn/steer", { | ||
| threadId: providerThreadId, | ||
| expectedTurnId: activeSession.activeTurnId, | ||
| input: params.input, | ||
| }) | ||
| .pipe( | ||
| Effect.map((steerResponse) => TurnId.make(steerResponse.turnId)), | ||
| Effect.catch((cause) => | ||
| Effect.as( | ||
| Effect.logDebug("Codex turn/steer rejected; falling back to turn/start.", { | ||
| cause, | ||
| }), | ||
| null, | ||
| ), | ||
| ), | ||
| ); | ||
| if (steeredTurnId !== null) { | ||
| const steeredProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); | ||
| return { | ||
| threadId: options.threadId, | ||
| turnId: steeredTurnId, | ||
| ...(steeredProviderThreadId | ||
| ? { resumeCursor: { threadId: steeredProviderThreadId } } | ||
| : {}), | ||
| } satisfies ProviderTurnStartResult; | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect the Codex client's typed error surface for turn/steer to see whether
# a specific "not steerable" tag/reason is available to narrow this catch.
rg -n 'ActiveTurnNotSteerable|turn/steer' apps/server/src/provider -g '*.ts'
ast-grep run --pattern 'turn/steer' --lang typescript apps/server/src/providerRepository: pingdotgg/t3code
Length of output: 543
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the local implementation and any typed error definitions related to turn/steer.
rg -n --hidden --glob '!node_modules' --glob '!dist' --glob '!build' \
'ActiveTurnNotSteerable|turn/steer|not steerable|steer rejected|turn/start' \
apps/server/src vendor .repos 2>/dev/null || true
echo '--- CodexSessionRuntime.ts (around the review range) ---'
sed -n '1270,1335p' apps/server/src/provider/Layers/CodexSessionRuntime.ts
echo '--- Files in apps/server/src/provider likely related to schema/error types ---'
fd -a -t f '.*(schema|Schema|error|Error|protocol|client).*\.ts$' apps/server/src/provider 2>/dev/null || trueRepository: pingdotgg/t3code
Length of output: 5664
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- apps/server/src/provider/Errors.ts ---'
cat -n apps/server/src/provider/Errors.ts
echo '--- turn/steer-related request/error mapping in CodexSessionRuntime.ts ---'
rg -n 'mapError|RequestError|Request.*Error|turn/steer|turn/start|not steerable|ActiveTurn' apps/server/src/provider/Layers/CodexSessionRuntime.ts apps/server/src/provider/Layers/CodexAdapter.ts apps/server/src/provider/Layers/ClaudeAdapter.ts apps/server/src/provider/Errors.ts
echo '--- nearby CodexSessionRuntime request helper definitions ---'
sed -n '1,220p' apps/server/src/provider/Errors.tsRepository: pingdotgg/t3code
Length of output: 20493
Scope the turn/steer fallback to the not-steerable case. Effect.catch treats every turn/steer failure as a benign precondition miss and immediately starts a new turn. That can mask transient RPC/protocol errors and launch a duplicate turn while the original one is still active. Let other failures propagate, or match only the specific not-steerable error.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/server/src/provider/Layers/CodexSessionRuntime.ts` around lines 1290 -
1326, Restrict the error handling in the turn/steer request within the
activeSession branch to the specific not-steerable/precondition failure,
allowing transient RPC or protocol errors to propagate instead of triggering
turn/start. Preserve the existing fallback behavior for that matched case and
keep successful steer responses unchanged.
299d961 to
f222834
Compare
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
|
Addressed all review feedback (rebased onto latest main; migration renumbered 033 → 034 after #4243 took slot 033): Bugbot — queue cap desync (high): Fixed by moving the 50-message cap into the decider: a CodeRabbit — unscoped CodeRabbit — blob preview leak (minor): Optimistic sends acknowledged only by the queue now revoke their blob preview URLs instead of handing them off (queued chips render text only, so the handoff would never be promoted). CodeRabbit — CodeRabbit — reducer test coverage (trivial): Added threadReducer lifecycle tests for queue, replace-by-messageId, and remove. An independent gpt-5.6-sol (Codex CLI) review of the full branch diff found two additional issues, both fixed:
Validation: monorepo typecheck clean, 🤖 Generated with Claude Code |
…h explicit steer Messages sent while a turn is running are now held server-side in a per-thread queue and drained automatically on natural turn completion. Queued messages render as chips above the composer with Steer (dispatch now, steering the active turn) and Remove actions. Codex steers use the protocol-native turn/steer with a turn/start fallback. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A mid-turn send now projects as a queued message rather than a steered timeline message, so the composer's local-dispatch acknowledgment also watches the queued-message tail. Remove the unused listAll repository method. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ew handling - Move the 50-message queue cap from the projector into the decider so the event stream, in-memory read model, and SQL projection can never diverge (Bugbot); overflow rejects with an invariant error surfaced to the composer. Cap covered by a new decider test. - Scope the Codex turn/steer fallback to CodexAppServerRequestError so transport/protocol failures propagate instead of racing a duplicate turn/start (CodeRabbit). - Only hand off optimistic blob previews for messages entering the timeline; queued-only acknowledgments revoke them, fixing an object-URL leak (CodeRabbit). - Advance getCommandReadModel updatedAt from queued-message timestamps (CodeRabbit). - Cover queued-message lifecycle in the threadReducer tests (CodeRabbit). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ents on revert - Reject thread.queue.steer while the session is starting: there is no provider turn to steer into yet, and dispatching would race the pending turn/start with a second provider turn. The message stays queued and the composer surfaces the invariant error. - Include queued-message attachments in the retained set during checkpoint-revert pruning so reverting a thread no longer deletes image files still referenced by queued messages. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
c36ded6 to
148c355
Compare
…ion consistency - Correlate composer 'Sending' acknowledgment with the dispatched messageId (timeline or queue projection) instead of queue-tail heuristics that another client's queue activity could satisfy (Bugbot + Macroscope). Dispatches without a known messageId keep the legacy latest-user-message heuristic. - Drain queued messages in rowid (insertion) order instead of queued_at + message_id, which mis-ordered same-timestamp messages after a restart (Macroscope). Requeue replaces the row so an edited messageId moves to the tail, matching the in-memory projector. - Clear queuedMessages in the in-memory projector on thread.deleted, mirroring the SQL projection (Bugbot). - Bootstrap the queuedMessages projector before threadMessages so revert pruning sees queued attachments during replay (Macroscope). - Allow steer while a session reports 'starting' with a live activeTurnId (provider reconnect mid-turn); only the pending-start shape (starting, no active turn) is rejected (Bugbot). - Settle mock provider sessions in ProviderCommandReactor tests whose follow-up sends now queue-by-default against a still-running session. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
148c355 to
6cf940c
Compare
|
Rebased onto latest main and addressed the second round of review findings: Bugbot / Macroscope — composer acknowledges unrelated queue changes (medium): Rewrote the acknowledgment to be correlated instead of heuristic. Macroscope — same-timestamp queue mis-ordering after restart (medium): Queue drain order is now SQLite Bugbot — deleted thread keeps in-memory queue (low): The in-memory projector now clears Macroscope — bootstrap projector ordering deletes queued attachments (high): The Bugbot — steer blocked during mid-turn reconnect (medium): The steer guard now rejects only the pending-start shape ( Also updated Validation: monorepo typecheck clean, 🤖 Generated with Claude Code |
…ases, attachment cleanup - Add pendingTurnStart to the thread read model, set by thread.turn-start-requested and cleared when the session adopts the turn (running with activeTurnId) or fails/stops. The decider's queue-by-default, drain, and steer guards now cover the window between dispatching a turn start and the provider reporting session status — a send racing in behind a drain queues instead of opening a second concurrent turn (Bugbot + Macroscope). Hydrated from the existing SQL pending-turn-start rows on restart; the settle guard keeps its timestamp heuristic (renamed hasRecentUnadoptedUserMessage) since client-supplied clocks must not drive dispatch decisions. - Acknowledge a projected expectedMessageId in every composer phase, not just running: sends during 'connecting' also queue server-side (Bugbot + Macroscope). - Prune attachment files when a queued message is removed by the user; dispatch removals keep them since the same attachments re-enter the timeline via the paired message-sent (Macroscope). - Restrict the Codex turn/steer fallback to receive-response request errors (server rejected the steer). Response-decode failures on an accepted steer propagate instead of double-sending the input via turn/start (Macroscope). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Round 3 findings addressed: Bugbot — idle send bypasses pending queue / Macroscope — drain race window (medium): This was the real design gap in the round. Fixed with an event-driven signal instead of timestamps: the thread read model now carries Bugbot + Macroscope — connecting-phase sends never acknowledge (medium): The projected Macroscope — queued-message removal leaks attachment files (medium): User removals now record a prune of the thread's attachment set (retained timeline + remaining queue) so the removed message's files are deleted. Dispatch removals intentionally keep everything — the same attachments re-enter the timeline via the paired Macroscope — steer fallback on decode failures duplicates input (in the drain-race comment): The Codex Validation: monorepo typecheck clean, 🤖 Generated with Claude Code |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 2218b91. Configure here.
…ard, hydrate detail snapshot - Clear pendingTurnStart on every settled session status (including ready), in both the in-memory projector and the SQL projection. A mid-turn steer re-arms the flag with the session already running, so only terminal-status clearing left it stuck after natural completion, permanently rejecting queue drains. Ready-with-genuinely-pending starts never project as ready — ingestion maps that shape to starting. - Reject steer while a turn start awaits adoption whenever the session tracks no active turn, including running with a null activeTurnId. - Hydrate pendingTurnStart in getThreadDetailById so thread-detail snapshots agree with the command read model after a refresh. - Decider test covers the steer-then-complete-then-drain lifecycle. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Round 4 addressed — all three were real consequences of round 3's Bugbot — stuck pending blocks drain (high): Confirmed. A mid-turn steer emits Bugbot — steer races pending adoption (high): The guard now rejects steer whenever a turn start awaits adoption and the session tracks no active turn — including Bugbot — detail snapshot drops pending start (medium): Validation: typecheck clean, 🤖 Generated with Claude Code |
|
Need |
2 participants
Summary
Messages sent while a turn is running are now queued server-side by default and drained automatically when the turn completes naturally. Each queued message renders as a chip above the composer with two actions:
turn/steerwith anexpectedTurnIdprecondition, falling back toturn/start; other providers steer through their existing sendTurn-while-running paths).This replaces the previous behavior where a mid-turn send always steered immediately.
How it works
thread.queue.steer/thread.queue.remove, internalthread.queue.drain, eventsthread.message-queued/thread.queued-message-removed, andqueuedMessageson the thread schema (decoding default keeps old snapshots valid).thread.turn.starton a busy thread (sessionstarting/running) emitsthread.message-queuedinstead of starting a turn; bootstrap starts are exempt. Steer dispatches unconditionally (degrades to a normal turn start if the thread went idle). Drain is invariant-guarded: it refuses when the thread is busy again, so a user message racing the drain wins and the queue retries on the next completion.turn.completed— never after an interrupt or failure — and is dispatched after assistant finalization so the drained message sequences after the assistant text it follows up on.projection_queued_messagestable (migration 033) so queues survive server restarts.QueuedMessageChipsabove the composer; the composer's local-dispatch acknowledgment treats a projected queued message as server acknowledgment so it doesn't stick in "Sending".thread.message-sentevent when dispatched. StalesourceProposedPlanreferences are dropped rather than failing the send. Queue capped at 50 per thread.Known gaps (follow-ups)
resolveThreadOutboxDeliveryActionwaits onthreadBusy), so mobile sends generally won't hit the server queue — but once its outbox dispatches into a turn that just started, the message queues server-side invisibly. Mobile chips + steer/remove need a follow-up.Validation
vp run typecheckclean across the monorepovp checkclean on all 36 changed filesdecider.queue.test.ts) covering queue-on-busy, steer, remove, drain, drain-races, and stale plan referencesapps/serversuite passes exceptProviderRegistry.test.ts > re-probes when settings change the codex binaryPath, which fails only under the full-suite run and passes in isolation and on main — pre-existing flake, unrelated to this diff🤖 Generated with Claude Code
Note
High Risk
Changes core turn-start semantics, event sourcing projections, and provider steering paths across server, persistence, and clients; incorrect queue/drain or pending-turn handling could drop messages, double-dispatch, or leave threads stuck.
Overview
Follow-ups sent while a turn is starting or running are queued server-side instead of immediately starting another turn or steering. Idle threads still get an immediate
thread.message-sent+thread.turn-start-requested; busy threads emitthread.message-queued(50-message cap enforced in the decider).New commands
thread.queue.steer,thread.queue.remove, and internalthread.queue.draindispatch or drop queued items. After a natural turn completion, runtime ingestion auto-dispatches drain so the FIFO head becomes the next turn.pendingTurnStarton the thread read model blocks duplicate drains and follow-up sends in the gap before the provider adopts a turn; session-set projection clears stale pending rows on settled statuses (includingreadyafter mid-turn steer).Persistence: migration 034 adds
projection_queued_messages, a dedicated projector (ordered before thread-messages on replay so revert pruning keeps queued attachments), and snapshot/query hydration ofqueuedMessages+pendingTurnStart. Revert and user-removal paths update attachment pruning to include queued rows.Web:
QueuedMessageChipsabove the composer (steer/remove); composer acknowledgment correlates sends bymessageIdwhen timeline or queue projects the message. CodexsendTurntries protocolturn/steerwithexpectedTurnIdbefore falling back toturn/start. Contracts, client-runtime reducer/commands, and WS detail events extended for the new queue events.Reviewed by Cursor Bugbot for commit 6be48eb. Bugbot is set up for automated code reviews on this repo. Configure here.
Note
Queue follow-up messages during active turns and surface them as steerable chips in the chat UI
thread.turn.startcommand arrives while a turn is active or pending, the decider now emitsthread.message-queuedinstead of starting a turn immediately, capped at 50 queued messages per thread.thread.queue.drainto send the oldest queued message, enabling sequential follow-up delivery.thread.queue.steer(dispatch a specific queued message into the running turn via a protocol-native steer),thread.queue.remove(drop a queued message), andthread.queue.drain(internal drain after turn completion).QueuedMessageChipscomponent.projection_queued_messagesDB table (migration 034) persists queued messages server-side; the projector and snapshot query propagatequeuedMessagesandpendingTurnStartthrough the read model.messageIdappearing in either the timeline or the queue, replacing the global tail heuristic.CodexSessionRuntimefalls back toturn/startonly on a specificreceive-responserejection; other errors propagate without fallback.Macroscope summarized 6be48eb.
Summary by CodeRabbit