From 6f0e0a91a54f47f5e822ebf385a8e5d6a394b8e8 Mon Sep 17 00:00:00 2001 From: sirily11 <32106111+sirily11@users.noreply.github.com> Date: Tue, 26 May 2026 13:15:06 +0800 Subject: [PATCH 1/5] fix: unfreeze cross-project send target via event-driven completion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- RxCode/App/AppState+Messaging.swift | 75 ++++++++++++++++++++++++----- RxCode/App/AppState.swift | 6 +++ 2 files changed, 68 insertions(+), 13 deletions(-) diff --git a/RxCode/App/AppState+Messaging.swift b/RxCode/App/AppState+Messaging.swift index ff3b6add..eb27655d 100644 --- a/RxCode/App/AppState+Messaging.swift +++ b/RxCode/App/AppState+Messaging.swift @@ -468,42 +468,91 @@ extension AppState { // MARK: - Stream Completion (cross-project MCP) - /// Record that the stream `streamId` finished. Stored in - /// `pendingStreamCompletions` for any `awaitStreamCompletion(...)` caller - /// (currently `ide__send_to_thread`) to pick up. Latest call wins, except - /// we don't overwrite a success with an error from the fallback path. + /// Record that the stream `streamId` finished. If a caller is already + /// waiting via `awaitStreamCompletion(...)`, the result is handed to it + /// directly so it can return immediately; otherwise it's parked in + /// `pendingStreamCompletions` until someone picks it up. func recordStreamCompletion( streamId: UUID, sessionId: String, assistantText: String, error: String? ) { - pendingStreamCompletions[streamId] = StreamCompletion( + let completion = StreamCompletion( sessionId: sessionId, assistantText: assistantText, error: error ) + if let waiter = streamCompletionWaiters.removeValue(forKey: streamId) { + waiter.resume(with: completion) + return + } + pendingStreamCompletions[streamId] = completion } /// Wait up to `timeout` seconds for the stream identified by `streamId` - /// to record a completion. Polls every 100ms — MainActor serialization - /// means the recorder fires between sleeps. Returns the completion if - /// one arrived in time, otherwise `nil`. + /// to record a completion. Event-driven: `recordStreamCompletion` resumes + /// the continuation as soon as the result lands, with a parallel + /// `Task.sleep(timeout)` to surface `nil` if the deadline passes first. + /// Returns the completion if one arrived in time, otherwise `nil`. func awaitStreamCompletion(streamId: UUID, timeout: TimeInterval) async -> StreamCompletion? { - let deadline = Date().addingTimeInterval(timeout) - while Date() < deadline { + // Fast path — completion already landed before we registered. + if let completion = pendingStreamCompletions.removeValue(forKey: streamId) { + return completion + } + + 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) } - return pendingStreamCompletions.removeValue(forKey: streamId) + + timeoutTask.cancel() + return result } /// Discard a recorded completion. Called by long-running `wait_for_response=false` /// MCP sends so the dictionary doesn't grow unbounded with abandoned results. func discardStreamCompletion(streamId: UUID) { pendingStreamCompletions.removeValue(forKey: streamId) + if let waiter = streamCompletionWaiters.removeValue(forKey: streamId) { + waiter.resume(with: nil) + } } } + +/// Resumes a single waiting `awaitStreamCompletion` caller. The class wrapper +/// guards against double-resume in the race between `recordStreamCompletion` +/// and the timeout task: whichever fires first wins, and the other becomes a +/// no-op when it finds the continuation already consumed. +@MainActor +final class StreamCompletionWaiter { + private var continuation: CheckedContinuation? + + init(continuation: CheckedContinuation) { + self.continuation = continuation + } + + func resume(with value: AppState.StreamCompletion?) { + guard let cont = continuation else { return } + continuation = nil + cont.resume(returning: value) + } +} diff --git a/RxCode/App/AppState.swift b/RxCode/App/AppState.swift index 9a553115..7dd2678d 100644 --- a/RxCode/App/AppState.swift +++ b/RxCode/App/AppState.swift @@ -186,6 +186,12 @@ final class AppState { /// have not yet picked up the result. Keyed by `streamId`. var pendingStreamCompletions: [UUID: StreamCompletion] = [:] + /// Active callers waiting for a stream's completion. Resumed directly by + /// `recordStreamCompletion` so the cross-project MCP handler doesn't sit + /// in a polling loop on MainActor (which starves the target project's + /// `processStream` and freezes its thread until the sender is cancelled). + var streamCompletionWaiters: [UUID: StreamCompletionWaiter] = [:] + // MARK: - Session Summaries (shared — lightweight metadata for all projects) var allSessionSummaries: [ChatSession.Summary] = [] From 8f11029a48cabdba6c59e7d69764f5f500b0346a Mon Sep 17 00:00:00 2001 From: sirily11 <32106111+sirily11@users.noreply.github.com> Date: Tue, 26 May 2026 13:31:22 +0800 Subject: [PATCH 2/5] chore: add diagnostics for cross-project send stall 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) --- RxCode/App/AppState+CrossProject.swift | 4 +++ RxCode/Services/IDEServer/IDEMCPServer.swift | 34 +++++++++++++++++--- 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/RxCode/App/AppState+CrossProject.swift b/RxCode/App/AppState+CrossProject.swift index 3655b628..de62ea3f 100644 --- a/RxCode/App/AppState+CrossProject.swift +++ b/RxCode/App/AppState+CrossProject.swift @@ -395,11 +395,15 @@ extension AppState { var eventCount = 0 var lastEventTime = Date() + logger.info("[Stream:UI] entering for-await session=\(sessionKey, privacy: .public) stream=\(streamId) cwd=\(cwd, privacy: .public)") do { for await event in stream { eventCount += 1 let gap = Date().timeIntervalSince(lastEventTime) + if eventCount == 1 { + logger.info("[Stream:UI] first event arrived session=\(sessionKey, privacy: .public) stream=\(streamId) after=\(String(format: "%.2f", gap))s") + } lastEventTime = Date() updateState(sessionKey) { $0.lastStreamEventDate = lastEventTime } diff --git a/RxCode/Services/IDEServer/IDEMCPServer.swift b/RxCode/Services/IDEServer/IDEMCPServer.swift index 14d2db10..4d44f5e7 100644 --- a/RxCode/Services/IDEServer/IDEMCPServer.swift +++ b/RxCode/Services/IDEServer/IDEMCPServer.swift @@ -138,6 +138,9 @@ actor IDEMCPServer { } let id = UUID() allocations[sessionKey]?.connections.insert(id) + connection.stateUpdateHandler = { state in + Self.logger.info("[IDE.conn] conn=\(id.uuidString.prefix(8), privacy: .public) session=\(sessionKey, privacy: .public) state=\(String(describing: state), privacy: .public)") + } connection.start(queue: .global(qos: .userInitiated)) Self.logger.info("[IDE] accepted connection id=\(id.uuidString, privacy: .public) on port \(port) session=\(sessionKey, privacy: .public)") @@ -162,6 +165,7 @@ actor IDEMCPServer { capabilities: CapabilitySet ) async { var pendingBuffer = Data() + var chunkCount = 0 while true { // Drain any complete lines already buffered. while let newline = pendingBuffer.firstIndex(of: 0x0A) { @@ -172,20 +176,25 @@ actor IDEMCPServer { data: lineData, connection: connection, sessionKey: sessionKey, - capabilities: capabilities + capabilities: capabilities, + connectionId: connectionId ) } // Read more bytes; bail when the peer half-closes. do { let chunk = try await readChunk(connection: connection) - if chunk.isEmpty { return } + if chunk.isEmpty { + Self.logger.info("[IDE.runMCP] eof conn=\(connectionId.uuidString.prefix(8), privacy: .public) session=\(sessionKey, privacy: .public) chunks=\(chunkCount)") + return + } + chunkCount += 1 pendingBuffer.append(chunk) if pendingBuffer.count > Self.messageMaxLength { Self.logger.error("[IDE] message exceeded limit, closing connection id=\(connectionId.uuidString, privacy: .public)") return } } catch { - Self.logger.info("[IDE] connection read ended: \(error.localizedDescription, privacy: .public)") + Self.logger.info("[IDE.runMCP] err conn=\(connectionId.uuidString.prefix(8), privacy: .public) session=\(sessionKey, privacy: .public) chunks=\(chunkCount): \(error.localizedDescription, privacy: .public)") return } } @@ -213,7 +222,8 @@ actor IDEMCPServer { data: Data.SubSequence, connection: NWConnection, sessionKey: String, - capabilities: CapabilitySet + capabilities: CapabilitySet, + connectionId: UUID ) async { let bytes = Data(data) guard @@ -229,9 +239,14 @@ actor IDEMCPServer { // Notifications (no id) — we only care about `notifications/initialized` if id == nil { + Self.logger.info("[IDE.recv] conn=\(connectionId.uuidString.prefix(8), privacy: .public) session=\(sessionKey, privacy: .public) notif=\(method ?? "", privacy: .public)") return } + let connTag = connectionId.uuidString.prefix(8) + let toolName = (params["name"] as? String) ?? "" + Self.logger.info("[IDE.recv] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) method=\(method ?? "", privacy: .public) tool=\(toolName, privacy: .public)") + switch method { case "initialize": await reply( @@ -248,6 +263,7 @@ actor IDEMCPServer { ], on: connection ) + Self.logger.info("[IDE.sent] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) reply=initialize") case "tools/list": let tools = await currentTools(sessionKey: sessionKey, capabilities: capabilities) @@ -256,26 +272,36 @@ actor IDEMCPServer { result: ["tools": tools.map(Self.toolDescriptor)], on: connection ) + Self.logger.info("[IDE.sent] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) reply=tools/list count=\(tools.count)") case "tools/call": let name = params["name"] as? String ?? "" let arguments = params["arguments"] as? [String: Any] ?? [:] let argsValue = JSONValue.fromAny(arguments) + let callStart = Date() do { let handler = self.handler guard let handler else { throw IDEToolError.handlerFailed("IDE handler not attached") } + Self.logger.info("[IDE.call→] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) tool=\(name, privacy: .public)") let result = try await handler.ideHandleToolCall( name: name, arguments: argsValue, sessionKey: sessionKey ) + let elapsed = Date().timeIntervalSince(callStart) + Self.logger.info("[IDE.call←] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) tool=\(name, privacy: .public) elapsed=\(String(format: "%.1f", elapsed))s") let payload = Self.wrapToolResult(result) await reply(id: id!, result: payload, on: connection) + Self.logger.info("[IDE.sent] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) reply=tools/call tool=\(name, privacy: .public)") } catch let error as IDEToolError { + let elapsed = Date().timeIntervalSince(callStart) + Self.logger.warning("[IDE.call✗] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) tool=\(name, privacy: .public) elapsed=\(String(format: "%.1f", elapsed))s err=\(Self.errorMessage(for: error), privacy: .public)") await replyError(id: id!, code: Self.errorCode(for: error), message: Self.errorMessage(for: error), on: connection) } catch { + let elapsed = Date().timeIntervalSince(callStart) + Self.logger.warning("[IDE.call✗] conn=\(connTag, privacy: .public) session=\(sessionKey, privacy: .public) tool=\(name, privacy: .public) elapsed=\(String(format: "%.1f", elapsed))s err=\(error.localizedDescription, privacy: .public)") await replyError(id: id!, code: -32603, message: error.localizedDescription, on: connection) } From 47d5b057512902483c5455b8c4ca5d281ab0a13a Mon Sep 17 00:00:00 2001 From: sirily11 <32106111+sirily11@users.noreply.github.com> Date: Tue, 26 May 2026 13:48:15 +0800 Subject: [PATCH 3/5] fix: replace FileHandle.AsyncBytes.lines with readabilityHandler for stdout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- RxCode/Services/ClaudeService+Process.swift | 115 +++++++++++++------- 1 file changed, 78 insertions(+), 37 deletions(-) diff --git a/RxCode/Services/ClaudeService+Process.swift b/RxCode/Services/ClaudeService+Process.swift index 230f99f4..6b64d7af 100644 --- a/RxCode/Services/ClaudeService+Process.swift +++ b/RxCode/Services/ClaudeService+Process.swift @@ -99,7 +99,12 @@ extension ClaudeCodeServer { return } - // Read stdout line-by-line — ends naturally at EOF + // Read stdout line-by-line via `readabilityHandler` rather than + // `FileHandle.AsyncBytes.lines`: the AsyncBytes iterator wedges + // when a second concurrent pipe reader is active (cross-project + // send spawns a second stream while the first is mid-tool-call), + // so the second CLI's events sit in the pipe and never wake the + // for-await. Dispatch's readable source delivers reliably. var parsedCount = 0 var failedCount = 0 let decoder = JSONDecoder() @@ -107,44 +112,40 @@ extension ClaudeCodeServer { var rawLineCount = 0 var capturedSessionId: String? - do { - for try await line in stdout.fileHandleForReading.bytes.lines { - guard !line.isEmpty else { continue } - guard let data = line.data(using: .utf8) else { continue } - - rawLineCount += 1 - // Diagnostic logging of raw NDJSON — full content for first 30 lines, then type field only - if let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any] { - let type = (json["type"] as? String) ?? "?" - if rawLineCount <= 30 { - log.info("[Stream:RAW] #\(rawLineCount) type=\(type) line=\(line.prefix(600))") - } else if type == "stream_event" || rawLineCount % 50 == 0 { - log.info("[Stream:RAW] #\(rawLineCount) type=\(type)") - } - if capturedSessionId == nil, - let sid = (json["session_id"] as? String) ?? (json["sessionId"] as? String) { - capturedSessionId = sid - Task { await self.recordSessionId(streamId: streamId, sessionId: sid) } - } - } else if rawLineCount <= 30 { - log.info("[Stream:RAW] #\(rawLineCount) non-JSON line=\(line.prefix(600))") + for await line in Self.asyncLines(from: stdout.fileHandleForReading, log: log) { + guard !line.isEmpty else { continue } + guard let data = line.data(using: .utf8) else { continue } + + rawLineCount += 1 + // Diagnostic logging of raw NDJSON — full content for first 30 lines, then type field only + if let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any] { + let type = (json["type"] as? String) ?? "?" + if rawLineCount <= 30 { + log.info("[Stream:RAW] #\(rawLineCount) type=\(type) line=\(line.prefix(600))") + } else if type == "stream_event" || rawLineCount % 50 == 0 { + log.info("[Stream:RAW] #\(rawLineCount) type=\(type)") } + if capturedSessionId == nil, + let sid = (json["session_id"] as? String) ?? (json["sessionId"] as? String) { + capturedSessionId = sid + Task { await self.recordSessionId(streamId: streamId, sessionId: sid) } + } + } else if rawLineCount <= 30 { + log.info("[Stream:RAW] #\(rawLineCount) non-JSON line=\(line.prefix(600))") + } - do { - let event = try decoder.decode(StreamEvent.self, from: data) - parsedCount += 1 - continuation.yield(event) - } catch { - failedCount += 1 - // Yield raw string so partial events still reach the UI - continuation.yield(.unknown(line)) - if failedCount <= 5 { - log.warning("[Stream] parse failed #\(failedCount): \(line.prefix(200))") - } + do { + let event = try decoder.decode(StreamEvent.self, from: data) + parsedCount += 1 + continuation.yield(event) + } catch { + failedCount += 1 + // Yield raw string so partial events still reach the UI + continuation.yield(.unknown(line)) + if failedCount <= 5 { + log.warning("[Stream] parse failed #\(failedCount): \(line.prefix(200))") } } - } catch { - log.warning("[Stream] stdout read error: \(error.localizedDescription)") } log.info("[Stream] stdout ended (parsed=\(parsedCount), failed=\(failedCount))") @@ -154,13 +155,53 @@ extension ClaudeCodeServer { continuation.onTermination = { reason in log.info("[Stream] terminated (reason=\(String(describing: reason)))") task.cancel() - // Close the pipe after the stream ends to unblock the bytes.lines read. - // onTermination is called after finish(), so there is no data loss. + // Detach the readability handler first so closing the pipe doesn't + // race a pending callback dispatch, then close to release the FD. + stdout.fileHandleForReading.readabilityHandler = nil stdout.fileHandleForReading.closeFile() } } } + /// Stream lines from `handle` using a Dispatch-backed `readabilityHandler`. + /// We use this instead of `FileHandle.AsyncBytes.lines` because the async + /// iterator can wedge when multiple concurrent pipe readers exist (the + /// cross-project send case: one CLI is mid-tool-call while another is + /// just starting). Dispatch's readable source delivers each chunk via a + /// per-handle background callback that doesn't share global async state, + /// so a second simultaneous reader is unaffected by the first's progress. + private static func asyncLines(from handle: FileHandle, log: Logger) -> AsyncStream { + 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.. Date: Tue, 26 May 2026 13:57:13 +0800 Subject: [PATCH 4/5] fix: cross-project MCP send returns real thread id, not pending placeholder For a new-thread cross-project send, sendPrompt assigns a transient "pending-" 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) --- RxCode/App/AppState+CrossProject.swift | 30 ++++++++---- RxCode/App/AppState+Messaging.swift | 63 ++++++++++++++++++++++++++ RxCode/App/AppState.swift | 6 +++ 3 files changed, 91 insertions(+), 8 deletions(-) diff --git a/RxCode/App/AppState+CrossProject.swift b/RxCode/App/AppState+CrossProject.swift index de62ea3f..30480e08 100644 --- a/RxCode/App/AppState+CrossProject.swift +++ b/RxCode/App/AppState+CrossProject.swift @@ -105,9 +105,23 @@ extension AppState { } // After sendPrompt returns, window.currentSessionId is the (possibly - // pending-) key the stream is bound to. The CLI may rename it to its - // own sid mid-stream; we surface whichever id the completion lands on. - let postSendThreadId = window.currentSessionId ?? resolvedThreadId ?? "" + // pending-) key the stream is bound to. Resolve the real CLI session + // id before returning so the caller's agent never sees `pending-…` + // (which it can't use to follow up via `get_thread_messages` etc.). + let postSendKey = window.currentSessionId ?? resolvedThreadId ?? "" + let resolvedThreadIdForReturn: String + if postSendKey.hasPrefix("pending-") { + // 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 + } else { + resolvedThreadIdForReturn = postSendKey + } if !waitForResponse { // Don't leak the result in the dictionary — the caller is @@ -116,7 +130,7 @@ extension AppState { _ = await self?.awaitStreamCompletion(streamId: streamId, timeout: timeoutSeconds) } return CrossProjectSendResult( - threadId: postSendThreadId, + threadId: resolvedThreadIdForReturn, projectId: resolvedProject.id, done: false, assistantText: "", @@ -138,7 +152,7 @@ extension AppState { // the caller can decide whether to poll back via get_thread_messages. let partial = lastAssistantResponseText(in: stateForSession(window.currentSessionId ?? "").messages) return CrossProjectSendResult( - threadId: window.currentSessionId ?? postSendThreadId, + threadId: resolvedThreadIdForReturn, projectId: resolvedProject.id, done: false, assistantText: partial, @@ -422,7 +436,7 @@ extension AppState { if let state = sessionStates.removeValue(forKey: sessionKey) { sessionStates[resultEvent.sessionId] = state } - sessionIdRedirect[sessionKey] = resultEvent.sessionId + applySessionIdRedirect(from: sessionKey, to: resultEvent.sessionId) sessionKey = resultEvent.sessionId } let msgs = stateForSession(sessionKey).messages @@ -456,7 +470,7 @@ extension AppState { sessionStates[sid] = state } renameDraftState(from: previousSessionKey, to: sid, in: window) - sessionIdRedirect[previousSessionKey] = sid + applySessionIdRedirect(from: previousSessionKey, to: sid) sessionKey = sid startFlushTimer(for: sid) @@ -720,7 +734,7 @@ extension AppState { sessionStates[resultEvent.sessionId] = state } renameDraftState(from: previousSessionKey, to: resultEvent.sessionId, in: window) - sessionIdRedirect[previousSessionKey] = resultEvent.sessionId + applySessionIdRedirect(from: previousSessionKey, to: resultEvent.sessionId) sessionKey = resultEvent.sessionId if wasForeground { window.currentSessionId = resultEvent.sessionId diff --git a/RxCode/App/AppState+Messaging.swift b/RxCode/App/AppState+Messaging.swift index eb27655d..46161a15 100644 --- a/RxCode/App/AppState+Messaging.swift +++ b/RxCode/App/AppState+Messaging.swift @@ -536,6 +536,69 @@ extension AppState { } } + // MARK: - Session-id rename handoff (cross-project MCP) + + /// Record that `pendingKey` was renamed to `realSessionId` (the CLI's own + /// `session_id`). Mirror writes to `sessionIdRedirect` already; this also + /// resumes any `awaitSessionRename` caller so the cross-project send can + /// return the real thread id instead of `pending-…`. + func applySessionIdRedirect(from pendingKey: String, to realSessionId: String) { + sessionIdRedirect[pendingKey] = realSessionId + if let waiter = sessionIdRenameWaiters.removeValue(forKey: pendingKey) { + waiter.resume(with: realSessionId) + } + } + + /// Wait up to `timeout` seconds for `pendingKey` to be renamed to the + /// CLI's real `session_id`. Fast-paths the answer if the rename already + /// landed. Returns `nil` on timeout. + func awaitSessionRename(pendingKey: String, timeout: TimeInterval) async -> String? { + if let real = sessionIdRedirect[pendingKey] { + return real + } + + 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 + } + } + + timeoutTask.cancel() + return result + } + +} + +/// Resumes a single waiting `awaitSessionRename` caller. Same single-shot +/// pattern as `StreamCompletionWaiter` — guards against double-resume in the +/// race between the rename notification and the timeout task. +@MainActor +final class SessionRenameWaiter { + private var continuation: CheckedContinuation? + + init(continuation: CheckedContinuation) { + self.continuation = continuation + } + + func resume(with value: String?) { + guard let cont = continuation else { return } + continuation = nil + cont.resume(returning: value) + } } /// Resumes a single waiting `awaitStreamCompletion` caller. The class wrapper diff --git a/RxCode/App/AppState.swift b/RxCode/App/AppState.swift index 7dd2678d..af4792a7 100644 --- a/RxCode/App/AppState.swift +++ b/RxCode/App/AppState.swift @@ -172,6 +172,12 @@ final class AppState { /// id after the swap that happens mid-stream in `processStream`. var sessionIdRedirect: [String: String] = [:] + /// Callers waiting for `pending-` to be replaced by the CLI's + /// real `session_id` on the first `.system` event. The cross-project MCP + /// send (`ide__send_to_thread`) holds the JSON-RPC reply until the real + /// id is known so the sender's agent never sees a `pending-…` thread id. + var sessionIdRenameWaiters: [String: SessionRenameWaiter] = [:] + // MARK: - Stream Completion Tracking (cross-project MCP) /// Result of a finished stream. Used by `ide__send_to_thread` to surface From 9d4a837917112c207f70ca52416709939d6e2d34 Mon Sep 17 00:00:00 2001 From: sirily11 <32106111+sirily11@users.noreply.github.com> Date: Tue, 26 May 2026 14:14:48 +0800 Subject: [PATCH 5/5] fix: surface "No response requested." instead of silently dropping it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../CLISession/CLILineToBlocksMapper.swift | 2 -- RxCode/App/AppState+CrossProject.swift | 23 ------------------- RxCode/App/AppState+Messaging.swift | 1 - RxCode/App/AppState+Stream.swift | 6 +---- 4 files changed, 1 insertion(+), 31 deletions(-) diff --git a/Packages/Sources/RxCodeCore/CLISession/CLILineToBlocksMapper.swift b/Packages/Sources/RxCodeCore/CLISession/CLILineToBlocksMapper.swift index 44b8db36..08d5d3ef 100644 --- a/Packages/Sources/RxCodeCore/CLISession/CLILineToBlocksMapper.swift +++ b/Packages/Sources/RxCodeCore/CLISession/CLILineToBlocksMapper.swift @@ -129,8 +129,6 @@ public enum CLILineToBlocksMapper { switch part { case .text(let t): guard !t.isEmpty else { continue } - let trimmed = t.trimmingCharacters(in: .whitespacesAndNewlines) - if CLIMetaEnvelope.isNoResponseRequested(trimmed) { continue } blocks.append(.text(t)) case .toolUse(let id, let name, let input): blocks.append(.toolCall(ToolCall(id: id, name: name, input: input))) diff --git a/RxCode/App/AppState+CrossProject.swift b/RxCode/App/AppState+CrossProject.swift index 30480e08..46a86416 100644 --- a/RxCode/App/AppState+CrossProject.swift +++ b/RxCode/App/AppState+CrossProject.swift @@ -161,28 +161,6 @@ extension AppState { } } - /// Drop "No response requested." text blocks from the assistant message - /// at `idx`. If the message has no blocks left after the strip, remove - /// it entirely. Called at turn-finalization sites — the marker is the - /// model's response when a turn arrives without a user prompt - /// (ScheduleWakeup, hook re-entry) and reads as noise in the chat UI. - /// Strip CLI no-op meta text ("no response requested") from a message. - /// - /// `removeIfEmpty` controls whether a message left with no blocks is also - /// deleted. The normal stream path passes `true` to discard pure no-op - /// envelopes; the cancel path passes `false` so pausing a turn never makes - /// the partial assistant bubble disappear. - static func stripNoOpText(at idx: Int, in messages: inout [ChatMessage], removeIfEmpty: Bool = true) { - guard messages.indices.contains(idx) else { return } - messages[idx].blocks.removeAll { block in - guard let text = block.text else { return false } - return CLIMetaEnvelope.isNoResponseRequested(text.trimmingCharacters(in: .whitespacesAndNewlines)) - } - if removeIfEmpty, messages[idx].blocks.isEmpty { - messages.remove(at: idx) - } - } - /// Wrap a branch briefing into a system-prompt section the agent can use as /// background context. The briefing is auto-generated from earlier threads, /// so it is framed as advisory rather than authoritative. @@ -692,7 +670,6 @@ extension AppState { }) { state.messages[idx].isStreaming = false state.messages[idx].finalizeToolCalls() - Self.stripNoOpText(at: idx, in: &state.messages) } state.messages.append(ChatMessage(role: .assistant, isStreaming: true)) state.needsNewMessage = false diff --git a/RxCode/App/AppState+Messaging.swift b/RxCode/App/AppState+Messaging.swift index 46161a15..74fd6870 100644 --- a/RxCode/App/AppState+Messaging.swift +++ b/RxCode/App/AppState+Messaging.swift @@ -456,7 +456,6 @@ extension AppState { if let start = state.streamingStartDate { state.messages[idx].duration = Date().timeIntervalSince(start) } - Self.stripNoOpText(at: idx, in: &state.messages) } state.streamingStartDate = nil } diff --git a/RxCode/App/AppState+Stream.swift b/RxCode/App/AppState+Stream.swift index 2ca7433f..db77646b 100644 --- a/RxCode/App/AppState+Stream.swift +++ b/RxCode/App/AppState+Stream.swift @@ -220,7 +220,6 @@ extension AppState { // 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 { @@ -305,7 +304,6 @@ extension AppState { if let idx = state.messages.indices.reversed().first(where: { state.messages[$0].role == .assistant && state.messages[$0].isStreaming }) { state.messages[idx].isStreaming = false state.messages[idx].finalizeToolCalls() - Self.stripNoOpText(at: idx, in: &state.messages) } state.messages.append(ChatMessage(role: .assistant, isStreaming: true)) state.needsNewMessage = false @@ -444,13 +442,11 @@ extension AppState { // The user paused this turn — keep the partial assistant bubble // visible. markStreamInterrupted() clears the message's streaming // flag and retains in-progress tool calls (flagged as interrupted) - // instead of dropping them; the no-op strip below is told not to - // delete an emptied message. + // instead of dropping them. state.messages[idx].markStreamInterrupted() if let start = state.streamingStartDate { state.messages[idx].duration = Date().timeIntervalSince(start) } - Self.stripNoOpText(at: idx, in: &state.messages, removeIfEmpty: false) } state.streamingStartDate = nil state.inFlightUserAttachments = []