Open
Conversation
Refactor and extend abort/stop behavior to support targeted user stops. Introduces userStopRequest/plan/result types and handleUserStop/executeUserStopPlan to resolve and execute room-wide, active-turn, or queued-turn stops. Replaces direct abortRoom calls with handleUserStop in command and message handlers. Adds assistantStopMetadata and propagates it into streamingState and UI message metadata (including response status mapping). Tracks room run targets (source/initial events) and binds streaming state to room runs. Implements queue operations to drain or remove pending items by source event and finalizes stopped queue items, preserving ACK reaction removal and session notifications. Adjusts streaming finish logic to treat cancelled vs stop reasons appropriately. Includes unit tests for plan resolution, queued removal, and metadata emission.
Move pendingQueueHasSourceEvent into pending_queue.go and implement proper locking (pendingQueuesMu and queue.mu) to safely inspect queue items. Simplify drainPendingQueue to delete the queue map entry and return its items directly. Remove the duplicate helper from abort_helpers.go. Improve formatAbortNotice by capitalizing each sentence part and joining them with ". " for clearer messages. Remove redundant run field assignments in roomRunTarget. Adjust finishStreamingWithFailure to fall through from "cancelled" to "stop" so cancelled streams call End like stop cases and remove some redundant nil checks. These changes tidy concurrency handling, clarify abort messaging, and simplify streaming error handling.
Change streamingState.stop to an atomic.Pointer[assistantStopMetadata] and update all callsites to use .Load()/.Store() to avoid races. Fix pending queue drain to lock the queue when accessing items to prevent concurrent access. Improve room run logic to prefer current state.turn when present and store stop metadata atomically when marking a run stopped. Use utf8 + unicode for correct Unicode-aware capitalization in abort notices and update tests to store stop metadata via the new atomic API.
Move AckReactionRemoveAfter checks into removePendingAckReactions and remove duplicate guarded calls throughout the codebase, so callers simply invoke the removal and the function decides whether to act. Simplify pending queue management by replacing clearPendingQueue usage with drainPendingQueue and delete the pendingQueueHasSourceEvent helper. Adjust stop-plan handling to speculatively return queued stops and add a fallback in executeUserStopPlan to convert a queued plan to no-match if nothing was drained. Update tests to reflect the new speculative behavior and the fallback.
📝 WalkthroughSummary by CodeRabbit
WalkthroughImplements a structured user-stop pipeline: resolve a stop plan, execute targeted cancellations for active runs, queued turns, or whole rooms, propagate stop metadata into streaming state and UI metadata, add queue-drain/removal helpers, and add unit tests for stop behaviors. (34 words) Changes
Sequence DiagramsequenceDiagram
participant User as User
participant Handler as Stop Handler
participant Client as AIClient
participant Queue as Pending Queue
participant Streaming as Streaming State
participant UI as UI Metadata
User->>Handler: trigger stop (command or text)
Handler->>Client: handleUserStop(userStopRequest)
Client->>Client: resolveUserStopPlan()
alt room-wide
Client->>Client: mark room run stopped / cancel run
Client->>Queue: drainPendingQueue(roomID)
else active-turn
Client->>Client: cancel active run
Client->>Streaming: markRoomRunStopped(metadata)
else queued-turn
Client->>Queue: removePendingQueueBySourceEvent(sourceEventID)
else no-match
Client->>Client: return no-match result
end
Client->>Client: finalizeStoppedQueueItems()
Client->>Streaming: store stop metadata (atomic)
Client->>UI: buildStreamUIMessage with stop metadata
Handler->>User: send formatted abort notice
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@bridges/ai/abort_helpers.go`:
- Around line 159-165: The active-stop branch can race with the run disappearing
so you must not report “Stopped that turn” when nothing was persisted or
cancelled; after calling oc.markRoomRunStopped and oc.cancelRoomRun in the
stopPlanKindActive branch, re-check whether the target run still exists (e.g.,
re-run resolveUserStopPlan for the same plan/roomID) or, if
markRoomRunStopped/cancelRoomRun returned false/no-op, set result.Plan.Kind =
stopPlanKindNoMatch so formatAbortNotice will produce the downgraded message;
update the stopPlanKindActive branch around markRoomRunStopped, cancelRoomRun,
and result.SubagentsStopped to flip to stopPlanKindNoMatch on no-op/race.
In `@bridges/ai/client.go`:
- Around line 669-672: The defer currently guarded by hasDBMessage prevents
removePendingAckReactions from running for direct-dispatch debounced messages;
remove the hasDBMessage conditional and call
oc.removePendingAckReactions(oc.backgroundContext(ctx), portal,
queueItem.pending) unconditionally in that defer (the removePendingAckReactions
implementation already internally guards), so ack reactions are always cleared
regardless of the hasDBMessage path.
In `@bridges/ai/pending_queue.go`:
- Around line 88-109: The drainPendingQueue/clearPendingQueue duo currently
removes queued items from oc.pendingQueues (in drainPendingQueue) but does not
perform item finalization (removing ack reactions and notifying rooms), so
callers like clearPendingQueue and code in subagent_registry.go/client.go can
silently drop items; update the API so draining also finalizes each
pendingQueueItem: either merge finalize logic into drainPendingQueue (and have
clearPendingQueue call it) or add a new DrainAndFinalizePendingQueue helper that
calls oc.stopQueueTyping(roomID) and then iterates the returned
[]pendingQueueItem to run the existing finalization steps (remove ack reactions
and send queue-rejected statuses used by the logic in abort_helpers.go),
ensuring callers use this single helper instead of clearPendingQueue to avoid
silent drops.
In `@bridges/ai/streaming_error_handling.go`:
- Around line 53-60: The code currently falls through from the "cancelled" case
and calls state.turn.End(msgconv.MapFinishReason(reason)), which maps
"cancelled" to "other"; change this by special-casing "cancelled" so that when
reason == "cancelled" you call state.turn.End with an explicit cancellation
finish reason (do not use msgconv.MapFinishReason for that path) — use the
appropriate cancellation constant or the raw "cancelled" value as accepted by
state.turn.End; keep the existing behavior for other reasons (i.e., continue to
call msgconv.MapFinishReason(reason) in the "stop" branch).
In `@bridges/ai/subagent_registry.go`:
- Around line 46-47: The code calls drainPendingQueue(run.ChildRoomID) and
counts drained items as "stopped" but drainPendingQueue only removes queues
without per-item cleanup; instead, call finalizeStoppedQueueItems(ctx, drained)
(or the appropriate helper that performs rejection/ack cleanup) for the returned
items before incrementing stopped, and update the caller (executeUserStopPlan in
abort_helpers.go) to thread the current ctx through so finalizeStoppedQueueItems
receives the same context; ensure you reference drainPendingQueue,
finalizeStoppedQueueItems, and executeUserStopPlan when making these edits.
In `@bridges/ai/turn_data.go`:
- Around line 72-76: The cancel check is incorrectly gated by responseID, so
change the logic in the function handling turn status (using state.responseID
and state.stop.Load()) to evaluate state.stop.Load() first and return
"cancelled" immediately if non-nil, regardless of responseID; keep the
responseID trim check only for the later finishReason/status inference (e.g.,
preserve the existing responseID-based early return for in-progress detection
but after the stop check). Ensure references to state.stop.Load(),
state.responseID, and any finishReason inference logic remain unchanged except
for reordering so explicit local cancellation wins.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: f167e069-efa4-4bbb-a051-dd68b86305ba
📒 Files selected for processing (13)
bridges/ai/abort_helpers.gobridges/ai/abort_helpers_test.gobridges/ai/client.gobridges/ai/commands_parity.gobridges/ai/handlematrix.gobridges/ai/pending_queue.gobridges/ai/room_runs.gobridges/ai/streaming_error_handling.gobridges/ai/streaming_init.gobridges/ai/streaming_state.gobridges/ai/subagent_registry.gobridges/ai/turn_data.gobridges/ai/ui_message_metadata.go
📜 Review details
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: build-docker
- GitHub Check: build-agentremote-docker (amd64)
- GitHub Check: build-agentremote-docker (arm64)
- GitHub Check: Lint
- GitHub Check: build-docker
- GitHub Check: Lint
🔇 Additional comments (2)
bridges/ai/abort_helpers_test.go (1)
14-159: Good coverage of the new stop flow.These tests pin the plan-selection behavior, targeted queued removal, and the
metadata.stop/response_status == "cancelled"contract, which should make regressions in this refactor much easier to catch.bridges/ai/room_runs.go (1)
103-118: Nice stop-metadata handoff into late-bound run state.Copying
run.stopintostate.stopinbindRoomRunStatecloses the case where a user stop lands before the streaming state is attached, and caching the targeting fields here keeps later lookups cheap.Also applies to: 135-146
Pass context through pending-queue and subagent stop helpers (clearPendingQueue, stopSubagentRuns, finalizeStoppedQueueItems) and always finalize/drain pending items when clearing queues. Fix executeUserStopPlan logic to mark active stops before cancelling and fall back to a no-match when an active stop is a no-op. Ensure removePendingAckReactions is always invoked in goroutine cleanup. Adjust finishStreamingWithFailure to properly end turns on cancelled streams without falling through, and prefer explicit stop flag in canonicalResponseStatus. Add tests covering the no-op active stop fallback, cancelled finish behavior, and canonicalResponseStatus preference.
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
bridges/ai/streaming_error_handling_test.go (1)
104-106: Tighten the cancellation assertion to validate semantics, not just non-nil.At Line 104, asserting only
err != nilcan pass for unrelated errors. Consider assertingerrors.Is(err, context.Canceled)to make this test stricter.Suggested test refinement
if err == nil { t.Fatal("expected wrapped cancellation error") } + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected error wrapping context.Canceled, got %v", err) + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@bridges/ai/streaming_error_handling_test.go` around lines 104 - 106, The test currently only checks that err != nil; tighten it to assert the cancellation semantics by using errors.Is(err, context.Canceled) (or equivalent) instead of a nil check so the test validates a wrapped cancellation error; update the assertion around the err variable in the test (where err is produced) to use errors.Is(err, context.Canceled) and keep the existing failure message descriptive.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@bridges/ai/pending_queue.go`:
- Around line 96-106: The deletion of a pendingQueue from oc.pendingQueues while
producers may still hold a pointer can drop concurrent enqueues; in the delete
paths (around the block using oc.pendingQueuesMu, pendingQueues map and
queue.mu) acquire the queue.mu before removing the map entry (or otherwise
ensure the queue remains registered until no writer can hold the pointer) so
that AIClient.enqueuePendingItem() and AIClient.getPendingQueue() cannot append
to an orphaned queue; update both the delete sites (the shown block and the
similar one at 135-139) to lock queue.mu, perform the map delete while holding
that lock, then release locks in the correct order to prevent lost items.
- Around line 123-145: After removing items from queue.items, ensure
pendingQueue.lastItem is cleared if it pointed to a removed (tail) item: check
queue.lastItem != nil and if queue.lastItem.sourceEventID() == sourceEventID (or
otherwise matches an item in the removed slice) then set queue.lastItem = nil
(or assign it to the new tail from kept if you prefer preserving context); this
prevents AIClient.takePendingQueueDispatchCandidate()/snapshot.lastItem from
using a removed/stopped event as the next dispatch context.
---
Nitpick comments:
In `@bridges/ai/streaming_error_handling_test.go`:
- Around line 104-106: The test currently only checks that err != nil; tighten
it to assert the cancellation semantics by using errors.Is(err,
context.Canceled) (or equivalent) instead of a nil check so the test validates a
wrapped cancellation error; update the assertion around the err variable in the
test (where err is produced) to use errors.Is(err, context.Canceled) and keep
the existing failure message descriptive.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 1f1e0f1b-989c-4888-b423-86cefcd77750
📒 Files selected for processing (11)
bridges/ai/abort_helpers.gobridges/ai/abort_helpers_test.gobridges/ai/client.gobridges/ai/commands_parity.gobridges/ai/internal_dispatch.gobridges/ai/pending_queue.gobridges/ai/streaming_error_handling.gobridges/ai/streaming_error_handling_test.gobridges/ai/subagent_registry.gobridges/ai/turn_data.gobridges/ai/turn_data_test.go
🚧 Files skipped from review as they are similar to previous changes (7)
- bridges/ai/turn_data.go
- bridges/ai/commands_parity.go
- bridges/ai/client.go
- bridges/ai/streaming_error_handling.go
- bridges/ai/subagent_registry.go
- bridges/ai/abort_helpers.go
- bridges/ai/abort_helpers_test.go
📜 Review details
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: build-agentremote-docker (arm64)
- GitHub Check: build-agentremote-docker (amd64)
- GitHub Check: Lint
- GitHub Check: build-docker
- GitHub Check: build-docker
- GitHub Check: Lint
🔇 Additional comments (3)
bridges/ai/turn_data_test.go (1)
104-111: Good regression coverage for explicit stop precedence.This test cleanly locks in the expected
"cancelled"status when stop metadata is present without requiring a response ID.bridges/ai/streaming_error_handling_test.go (2)
8-13: Imports look appropriate and are used by the new cancellation-path test.
108-112: Nice check on terminal metadata mapping for cancelled flows.Verifying
metadata["finish_reason"] == "cancelled"is the right assertion for this pathway.
Rework pending queue locking and item housekeeping to avoid races and stale pointers. getPendingQueue now locks the queue.mu before releasing pendingQueuesMu and consistently applies settings for both new and existing queues. drainPendingQueue clears queue.items and lastItem when removing a queue. removePendingQueueBySourceEvent reassigns lastItem to the new tail if the removed item was the last. enqueuePendingItem removed a now-redundant explicit lock (the returned queue is already locked). Added a unit test to verify lastItem is cleared/reassigned, and tightened an error assertion in the streaming test to use errors.Is for wrapped cancellations.
There was a problem hiding this comment.
🧹 Nitpick comments (1)
bridges/ai/streaming_error_handling_test.go (1)
109-111: Make metadata assertion failure clearer.Consider asserting the metadata type assertion result before reading
finish_reason, so failures are more diagnostic when the snapshot shape changes.Proposed test assertion tightening
- metadata, _ := message["metadata"].(map[string]any) - if metadata["finish_reason"] != "cancelled" { - t.Fatalf("expected cancelled finish_reason, got %#v", metadata["finish_reason"]) - } + metadata, ok := message["metadata"].(map[string]any) + if !ok { + t.Fatalf("expected metadata map, got %T", message["metadata"]) + } + if metadata["finish_reason"] != "cancelled" { + t.Fatalf("expected cancelled finish_reason, got %#v", metadata["finish_reason"]) + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@bridges/ai/streaming_error_handling_test.go` around lines 109 - 111, The test currently does an unchecked type assertion for metadata then reads metadata["finish_reason"], which makes failures cryptic; update the test to assert the type assertion succeeded (capture the comma-ok like metadata, ok := message["metadata"].(map[string]any) and t.Fatalf if !ok, printing the actual type/value), then assert metadata["finish_reason"] == "cancelled" — this ensures the failure message clearly shows when the snapshot shape changes and points to the unexpected metadata value or type.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@bridges/ai/streaming_error_handling_test.go`:
- Around line 109-111: The test currently does an unchecked type assertion for
metadata then reads metadata["finish_reason"], which makes failures cryptic;
update the test to assert the type assertion succeeded (capture the comma-ok
like metadata, ok := message["metadata"].(map[string]any) and t.Fatalf if !ok,
printing the actual type/value), then assert metadata["finish_reason"] ==
"cancelled" — this ensures the failure message clearly shows when the snapshot
shape changes and points to the unexpected metadata value or type.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 681a68ff-711f-4f24-a45d-65c87579a63b
📒 Files selected for processing (3)
bridges/ai/pending_queue.gobridges/ai/queue_status_test.gobridges/ai/streaming_error_handling_test.go
✅ Files skipped from review due to trivial changes (1)
- bridges/ai/queue_status_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- bridges/ai/pending_queue.go
📜 Review details
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: build-docker
- GitHub Check: build-agentremote-docker (arm64)
- GitHub Check: build-agentremote-docker (amd64)
- GitHub Check: Lint
- GitHub Check: Lint
- GitHub Check: build-docker
🔇 Additional comments (2)
bridges/ai/streaming_error_handling_test.go (2)
8-13: Imports are correctly added and used.Both
zerologandstreamuiare used by the new cancellation-path test and keep the test setup explicit.
90-113: Good coverage for cancelled finish propagation.This test correctly validates both error wrapping (
errors.Is) and UI metadata propagation offinish_reason: "cancelled"for the cancellation path.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.