test: cross-project send concurrency regression suite#61
Merged
Conversation
awaitStreamCompletion was a 100ms polling loop on MainActor. When the sending project's agent invoked ide__send_to_thread, the MainActor-bound MCP handler entered that loop and starved the target project's processStream `for await event in stream` running on the same actor — the receiving thread stayed at "1 message, isStreaming=true" with no assistant tokens until the sender's tool call was cancelled. Replace the poll with an event-driven handoff: recordStreamCompletion resumes a CheckedContinuation directly the moment the result lands, with a parallel Task.sleep(timeout) as the deadline. A small @mainactor class guards against double-resume in the race between recorder and timeout. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Logs the full receiver-side handshake so we can see exactly where it hangs: every MCP method received per connection, every reply sent, tools/call latency, NWConnection state transitions, runMCP chunk count on EOF, and the moment processStream enters its for-await loop and receives its first event. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…stdout The CLI stdout reader used `for try await line in handle.bytes.lines`, whose underlying AsyncBytes iterator wedges when a second concurrent pipe reader is active — exactly the cross-project send case where the sender is mid-`ide__send_to_thread` while the target spawns its own CLI. Log evidence: the second CLI completed MCP handshake (initialize + tools/list reply observed) but emitted zero stdout to our reader, so processStream's `for await event in stream` never received the `.system init` event and the UI sat at "isStreaming=true, messages=1" indefinitely. Cancelling the sender released the AsyncBytes lock and the buffered events flushed in one burst — matching the symptom. Switch to a Dispatch-backed `readabilityHandler` pump (already used for stderr and known-good for multiple concurrent pipes). Lines are split out of an accumulating buffer; EOF flushes any trailing non-terminated line, then finishes the stream. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…holder For a new-thread cross-project send, sendPrompt assigns a transient "pending-<streamId>" key that gets replaced by the CLI's real session_id on the first `.system` event. Previously the MCP tool reply could carry the placeholder back to the caller's agent when wait_for_response=false (and on the wait_for_response=true timeout fallback path), and the caller had no usable id to follow up with later. Add a SessionRenameWaiter handoff: every site that writes to sessionIdRedirect goes through applySessionIdRedirect(from:to:), which also resumes any awaitSessionRename caller. sendCrossProject now blocks on awaitSessionRename (capped at min(timeout, 60s)) before returning the result, so the MCP reply always carries the real session_id once the CLI has produced one. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
stripNoOpText was wiping the assistant message entirely whenever the model replied with the no-op marker, on the assumption that the marker only appears for hook/ScheduleWakeup re-entries with no real prompt. That assumption breaks for ide__send_to_thread: the sender often passes context-only text (e.g. "For context, this project is RxAuthSwift…"), and the receiver model legitimately answers "No response requested." — which we then deleted, leaving the receiver thread visually empty and making the cross-project send look like the stream silently stopped. Surface the marker instead of hiding it. Removed all four live-stream strip call sites (Messaging.swift finalizeStreamSession, Stream.swift text-delta + tool-start + cancel paths, CrossProject.swift mid-stream new-message path) and the disk-replay filter in CLILineToBlocksMapper. stripNoOpText itself is deleted; isNoResponseRequested stays in case something downstream still needs the detector. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Adds four XCTests against the orchestration layer of `sendCrossProject` to prevent regression of the freeze fixed in PR #60: - `testSecondCrossProjectSendCompletesWhileFirstIsPausedMidStream`: while project A's mock stream is paused 3s mid-turn, a second `sendCrossProject` for project B must still complete (~0.5s in practice). Pre-fix the polling `awaitStreamCompletion` would have starved B's for-await loop and B would block for the full 3s. - `testCrossProjectSendReturnsRealSessionIdNotPendingPlaceholder`: the MCP reply must carry the CLI's real `session_id`, not the transient `pending-<streamId>` placeholder. Covers the `awaitSessionRename` handoff. - `testRecordStreamCompletionResumesAwaiterImmediately` and `testApplySessionIdRedirectResumesRenameWaiterImmediately`: unit-level assertions that `recordStreamCompletion` / `applySessionIdRedirect` resume their waiters within 50ms of firing — i.e. event-driven, not polled. Supporting infrastructure: - `MockAgentBackend`: actor conforming to `AgentBackend`, scripts scripted by `cwd` (FIFO per cwd) since `streamId` is generated inside `sendPrompt` and isn't visible to the test ahead of time. - Test-only seam on `AppState`: `agentBackendOverrides` dict that `backend(for:)` consults first. Production code never writes to it; the dispatch fall-through is unchanged when the dict is empty. - `setUp` snapshots/pins `selectedAgentProvider` in UserDefaults so a stale `.codex` value from a previous app run can't route past the mock into a half-initialised real backend. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
sirily11
force-pushed
the
test/cross-project-send-concurrency
branch
from
09c8265 to
984eaa5
Compare
There was a problem hiding this comment.
Pull request overview
This PR adds a regression suite to lock in the cross-project sendCrossProject concurrency fixes from #60 by introducing a scriptable test AgentBackend and an AppState injection seam, plus additional stream/MCP diagnostics.
Changes:
- Added
CrossProjectSendConcurrencyTestsplus a timeline-drivenMockAgentBackendto deterministically reproduce the prior MainActor starvation and session-id handoff issues. - Introduced continuation-based, event-driven waiters for stream completion and session-id rename, and wired cross-project sends to return the real CLI session id.
- Added additional diagnostics in the IDE MCP server and adjusted Claude stdout streaming to use a
readabilityHandler-backed line stream.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| RxCodeTests/MockAgentBackend.swift | Adds an actor-based mock AgentBackend that emits scripted StreamEvents keyed by cwd. |
| RxCodeTests/CrossProjectSendConcurrencyTests.swift | Adds 4 tests covering concurrent cross-project progress and immediate continuation-based resumption behavior. |
| RxCode/Services/IDEServer/IDEMCPServer.swift | Adds per-connection/per-request logging (connection state, recv/sent, chunk counts, tool call timing). |
| RxCode/Services/ClaudeService+Process.swift | Switches stdout streaming to a readabilityHandler-driven async line stream to avoid AsyncBytes concurrency wedging. |
| RxCode/App/AppState+Stream.swift | Removes stripNoOpText calls and adds stream diagnostic logging; session id redirect now goes through applySessionIdRedirect. |
| RxCode/App/AppState+Messaging.swift | Implements event-driven awaitStreamCompletion + new awaitSessionRename with continuation waiters and timeout tasks. |
| RxCode/App/AppState+CrossProject.swift | Makes cross-project send resolve pending session keys to real session ids before returning; adds stream logging and redirect helper usage. |
| RxCode/App/AppState+Agents.swift | Adds agentBackendOverrides lookup in backend(for:) to support test injection. |
| RxCode/App/AppState.swift | Adds sessionIdRenameWaiters, streamCompletionWaiters, and the agentBackendOverrides test seam. |
| RxCode.xcodeproj/project.pbxproj | Adds the new test files to the test target build sources. |
| Packages/Sources/RxCodeCore/CLISession/CLILineToBlocksMapper.swift | Stops filtering the “No response requested” marker from assistant text mapping. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+173
to
+196
| private static func asyncLines(from handle: FileHandle, log: Logger) -> AsyncStream<String> { | ||
| AsyncStream { continuation in | ||
| // `buffer` is touched only from the readabilityHandler, which Dispatch | ||
| // serializes onto a single internal queue per FileHandle — no lock needed. | ||
| nonisolated(unsafe) var buffer = Data() | ||
| handle.readabilityHandler = { fh in | ||
| let chunk = fh.availableData | ||
| if chunk.isEmpty { | ||
| // EOF — flush any trailing non-terminated line, then finish. | ||
| if !buffer.isEmpty, let trailing = String(data: buffer, encoding: .utf8) { | ||
| continuation.yield(trailing) | ||
| buffer.removeAll(keepingCapacity: false) | ||
| } | ||
| fh.readabilityHandler = nil | ||
| continuation.finish() | ||
| return | ||
| } | ||
| buffer.append(chunk) | ||
| while let newlineIdx = buffer.firstIndex(of: 0x0A) { | ||
| let lineData = buffer[buffer.startIndex..<newlineIdx] | ||
| buffer.removeSubrange(buffer.startIndex...newlineIdx) | ||
| if let line = String(data: lineData, encoding: .utf8) { | ||
| continuation.yield(line) | ||
| } |
Comment on lines
+503
to
523
| let timeoutTask = Task { [weak self] in | ||
| let nanos = UInt64(max(0, timeout) * 1_000_000_000) | ||
| try? await Task.sleep(nanoseconds: nanos) | ||
| guard let self else { return } | ||
| await MainActor.run { | ||
| if let waiter = self.streamCompletionWaiters.removeValue(forKey: streamId) { | ||
| waiter.resume(with: nil) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| let result: StreamCompletion? = await withCheckedContinuation { cont in | ||
| let waiter = StreamCompletionWaiter(continuation: cont) | ||
| // Re-check between the fast-path read and continuation install — the | ||
| // recorder may have fired in between if any MainActor work yielded. | ||
| if let completion = pendingStreamCompletions.removeValue(forKey: streamId) { | ||
| return completion | ||
| waiter.resume(with: completion) | ||
| } else { | ||
| streamCompletionWaiters[streamId] = waiter | ||
| } | ||
| try? await Task.sleep(nanoseconds: 100_000_000) | ||
| } |
Comment on lines
+514
to
+526
| let result: StreamCompletion? = await withCheckedContinuation { cont in | ||
| let waiter = StreamCompletionWaiter(continuation: cont) | ||
| // Re-check between the fast-path read and continuation install — the | ||
| // recorder may have fired in between if any MainActor work yielded. | ||
| if let completion = pendingStreamCompletions.removeValue(forKey: streamId) { | ||
| return completion | ||
| waiter.resume(with: completion) | ||
| } else { | ||
| streamCompletionWaiters[streamId] = waiter | ||
| } | ||
| try? await Task.sleep(nanoseconds: 100_000_000) | ||
| } | ||
| return pendingStreamCompletions.removeValue(forKey: streamId) | ||
|
|
||
| timeoutTask.cancel() | ||
| return result |
Comment on lines
+559
to
+577
| let timeoutTask = Task { [weak self] in | ||
| let nanos = UInt64(max(0, timeout) * 1_000_000_000) | ||
| try? await Task.sleep(nanoseconds: nanos) | ||
| guard let self else { return } | ||
| await MainActor.run { | ||
| if let waiter = self.sessionIdRenameWaiters.removeValue(forKey: pendingKey) { | ||
| waiter.resume(with: nil) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| let result: String? = await withCheckedContinuation { cont in | ||
| let waiter = SessionRenameWaiter(continuation: cont) | ||
| if let real = sessionIdRedirect[pendingKey] { | ||
| waiter.resume(with: real) | ||
| } else { | ||
| sessionIdRenameWaiters[pendingKey] = waiter | ||
| } | ||
| } |
Comment on lines
+570
to
+580
| let result: String? = await withCheckedContinuation { cont in | ||
| let waiter = SessionRenameWaiter(continuation: cont) | ||
| if let real = sessionIdRedirect[pendingKey] { | ||
| waiter.resume(with: real) | ||
| } else { | ||
| sessionIdRenameWaiters[pendingKey] = waiter | ||
| } | ||
| } | ||
|
|
||
| timeoutTask.cancel() | ||
| return result |
Comment on lines
+114
to
+121
| // Cap the rename wait at the request's timeout so we still honor | ||
| // the caller's deadline; 60s upper bound matches typical first-token | ||
| // latency under healthy conditions. | ||
| let renameTimeout = min(max(timeoutSeconds, 1), 60) | ||
| resolvedThreadIdForReturn = await awaitSessionRename( | ||
| pendingKey: postSendKey, | ||
| timeout: renameTimeout | ||
| ) ?? postSendKey |
Comment on lines
218
to
225
| if let idx = lastStreamingAssistantIdx() { | ||
| if state.needsNewMessage { | ||
| // New Claude turn after receiving tool result — start a new ChatMessage | ||
| state.messages[idx].isStreaming = false | ||
| state.messages[idx].finalizeToolCalls() | ||
| Self.stripNoOpText(at: idx, in: &state.messages) | ||
| state.needsNewMessage = false | ||
| state.messages.append(ChatMessage(role: .assistant, content: buffered, isStreaming: true)) | ||
| } else { |
| @@ -129,8 +129,6 @@ public enum CLILineToBlocksMapper { | |||
| switch part { | |||
| case .text(let t): | |||
| guard !t.isEmpty else { continue } | |||
Comment on lines
+480
to
490
| let completion = StreamCompletion( | ||
| sessionId: sessionId, | ||
| assistantText: assistantText, | ||
| error: error | ||
| ) | ||
| if let waiter = streamCompletionWaiters.removeValue(forKey: streamId) { | ||
| waiter.resume(with: completion) | ||
| return | ||
| } | ||
| pendingStreamCompletions[streamId] = completion | ||
| } |
The trailing `isAtBottom` assertion was racy: after releasePinnedUserMessage flips the binding true and starts an animated scroll-to-bottom, mid-animation scroll metrics briefly reset isAtBottom to false before settling. Assert the monotonic observedBottomRelease flag we're already polling for instead. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Contributor
Author
|
🎉 This PR is included in version 1.12.0 🎉 The release is available on GitHub release Your semantic-release bot 📦🚀 |
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
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.
Summary
MockAgentBackendactor (scriptable per-cwd timeline ofStreamEvents) and a tiny test-only injection seam onAppState(agentBackendOverrides).Tests added
testSecondCrossProjectSendCompletesWhileFirstIsPausedMidStream— project A's mock stream pauses 3 s mid-turn; project B'ssendCrossProjectmust still complete. Asserts ~0.5 s. Pre-fix: unfreeze cross-project send target via event-driven completion #60 the pollingawaitStreamCompletionwould starve B for the full pause.testCrossProjectSendReturnsRealSessionIdNotPendingPlaceholder— the MCP reply must carry the CLI's realsession_id, not thepending-<streamId>placeholder. Covers theawaitSessionRenamehandoff.testRecordStreamCompletionResumesAwaiterImmediately/testApplySessionIdRedirectResumesRenameWaiterImmediately— assert the continuation-based waiters resume within 50 ms (event-driven, not polled).Notes on the seam
AppState.agentBackendOverridesis empty in production;backend(for:)consults it first and falls through to the real services. The mock registers for all three providers insetUp, and the test also pinsselectedAgentProvider = .claudeCodeinUserDefaultsto defeat stale values from prior app runs (which previously caused the test to hit a real-but-uninitialisedCodexAppServer).Test plan
xcodebuild test -scheme RxCode -testPlan UnitTestPlan -only-testing:RxCodeTests/CrossProjectSendConcurrencyTestsagentBackendOverridesseam is acceptable for the test target🤖 Generated with Claude Code