diff --git a/Packages/Package.swift b/Packages/Package.swift index 4c960bcc..6651ef53 100644 --- a/Packages/Package.swift +++ b/Packages/Package.swift @@ -20,6 +20,7 @@ let package = Package( targets: [ .target( name: "MessageList", + dependencies: ["RxCodeCore"], path: "Sources/MessageList", swiftSettings: [ .defaultIsolation(MainActor.self), diff --git a/Packages/Sources/MessageList/MessageList.swift b/Packages/Sources/MessageList/MessageList.swift index 75f68ce5..2888db9e 100644 --- a/Packages/Sources/MessageList/MessageList.swift +++ b/Packages/Sources/MessageList/MessageList.swift @@ -1,40 +1,23 @@ import Foundation +import RxCodeCore import SwiftUI -import os -/// Temporary diagnostics for the "list scrolls a lot when streaming finishes" -/// report. Counts every real `proxy.scrollTo(bottomAnchor)` call and logs it -/// with a millisecond timestamp + reason so we can see how many fire and from -/// which trigger. Remove once the scroll-churn is understood. +/// Counts scroll activity without logging on the hot main-actor path. The app +/// periodically drains these counters into its performance diagnostic file. @MainActor enum ScrollToBottomDiag { - /// Counts only real `proxy.scrollTo(bottomAnchor)` calls — the answer to - /// "how many scroll-to-bottom API calls were triggered". - private static var count = 0 - /// Counts requests that were *asked for* but bailed before any scroll API - /// call (spacer absorbed the growth, or a pin owns the position). Kept on a - /// separate counter/label so it never inflates the actual-call count above. - private static var skippedCount = 0 - private static let log = Logger(subsystem: "com.claudework", category: "ScrollToBottomDiag") - private static let formatter: DateFormatter = { - let f = DateFormatter() - f.dateFormat = "HH:mm:ss.SSS" - return f - }() - - /// Log an actual `proxy.scrollTo` call. Invoke immediately before the real - /// scroll so the count stays an accurate tally of scroll API invocations. static func record(_ reason: String, animated: Bool, streaming: Bool) { - count += 1 - let ts = formatter.string(from: Date()) - log.info("[ScrollToBottom] #\(count, privacy: .public) \(ts, privacy: .public) reason=\(reason, privacy: .public) animated=\(animated, privacy: .public) streaming=\(streaming, privacy: .public)") + PerformanceDiagnostics.increment("scroll.actual.total") + PerformanceDiagnostics.increment("scroll.actual.reason.\(reason)") + if animated { PerformanceDiagnostics.increment("scroll.actual.animated") } + if streaming { PerformanceDiagnostics.increment("scroll.actual.streaming") } } - /// Log a scroll request that was requested but skipped (no scroll API call). static func recordSkipped(_ reason: String, animated: Bool, streaming: Bool) { - skippedCount += 1 - let ts = formatter.string(from: Date()) - log.info("[ScrollToBottomSkipped] #\(skippedCount, privacy: .public) \(ts, privacy: .public) reason=\(reason, privacy: .public) animated=\(animated, privacy: .public) streaming=\(streaming, privacy: .public)") + PerformanceDiagnostics.increment("scroll.skipped.total") + PerformanceDiagnostics.increment("scroll.skipped.reason.\(reason)") + if animated { PerformanceDiagnostics.increment("scroll.skipped.animated") } + if streaming { PerformanceDiagnostics.increment("scroll.skipped.streaming") } } } @@ -438,7 +421,11 @@ public struct MessageList: View { } private func scheduleScrollToBottom(proxy: ScrollViewProxy) { - guard bottomScrollTask == nil else { return } + guard bottomScrollTask == nil else { + PerformanceDiagnostics.increment("scroll.schedule.coalesced") + return + } + PerformanceDiagnostics.increment("scroll.schedule.created") let delayNanoseconds = scrollToBottomDelayNanoseconds() bottomScrollTask = Task { @MainActor in if delayNanoseconds > 0 { diff --git a/Packages/Sources/RxCodeChatKit/MessageListView.swift b/Packages/Sources/RxCodeChatKit/MessageListView.swift index bdc60625..556aa6a0 100644 --- a/Packages/Sources/RxCodeChatKit/MessageListView.swift +++ b/Packages/Sources/RxCodeChatKit/MessageListView.swift @@ -1,7 +1,6 @@ import SwiftUI import Combine import RxCodeCore -import os #if os(macOS) @@ -58,19 +57,6 @@ struct MessageListView: View { @State private var isAtBottom = true @State private var scrollRequestTask: Task? - private static let log = Logger(subsystem: "com.claudework", category: "MessageListView") - /// Temporary diagnostics: counts scroll-to-bottom *requests* (pre-coalesce) - /// so we can correlate per-token request churn with the actual scrollTo calls - /// logged in MessageList. Remove once scroll-churn is understood. - private static var scrollRequestCount = 0 - private static let diagTimestampFormatter: DateFormatter = { - let f = DateFormatter() - f.dateFormat = "HH:mm:ss.SSS" - return f - }() - private static func diagTimestamp() -> String { - diagTimestampFormatter.string(from: Date()) - } private static let bottomAnchorID = "message-list-bottom-anchor" private static let endOfScreenAnchorID = "message-list-end-of-screen" private static let userScrollDownDelta: CGFloat = 4 @@ -193,10 +179,10 @@ struct MessageListView: View { requestScrollToBottomIfAtBottom() } .onChange(of: isSessionReady) { _, new in - Self.log.info("[MessageList.ready] isSessionReady=\(new) sid=\(windowState.currentSessionId ?? "", privacy: .public) settled=\(settledItems.count)") + PerformanceDiagnostics.increment(new ? "chat.session_ready.true" : "chat.session_ready.false") } .onChange(of: settledItems.count) { _, new in - Self.log.info("[MessageList.settled] settled=\(new) sid=\(windowState.currentSessionId ?? "", privacy: .public) isSessionReady=\(isSessionReady) isLoadingFromDisk=\(chatBridge.isLoadingFromDisk)") + PerformanceDiagnostics.record("chat.settled_items", value: Double(new)) } .overlay { if settledItems.isEmpty && !chatBridge.isStreaming && windowState.currentSessionId == nil { @@ -209,8 +195,8 @@ struct MessageListView: View { // MARK: - Session lifecycle private func handleSessionTask() async { - let sid = windowState.currentSessionId ?? "" - Self.log.info("[MessageList.task] fired sid=\(sid, privacy: .public) bridgeMessages=\(chatBridge.messages.count) isStreaming=\(chatBridge.isStreaming) isLoadingFromDisk=\(chatBridge.isLoadingFromDisk)") + PerformanceDiagnostics.increment("chat.session_task.fired") + PerformanceDiagnostics.record("chat.bridge_messages", value: Double(chatBridge.messages.count)) // When the CLI emits its first `system:init` event mid-stream, AppState // swaps currentSessionId from the local "pending-..." placeholder to // the real CLI sid. That id change re-fires this task even though the @@ -220,7 +206,7 @@ struct MessageListView: View { rebuildSettledItems() if !isSessionReady { isSessionReady = true } requestScrollToBottomIfAtBottom() - Self.log.info("[MessageList.task] streaming-path settled=\(settledItems.count) sid=\(sid, privacy: .public)") + PerformanceDiagnostics.increment("chat.session_task.streaming_path") return } isSessionReady = false @@ -236,16 +222,12 @@ struct MessageListView: View { pendingIndicatorSpacerReduction = 0 activeTurnMaxMeasuredHeight = 0 rebuildSettledItems() - Self.log.info("[MessageList.task] post-rebuild settled=\(settledItems.count) sid=\(sid, privacy: .public) isLoadingFromDisk=\(chatBridge.isLoadingFromDisk)") // Empty sessions appear instantly — unless we're still loading persisted // messages from disk, in which case `handleLoadingChange` fades the list // in once messages arrive, avoiding the empty → populated "blink". guard !settledItems.isEmpty else { if !chatBridge.isLoadingFromDisk { isSessionReady = true - Self.log.info("[MessageList.task] empty + not-loading → ready sid=\(sid, privacy: .public)") - } else { - Self.log.info("[MessageList.task] empty + still-loading → waiting for disk sid=\(sid, privacy: .public)") } return } @@ -259,13 +241,11 @@ struct MessageListView: View { } private func handleLoadingChange(_ isLoading: Bool) { - let sid = windowState.currentSessionId ?? "" - Self.log.info("[MessageList.onLoadChange] isLoading=\(isLoading) sid=\(sid, privacy: .public) bridgeMessages=\(chatBridge.messages.count) settled=\(settledItems.count)") + PerformanceDiagnostics.increment(isLoading ? "chat.disk_loading.started" : "chat.disk_loading.finished") // When a background disk load finishes for a freshly switched session, // rebuild the settled list and fade in — same sequence as the .task above. guard !isLoading else { return } rebuildSettledItems() - Self.log.info("[MessageList.onLoadChange] post-rebuild settled=\(settledItems.count) sid=\(sid, privacy: .public)") readyTask?.cancel() requestScrollToBottom() anchor.resetToBottom() @@ -279,7 +259,7 @@ struct MessageListView: View { } private func handleStreamingChange(old: Bool, new: Bool) { - logScrollState("streamingChange", extra: "old=\(old) new=\(new) lastRole=\(chatBridge.messages.last?.role.rawValue ?? "")") + logScrollState("streamingChange") let wasAtBottom = isAtBottom // Only react when streaming ends — the settled list doesn't change at start. guard old && !new else { @@ -361,6 +341,7 @@ struct MessageListView: View { // MARK: - Settled Items private func rebuildSettledItems() { + PerformanceDiagnostics.increment("chat.settled_rebuild.total") let messages = settledOnlyMessages(from: chatBridge.messages) var t = Transaction() t.animation = nil @@ -372,7 +353,7 @@ struct MessageListView: View { let wasAtBottom = isAtBottom rebuildSettledItems() guard let last = chatBridge.messages.last else { return } - logScrollState("lastMessageChange", extra: "lastRole=\(last.role.rawValue) lastID=\(last.id.uuidString)") + logScrollState("lastMessageChange") if last.role != .user { requestScrollToBottomIfAtBottom(wasAtBottom) } @@ -399,9 +380,11 @@ struct MessageListView: View { // churn. If a request with the same animation intent is already in // flight, let it land instead of rebuilding it; only a differing intent // (e.g. the non-animated stream-end re-assert) supersedes it. - Self.scrollRequestCount += 1 let coalesced = scrollRequestTask != nil && scrollToBottomAnimated == animated - Self.log.info("[ScrollRequest] #\(Self.scrollRequestCount, privacy: .public) \(Self.diagTimestamp(), privacy: .public) animated=\(animated, privacy: .public) coalesced=\(coalesced, privacy: .public) streaming=\(chatBridge.isStreaming, privacy: .public)") + PerformanceDiagnostics.increment("chat.scroll_request.total") + if coalesced { PerformanceDiagnostics.increment("chat.scroll_request.coalesced") } + if chatBridge.isStreaming { PerformanceDiagnostics.increment("chat.scroll_request.streaming") } + if animated { PerformanceDiagnostics.increment("chat.scroll_request.animated") } if scrollRequestTask != nil, scrollToBottomAnimated == animated { return } @@ -537,7 +520,7 @@ struct MessageListView: View { /// drag — our own `.animating` scroll is not user-driven. private func settleAtBottom(proxy: ScrollViewProxy, reason: String) { settleScrollTask?.cancel() - Self.log.info("[ScrollSettle] reason=\(reason, privacy: .public) sid=\(windowState.currentSessionId ?? "", privacy: .public) settled=\(settledItems.count)") + PerformanceDiagnostics.increment("chat.scroll_settle.\(reason)") settleScrollTask = Task { @MainActor in for _ in 0..<12 { guard !Task.isCancelled, !isUserDrivenScroll else { return } @@ -554,12 +537,12 @@ struct MessageListView: View { settleScrollTask?.cancel() pinToTopTask?.cancel() canReleasePinnedTurnByScroll = false - logScrollState("pinTop.scheduled", extra: "target=\(id.uuidString) animated=\(animated)") + logScrollState("pinTop.scheduled") pinToTopTask = Task { @MainActor in try? await Task.sleep(for: .milliseconds(16)) if animated { guard !Task.isCancelled else { return } - logScrollState("pinTop.animated", extra: "target=\(id.uuidString)") + logScrollState("pinTop.animated") withAnimation(.easeInOut(duration: Self.pinToTopAnimationSeconds)) { proxy.scrollTo(id, anchor: .top) } @@ -568,7 +551,7 @@ struct MessageListView: View { for attempt in 0..<8 { guard !Task.isCancelled else { return } if attempt == 0 || attempt == 7 { - logScrollState("pinTop.settle", extra: "target=\(id.uuidString) attempt=\(attempt)") + logScrollState("pinTop.settle") } var transaction = Transaction() transaction.animation = nil @@ -579,7 +562,7 @@ struct MessageListView: View { } guard !Task.isCancelled, isPinningLatestTurnToTop else { return } canReleasePinnedTurnByScroll = true - logScrollState("pinTop.armedRelease", extra: "target=\(id.uuidString)") + logScrollState("pinTop.armedRelease") } } @@ -611,7 +594,7 @@ struct MessageListView: View { transaction.animation = nil withTransaction(transaction) { latestUserMinY = value } updateActiveTurnMaxMeasuredHeight() - logScrollState("latestUserMinY.updated", extra: "value=\(value)") + logScrollState("latestUserMinY.updated") } /// `minY` of the tail spacer — its distance from the user message is the @@ -628,7 +611,7 @@ struct MessageListView: View { } } updateActiveTurnMaxMeasuredHeight() - logScrollState("tailSpacerMinY.updated", extra: "value=\(value) old=\(oldValue)") + logScrollState("tailSpacerMinY.updated") } private var rawActiveTurnMeasuredHeight: CGFloat { @@ -652,15 +635,8 @@ struct MessageListView: View { } } - private func logScrollState(_ label: String, extra: String = "") { - let active = activeTurnUserMessageID?.uuidString ?? "" - let rawTurnHeight = rawActiveTurnMeasuredHeight - let latestTurnHeight = max(activeTurnMaxMeasuredHeight, rawTurnHeight) - let extraSpacer = pinTailSpacerExtraHeight - let suffix = extra.isEmpty ? "" : " \(extra)" - Self.log.info( - "[ScrollPin] \(label, privacy: .public) sid=\(windowState.currentSessionId ?? "", privacy: .public) active=\(active, privacy: .public) pinning=\(isPinningLatestTurnToTop) releaseArmed=\(canReleasePinnedTurnByScroll) userDriven=\(isUserDrivenScroll) directUser=\(isDirectUserScroll) stream=\(chatBridge.isStreaming) scrollH=\(Double(scrollViewHeight), privacy: .public) userY=\(Double(latestUserMinY), privacy: .public) tailY=\(Double(tailSpacerMinY), privacy: .public) rawTurnH=\(Double(rawTurnHeight), privacy: .public) turnH=\(Double(latestTurnHeight), privacy: .public) minTail=\(Double(minTailSpacer), privacy: .public) extraSpacer=\(Double(extraSpacer), privacy: .public) pendingReduce=\(Double(pendingIndicatorSpacerReduction), privacy: .public)\(suffix, privacy: .public)" - ) + private func logScrollState(_ label: String) { + PerformanceDiagnostics.increment("chat.scroll_state.\(label)") } } diff --git a/Packages/Sources/RxCodeCore/Models/StreamEvent.swift b/Packages/Sources/RxCodeCore/Models/StreamEvent.swift index 230e7509..75ae817d 100644 --- a/Packages/Sources/RxCodeCore/Models/StreamEvent.swift +++ b/Packages/Sources/RxCodeCore/Models/StreamEvent.swift @@ -36,13 +36,24 @@ public struct SystemEvent: Sendable { public let tools: [String]? public let model: String? public let claudeCodeVersion: String? - - public init(subtype: String, sessionId: String?, tools: [String]?, model: String?, claudeCodeVersion: String?) { + /// Background-task id for `task_started` / `task_updated` / `task_notification` + /// system events. Recent Claude Code runs long tasks (background shells, + /// subagents) as "backend agents" that outlive the turn that spawned them. + public let taskId: String? + /// Normalized status for a background-task event: the top-level `status` + /// (task_notification) or `patch.status` (task_updated). Nil for + /// `task_started` (which has no status and means "running"). + public let taskStatus: String? + + public init(subtype: String, sessionId: String?, tools: [String]?, model: String?, + claudeCodeVersion: String?, taskId: String? = nil, taskStatus: String? = nil) { self.subtype = subtype self.sessionId = sessionId self.tools = tools self.model = model self.claudeCodeVersion = claudeCodeVersion + self.taskId = taskId + self.taskStatus = taskStatus } } @@ -134,9 +145,21 @@ public struct ResultEvent: Sendable { public let totalTurns: Int? public let usage: UsageInfo? public let contextWindow: ContextWindowInfo? + /// `origin.kind` from the CLI's `result` message. Recent Claude Code emits + /// autonomous follow-up results for background/"backend agent" work tagged + /// `origin.kind == "task-notification"`. A user-driven turn's own result has + /// no origin (or a non-task-notification kind). See `isTaskNotification`. + public let originKind: String? + + /// True when this result is an autonomous background follow-up rather than + /// the end of the user's turn. Such results must NOT tear down the turn + /// (keep streaming, keep the process alive). Mirrors `claude-agent-acp`'s + /// `message.origin?.kind === "task-notification"` gate. + public var isTaskNotification: Bool { originKind == "task-notification" } public init(durationMs: Double?, totalCostUsd: Double?, sessionId: String, isError: Bool, - totalTurns: Int?, usage: UsageInfo?, contextWindow: ContextWindowInfo?) { + totalTurns: Int?, usage: UsageInfo?, contextWindow: ContextWindowInfo?, + originKind: String? = nil) { self.durationMs = durationMs self.totalCostUsd = totalCostUsd self.sessionId = sessionId @@ -144,6 +167,7 @@ public struct ResultEvent: Sendable { self.totalTurns = totalTurns self.usage = usage self.contextWindow = contextWindow + self.originKind = originKind } } diff --git a/Packages/Sources/RxCodeCore/Models/StreamEventDecoding.swift b/Packages/Sources/RxCodeCore/Models/StreamEventDecoding.swift index 8f5a68f0..21b25f55 100644 --- a/Packages/Sources/RxCodeCore/Models/StreamEventDecoding.swift +++ b/Packages/Sources/RxCodeCore/Models/StreamEventDecoding.swift @@ -44,6 +44,13 @@ extension SystemEvent: Decodable { case tools case model case claudeCodeVersion = "claude_code_version" + case taskId = "task_id" + case status + case patch + } + + private enum PatchCodingKeys: String, CodingKey { + case status } public init(from decoder: Decoder) throws { @@ -53,6 +60,16 @@ extension SystemEvent: Decodable { self.tools = try container.decodeIfPresent([String].self, forKey: .tools) self.model = try container.decodeIfPresent(String.self, forKey: .model) self.claudeCodeVersion = try container.decodeIfPresent(String.self, forKey: .claudeCodeVersion) + self.taskId = try container.decodeIfPresent(String.self, forKey: .taskId) + // Background-task status lives at the top level (task_notification) or + // nested under `patch` (task_updated). task_started carries neither. + if let status = try container.decodeIfPresent(String.self, forKey: .status) { + self.taskStatus = status + } else if let patch = try? container.nestedContainer(keyedBy: PatchCodingKeys.self, forKey: .patch) { + self.taskStatus = try patch.decodeIfPresent(String.self, forKey: .status) + } else { + self.taskStatus = nil + } } } @@ -211,6 +228,11 @@ extension ResultEvent: Decodable { case totalTurns = "total_turns" case usage case contextWindow = "context_window" + case origin + } + + private enum OriginCodingKeys: String, CodingKey { + case kind } public init(from decoder: Decoder) throws { @@ -222,6 +244,13 @@ extension ResultEvent: Decodable { self.totalTurns = try container.decodeIfPresent(Int.self, forKey: .totalTurns) self.usage = try container.decodeIfPresent(UsageInfo.self, forKey: .usage) self.contextWindow = try container.decodeIfPresent(ContextWindowInfo.self, forKey: .contextWindow) + // Recent Claude Code tags autonomous background follow-up results with + // `origin: { "kind": "task-notification" }`. Absent on user-turn results. + if let originContainer = try? container.nestedContainer(keyedBy: OriginCodingKeys.self, forKey: .origin) { + self.originKind = try originContainer.decodeIfPresent(String.self, forKey: .kind) + } else { + self.originKind = nil + } } } diff --git a/Packages/Sources/RxCodeCore/Utilities/PerformanceDiagnostics.swift b/Packages/Sources/RxCodeCore/Utilities/PerformanceDiagnostics.swift new file mode 100644 index 00000000..5f50f627 --- /dev/null +++ b/Packages/Sources/RxCodeCore/Utilities/PerformanceDiagnostics.swift @@ -0,0 +1,92 @@ +import Foundation + +/// Low-overhead, process-wide counters for performance-sensitive paths. +/// +/// Hot UI code records only numeric values under fixed keys. The app drains the +/// aggregate periodically and writes a single bounded diagnostic record, which +/// avoids doing file I/O or constructing detailed log strings while scrolling +/// and streaming. +public enum PerformanceDiagnostics { + public struct Measurement: Codable, Sendable, Equatable { + public let count: Int64 + public let total: Double + public let maximum: Double + + fileprivate init(count: Int64, total: Double, maximum: Double) { + self.count = count + self.total = total + self.maximum = maximum + } + } + + public struct Snapshot: Codable, Sendable, Equatable { + public let counters: [String: Int64] + public let measurements: [String: Measurement] + + fileprivate init( + counters: [String: Int64], + measurements: [String: Measurement] + ) { + self.counters = counters + self.measurements = measurements + } + } + + private final class Accumulator: @unchecked Sendable { + private struct MutableMeasurement { + var count: Int64 = 0 + var total: Double = 0 + var maximum: Double = 0 + } + + private let lock = NSLock() + private var counters: [String: Int64] = [:] + private var measurements: [String: MutableMeasurement] = [:] + + func increment(_ key: String, by amount: Int64) { + lock.lock() + counters[key, default: 0] += amount + lock.unlock() + } + + func record(_ key: String, value: Double) { + lock.lock() + var measurement = measurements[key, default: MutableMeasurement()] + measurement.count += 1 + measurement.total += value + measurement.maximum = max(measurement.maximum, value) + measurements[key] = measurement + lock.unlock() + } + + func drain() -> Snapshot { + lock.lock() + let counterSnapshot = counters + let measurementSnapshot = measurements.mapValues { + Measurement(count: $0.count, total: $0.total, maximum: $0.maximum) + } + counters.removeAll(keepingCapacity: true) + measurements.removeAll(keepingCapacity: true) + lock.unlock() + return Snapshot(counters: counterSnapshot, measurements: measurementSnapshot) + } + } + + private static let accumulator = Accumulator() + + public static func increment(_ key: String, by amount: Int64 = 1) { + accumulator.increment(key, by: amount) + } + + /// Records a numeric distribution. The snapshot keeps count, total, and + /// maximum, which is enough to calculate an interval average without + /// retaining individual events. + public static func record(_ key: String, value: Double) { + accumulator.record(key, value: value) + } + + /// Returns and resets the current interval's aggregate values. + public static func drain() -> Snapshot { + accumulator.drain() + } +} diff --git a/Packages/Sources/RxCodeMarkdown/MarkdownView.swift b/Packages/Sources/RxCodeMarkdown/MarkdownView.swift index f1850aae..9615c413 100644 --- a/Packages/Sources/RxCodeMarkdown/MarkdownView.swift +++ b/Packages/Sources/RxCodeMarkdown/MarkdownView.swift @@ -112,16 +112,34 @@ final class MarkdownParseCache: @unchecked Sendable { func document(for text: String, cacheable: Bool) -> MarkdownDocument { let cost = text.utf8.count guard cacheable, cost <= maxCacheableTextBytes else { - return MarkdownDocumentParser.parse(text) + PerformanceDiagnostics.increment( + cacheable ? "markdown.cache.skipped_large" : "markdown.cache.skipped_live" + ) + return parseMeasured(text, cost: cost) } let key = text as NSString if let box = cache.object(forKey: key) { + PerformanceDiagnostics.increment("markdown.cache.hit") return box.document } - let parsed = MarkdownDocumentParser.parse(text) + PerformanceDiagnostics.increment("markdown.cache.miss") + let parsed = parseMeasured(text, cost: cost) cache.setObject(Box(parsed), forKey: key, cost: cost) return parsed } + + private func parseMeasured(_ text: String, cost: Int) -> MarkdownDocument { + let clock = ContinuousClock() + let start = clock.now + let parsed = MarkdownDocumentParser.parse(text) + let elapsed = start.duration(to: clock.now).components + let milliseconds = Double(elapsed.seconds) * 1_000 + + Double(elapsed.attoseconds) / 1_000_000_000_000_000 + PerformanceDiagnostics.increment("markdown.parse.total") + PerformanceDiagnostics.record("markdown.parse.input_bytes", value: Double(cost)) + PerformanceDiagnostics.record("markdown.parse.duration_ms", value: milliseconds) + return parsed + } } public struct MarkdownView: View { diff --git a/Packages/Tests/RxCodeCoreTests/StreamEventResultOriginTests.swift b/Packages/Tests/RxCodeCoreTests/StreamEventResultOriginTests.swift new file mode 100644 index 00000000..232db647 --- /dev/null +++ b/Packages/Tests/RxCodeCoreTests/StreamEventResultOriginTests.swift @@ -0,0 +1,107 @@ +import Foundation +import Testing +@testable import RxCodeCore + +@Suite("ResultEvent origin decoding") +struct StreamEventResultOriginTests { + + private func decode(_ json: String) throws -> StreamEvent { + try JSONDecoder().decode(StreamEvent.self, from: Data(json.utf8)) + } + + @Test("result with origin.kind == task-notification is a background follow-up") + func taskNotificationResult() throws { + let json = """ + {"type":"result","subtype":"success","session_id":"s-1","is_error":false,"origin":{"kind":"task-notification"}} + """ + guard case let .result(event) = try decode(json) else { + Issue.record("expected .result") + return + } + #expect(event.originKind == "task-notification") + #expect(event.isTaskNotification == true) + } + + @Test("result without origin is a normal end-of-turn result") + func userTurnResult() throws { + let json = """ + {"type":"result","subtype":"success","session_id":"s-1","is_error":false} + """ + guard case let .result(event) = try decode(json) else { + Issue.record("expected .result") + return + } + #expect(event.originKind == nil) + #expect(event.isTaskNotification == false) + } + + @Test("result with a non-task-notification origin kind still ends the turn") + func otherOriginResult() throws { + let json = """ + {"type":"result","subtype":"success","session_id":"s-1","is_error":false,"origin":{"kind":"user"}} + """ + guard case let .result(event) = try decode(json) else { + Issue.record("expected .result") + return + } + #expect(event.originKind == "user") + #expect(event.isTaskNotification == false) + } + + // MARK: - Background-task system events + + @Test("task_started exposes task_id with no status") + func taskStarted() throws { + let json = """ + {"type":"system","subtype":"task_started","task_id":"ba8pw5l47","task_type":"local_bash","session_id":"s-1"} + """ + guard case let .system(event) = try decode(json) else { + Issue.record("expected .system") + return + } + #expect(event.subtype == "task_started") + #expect(event.taskId == "ba8pw5l47") + #expect(event.taskStatus == nil) + } + + @Test("task_updated reads status from the nested patch") + func taskUpdatedPatchStatus() throws { + let json = """ + {"type":"system","subtype":"task_updated","task_id":"ba8pw5l47","patch":{"status":"completed","end_time":1783678637357},"session_id":"s-1"} + """ + guard case let .system(event) = try decode(json) else { + Issue.record("expected .system") + return + } + #expect(event.taskId == "ba8pw5l47") + #expect(event.taskStatus == "completed") + } + + @Test("task_notification reads top-level status") + func taskNotificationStatus() throws { + let json = """ + {"type":"system","subtype":"task_notification","task_id":"ba8pw5l47","status":"completed","summary":"done","session_id":"s-1"} + """ + guard case let .system(event) = try decode(json) else { + Issue.record("expected .system") + return + } + #expect(event.taskId == "ba8pw5l47") + #expect(event.taskStatus == "completed") + } + + @Test("plain status system event carries no task id") + func plainStatusEvent() throws { + let json = """ + {"type":"system","subtype":"status","status":"requesting","session_id":"s-1"} + """ + guard case let .system(event) = try decode(json) else { + Issue.record("expected .system") + return + } + #expect(event.taskId == nil) + // taskStatus decodes the top-level `status`, but with no taskId the + // background-task tracker ignores it. + #expect(event.taskStatus == "requesting") + } +} diff --git a/RxCode.xcodeproj/xcshareddata/xcschemes/RxCode.xcscheme b/RxCode.xcodeproj/xcshareddata/xcschemes/RxCode.xcscheme index 658aa4be..1584413b 100644 --- a/RxCode.xcodeproj/xcshareddata/xcschemes/RxCode.xcscheme +++ b/RxCode.xcodeproj/xcshareddata/xcschemes/RxCode.xcscheme @@ -1,7 +1,7 @@ + version = "1.7"> ? = nil - - func appendExtraSystemPrompt(_ context: String) { - let trimmed = context.trimmingCharacters(in: .whitespacesAndNewlines) - guard !trimmed.isEmpty else { return } - if let existing = extraSystemPrompt, !existing.isEmpty { - extraSystemPrompt = "\(existing)\n\n\(trimmed)" - } else { - extraSystemPrompt = trimmed - } - } - - // Kick off the expensive, independent pre-spawn work concurrently. - // Memory lookup, git branch + briefing, IDE-MCP port allocation, and - // skill-context resolution all hop off MainActor and previously ran - // stacked serially — pushing the first stream event well after the - // user pressed send. Each `async let` starts immediately and is only - // joined when its value is read below. - // - // Snapshot the MainActor flags into locals first: `async let` evaluates - // the right-hand side in a nonisolated autoclosure, so it can't read - // `memoryEnabled` / `memoryInjectEnabled` / `memoryMaxContextItems` - // directly. The local `let` capture solves both isolation and the - // "captured var" warning for `sessionKey`. - let memoryActive = memoryEnabled && memoryInjectEnabled - let memoryLimit = memoryMaxContextItems - let memoryMode = memoryRetrievalMode - let memoryMinScore = memoryInjectionScoreThreshold - let capturedSessionKey = sessionKey - let memoryService = self.memoryService - let marketplace = self.marketplace - let ideMCPServer = self.ideMCPServer - let shouldIncludeIDEMCP = includeIDEMCP - - func logPreflight(_ label: String, detail: String = "") { - let elapsed = Date().timeIntervalSince(streamStart) - if detail.isEmpty { - logger.info("[Stream:UI] preflight \(label, privacy: .public) stream=\(streamId) after=\(String(format: "%.2f", elapsed), privacy: .public)s") - } else { - logger.info("[Stream:UI] preflight \(label, privacy: .public) stream=\(streamId) after=\(String(format: "%.2f", elapsed), privacy: .public)s \(detail, privacy: .public)") - } - } - - async let memoryHitsAsync = memoryActive - ? await memoryService.search( - prompt, - projectId: projectId, - limit: memoryLimit, - minScore: memoryMinScore - ) - : [] - async let currentBranchAsync = GitHelper.currentBranch(at: cwd) - async let idePortAsync: UInt16? = shouldIncludeIDEMCP - ? ideMCPServer.allocate( - sessionKey: capturedSessionKey, - capabilities: agentProvider.staticCapabilities - ) - : nil - async let skillContextAsync: String? = marketplace.promptContext(for: agentProvider) - async let codexSkillOverridesAsync: [String] = shouldIncludeIDEMCP && agentProvider == .codex - ? await marketplace.codexConfigOverrides() - : [] - - let resolvedMemoryContext: String - let memoryHitCount: Int - if memoryActive { - let hits = await memoryHitsAsync - memoryHitCount = hits.count - resolvedMemoryContext = memoryContextSystemPrompt(relatedHits: hits) - } else { - memoryHitCount = 0 - resolvedMemoryContext = "" - } - logPreflight( - "memory", - detail: "enabled=\(memoryActive) mode=\(memoryMode.title) minScore=\(String(format: "%.2f", memoryMinScore)) hits=\(memoryHitCount) contextChars=\(resolvedMemoryContext.count)" - ) - - let branchBriefingContext: String - if let branch = await currentBranchAsync, - let briefing = threadStore.branchBriefingItem(projectId: projectId, branch: branch) { - branchBriefingContext = Self.branchBriefingSystemPrompt( - branch: branch, - briefing: briefing.briefing - ) - logPreflight("branchBriefing", detail: "branch=\(branch) contextChars=\(branchBriefingContext.count)") - } else { - branchBriefingContext = "" - logPreflight("branchBriefing", detail: "contextChars=0") - } - - // The IDE-MCP port is provider-agnostic at allocation time — the - // bridge command is built from the port. Per-backend MCP config - // writes still happen serially after this since they consume the - // bridge, but they no longer block memory/git/skill resolution. - let idePort = await idePortAsync - let bridge = idePort.map { IDEMCPServer.bridgeCommand(forPort: $0) } - logPreflight( - "ideMCP", - detail: "enabled=\(shouldIncludeIDEMCP) port=\(idePort.map(String.init) ?? "")" + // spec, model split, background context) concurrently before dispatching + // through the unified protocol. See `resolveStreamPreflight`. + let preflight = await resolveStreamPreflight( + streamId: streamId, + prompt: prompt, + cwd: cwd, + cliSessionId: cliSessionId, + sessionKey: sessionKey, + agentProvider: agentProvider, + model: model, + permissionMode: permissionMode, + registerMode: registerMode, + projectId: projectId, + includeIDEMCP: includeIDEMCP, + streamStart: streamStart ) - // Session-start hooks fire once, only for a brand-new thread (no resumed - // CLI session). Their stdout is injected into this turn's agent context - // the same way the branch briefing / memory context is, and the status - // cards inserted by UserAddedHook persist via the `.result` save. - // - // Project-new-chat hooks update passive banners and are driven by the - // visible chat views. Keep them out of this preflight path: awaiting - // network-backed banner checks here delays the first backend event. - var hookStartContext = "" - if cliSessionId == nil, let project = projects.first(where: { $0.id == projectId }) { - hookStartContext = await hookManager.dispatchSessionStart( - SessionStartPayload(project: project, sessionKey: sessionKey) - ).combinedOutput - logPreflight("hooksStart", detail: "contextChars=\(hookStartContext.count)") - } - - switch agentProvider { - case .claudeCode: - mcpClaudeConfigPath = await mcp.writeClaudeConfig(projectPath: cwd, bridgeCommand: bridge) - logPreflight("claudeMCP", detail: "hasConfig=\(mcpClaudeConfigPath != nil)") - // Surface the accumulated briefing for the project's current branch - // to the agent as background context via `--append-system-prompt`. - appendExtraSystemPrompt(branchBriefingContext) - appendExtraSystemPrompt(resolvedMemoryContext) - appendExtraSystemPrompt(hookStartContext) - if let skillContext = await skillContextAsync { - logPreflight("skillContext", detail: "chars=\(skillContext.count)") - appendExtraSystemPrompt(skillContext) - } else { - logPreflight("skillContext", detail: "chars=0") - } - case .codex: - mcpCodexOverrides = await mcp.codexConfigOverrides( - projectPath: cwd, - bridgeCommand: bridge, - disableAllServers: !shouldIncludeIDEMCP - ) - if !shouldIncludeIDEMCP { - mcpCodexOverrides += ["--disable", "plugins"] - } - logPreflight( - "codexMCP", - detail: "args=\(mcpCodexOverrides.count) disabledAll=\(!shouldIncludeIDEMCP)" - ) - let codexSkillOverrides = await codexSkillOverridesAsync - logPreflight("codexSkillOverrides", detail: "args=\(codexSkillOverrides.count)") - mcpCodexOverrides += codexSkillOverrides - resolvedPrompt = Self.promptWithBackgroundContext( - [branchBriefingContext, resolvedMemoryContext, hookStartContext], - prompt: resolvedPrompt - ) - if let skillContext = await skillContextAsync { - logPreflight("skillContext", detail: "chars=\(skillContext.count)") - resolvedPrompt = "\(skillContext)\n\nUser request:\n\(resolvedPrompt)" - } else { - logPreflight("skillContext", detail: "chars=0") - } - resolvedSendMode = registerMode - case .acp: - acpMCPServers = await mcp.acpMCPServers( - projectPath: cwd, - bridgeCommand: bridge - ) - logPreflight("acpMCP", detail: "servers=\(acpMCPServers.count)") - resolvedPrompt = Self.promptWithBackgroundContext( - [branchBriefingContext, resolvedMemoryContext, hookStartContext], - prompt: resolvedPrompt - ) - if let skillContext = await skillContextAsync { - logPreflight("skillContext", detail: "chars=\(skillContext.count)") - resolvedPrompt = "\(skillContext)\n\nUser request:\n\(resolvedPrompt)" - } else { - logPreflight("skillContext", detail: "chars=0") - } - // `model` may be a composite `::` key (from the picker) - // or a bare model id (from a per-session override). - let split = acpSelectionParts(for: model) - let resolvedClientId = split?.clientId - ?? sessionStates[sessionKey]?.acpClientId - ?? selectedACPClientId - resolvedModel = split?.model ?? model - resolvedSendMode = registerMode - if let spec = acpClients.first(where: { $0.id == resolvedClientId && $0.enabled }) { - acpSpec = spec - } else { - logger.error("[ACP] no enabled client for id=\(resolvedClientId, privacy: .public)") - earlyStream = AsyncStream { c in - c.yield(.user(UserMessage( - toolUseId: nil, - content: "No ACP client configured. Add one in Settings → ACP Clients.", - isError: true - ))) - c.yield(.result(ResultEvent( - durationMs: nil, totalCostUsd: nil, - sessionId: cliSessionId ?? sessionKey, - isError: true, totalTurns: nil, usage: nil, contextWindow: nil - ))) - c.finish() - } - } - } - let stream: AsyncStream - if let earlyStream { + if let earlyStream = preflight.earlyStream { stream = earlyStream } else { let preflightElapsed = Date().timeIntervalSince(streamStart) @@ -260,19 +64,19 @@ extension AppState { } let request = BackendSendRequest( streamId: streamId, - prompt: resolvedPrompt, + prompt: preflight.resolvedPrompt, cwd: cwd, sessionId: cliSessionId, - model: resolvedModel, + model: preflight.resolvedModel, effort: effort, - permissionMode: resolvedSendMode, + permissionMode: preflight.resolvedSendMode, planMode: permissionMode == .plan, hookSettingsPath: hookSettingsPath, - mcpClaudeConfigPath: mcpClaudeConfigPath, - extraSystemPrompt: extraSystemPrompt, - mcpCodexOverrides: mcpCodexOverrides, - acpMCPServers: acpMCPServers, - acpSpec: acpSpec, + mcpClaudeConfigPath: preflight.mcpClaudeConfigPath, + extraSystemPrompt: preflight.extraSystemPrompt, + mcpCodexOverrides: preflight.mcpCodexOverrides, + acpMCPServers: preflight.acpMCPServers, + acpSpec: preflight.acpSpec, clientSessionKey: sessionKey ) stream = await backend(for: agentProvider).send(request) @@ -351,6 +155,28 @@ extension AppState { if let model = systemEvent.model { updateState(sessionKey) { $0.activeModelName = model } } + + // Track background "backend agent" tasks so a `result` that + // merely yields the turn while such a task is still running + // doesn't tear the turn down (see the `.result` handler). + if let taskId = systemEvent.taskId { + let terminalStatuses: Set = ["completed", "killed", "failed", "stopped", "canceled", "cancelled", "timeout", "error"] + switch systemEvent.subtype { + case "task_started": + updateState(sessionKey) { $0.liveBackgroundTaskIds.insert(taskId) } + logger.info("[Stream:UI] background task started id=\(taskId, privacy: .public) (live=\(self.stateForSession(sessionKey).liveBackgroundTaskIds.count))") + case "task_notification": + // A notification is always the task's terminal wake signal. + updateState(sessionKey) { $0.liveBackgroundTaskIds.remove(taskId) } + logger.info("[Stream:UI] background task notified id=\(taskId, privacy: .public) status=\(systemEvent.taskStatus ?? "?", privacy: .public) (live=\(self.stateForSession(sessionKey).liveBackgroundTaskIds.count))") + case "task_updated": + if let status = systemEvent.taskStatus, terminalStatuses.contains(status) { + updateState(sessionKey) { $0.liveBackgroundTaskIds.remove(taskId) } + } + default: + break + } + } // Hook events (SessionStart, PreToolUse, etc.) carry the parent's session_id, // not this subprocess's. Acting on them flips currentSessionId mid-stream and // triggers MessageListView's fade-out/in — visible as a blink. @@ -640,6 +466,34 @@ extension AppState { logger.info("\(debugLogPrefix, privacy: .public) phase=resultEvent stream=\(streamId) eventCount=\(eventCount, privacy: .public) total=\(String(format: "%.2f", totalElapsed), privacy: .public)s gap=\(String(format: "%.1f", gap), privacy: .public)s isError=\(resultEvent.isError, privacy: .public) session=\(resultEvent.sessionId, privacy: .public)") } + // Recent Claude Code runs long tasks (background shells, subagents) as + // "backend agents". The turn that spawns one ends with a normal `result` + // (`origin == null`, `stop_reason == end_turn`) WHILE the task is still + // running; when it finishes the CLI autonomously runs a follow-up turn + // (a fresh `init` + a `result` with `origin.kind == "task-notification"`). + // If we tore the turn down on that yield `result` we'd close stdin — which + // kills the still-running CLI/task and loses the follow-up — and flip the + // UI to idle. So while any background task is still live, keep the process + // alive and the UI "in progress"; only fold in this result's (real) cost. + // The follow-up result arrives once tasks drain (set now empty) and takes + // the normal end-of-turn path below. + if !stateForSession(sessionKey).liveBackgroundTaskIds.isEmpty { + let live = stateForSession(sessionKey).liveBackgroundTaskIds.count + logger.info("[Stream:UI] event #\(eventCount) .result yields with \(live) live background task(s); keeping turn in progress (origin=\(resultEvent.originKind ?? "none", privacy: .public))") + updateState(sessionKey) { state in + if let cost = resultEvent.totalCostUsd { state.costUsd = cost } + if let duration = resultEvent.durationMs { state.durationMs += duration } + if let turns = resultEvent.totalTurns { state.turns += turns } + if let usage = resultEvent.usage { + state.inputTokens += usage.inputTokens + state.outputTokens += usage.outputTokens + state.cacheCreationTokens += usage.cacheCreationInputTokens + state.cacheReadTokens += usage.cacheReadInputTokens + } + } + break + } + // With `--input-format stream-json` the CLI stays alive waiting for more // input. Close stdin on `result` so it exits cleanly, then finalize so // any subagent children that survived the parent CLI get reaped. diff --git a/RxCode/App/AppState+Messaging.swift b/RxCode/App/AppState+Messaging.swift index 835a4d3c..3b694dd2 100644 --- a/RxCode/App/AppState+Messaging.swift +++ b/RxCode/App/AppState+Messaging.swift @@ -494,6 +494,7 @@ extension AppState { state.textDeltaBuffer = "" state.pendingToolResults.removeAll() state.lastStreamEventDate = nil + state.liveBackgroundTaskIds.removeAll() extraMutations?(&state) diff --git a/RxCode/App/AppState+Stream.swift b/RxCode/App/AppState+Stream.swift index 4fa5aa26..8447a453 100644 --- a/RxCode/App/AppState+Stream.swift +++ b/RxCode/App/AppState+Stream.swift @@ -53,6 +53,13 @@ extension AppState { let state = self.stateForSession(currentKey) guard state.isStreaming, state.activeStreamId == streamId else { return } + + // A live "backend agent" background task can legitimately run + // silent for far longer than the inactivity timeout while the + // turn waits to run its autonomous follow-up. Don't force-clean + // (which would SIGKILL the CLI and the task) while one is pending. + guard state.liveBackgroundTaskIds.isEmpty else { continue } + guard let last = state.lastStreamEventDate else { continue } let gap = Date().timeIntervalSince(last) @@ -445,6 +452,7 @@ extension AppState { state.textDeltaBuffer = "" state.pendingToolResults.removeAll() state.lastStreamEventDate = nil + state.liveBackgroundTaskIds.removeAll() if let idx = state.messages.indices.reversed().first(where: { state.messages[$0].role == .assistant && state.messages[$0].isStreaming }) { diff --git a/RxCode/App/AppState+StreamPreflight.swift b/RxCode/App/AppState+StreamPreflight.swift new file mode 100644 index 00000000..529dda0b --- /dev/null +++ b/RxCode/App/AppState+StreamPreflight.swift @@ -0,0 +1,269 @@ +import Foundation +import os +import RxCodeChatKit +import RxCodeCore +import RxCodeSync +import SwiftUI + +extension AppState { + /// Resolved, per-backend send-request fields produced by the pre-spawn + /// preflight (MCP injection, ACP client spec, model split, background + /// context) before dispatching through the unified backend protocol. + struct StreamPreflight { + var mcpClaudeConfigPath: String? + var extraSystemPrompt: String? + var mcpCodexOverrides: [String] + var acpMCPServers: [JSONValue] + var acpSpec: ACPClientSpec? + var resolvedPrompt: String + var resolvedModel: String? + var resolvedSendMode: PermissionMode + var earlyStream: AsyncStream? + } + + /// Runs the expensive, independent pre-spawn work (memory lookup, git + /// branch + briefing, IDE-MCP port allocation, skill-context resolution, + /// session-start hooks) concurrently, then resolves the per-backend + /// send-request fields. Reads `sessionKey` but never mutates it — the + /// caller owns the `var sessionKey` used by the event loop. + func resolveStreamPreflight( + streamId: UUID, + prompt: String, + cwd: String, + cliSessionId: String?, + sessionKey: String, + agentProvider: AgentProvider, + model: String?, + permissionMode: PermissionMode, + registerMode: PermissionMode, + projectId: UUID, + includeIDEMCP: Bool, + streamStart: Date + ) async -> StreamPreflight { + // Resolve per-backend send-request fields (MCP injection, ACP client + // spec, model split) before dispatching through the unified protocol. + var mcpClaudeConfigPath: String? = nil + var extraSystemPrompt: String? = nil + var mcpCodexOverrides: [String] = [] + var acpMCPServers: [JSONValue] = [] + var acpSpec: ACPClientSpec? = nil + var resolvedPrompt = prompt + var resolvedModel: String? = model + var resolvedSendMode: PermissionMode = permissionMode + var earlyStream: AsyncStream? = nil + + func appendExtraSystemPrompt(_ context: String) { + let trimmed = context.trimmingCharacters(in: .whitespacesAndNewlines) + guard !trimmed.isEmpty else { return } + if let existing = extraSystemPrompt, !existing.isEmpty { + extraSystemPrompt = "\(existing)\n\n\(trimmed)" + } else { + extraSystemPrompt = trimmed + } + } + + // Kick off the expensive, independent pre-spawn work concurrently. + // Memory lookup, git branch + briefing, IDE-MCP port allocation, and + // skill-context resolution all hop off MainActor and previously ran + // stacked serially — pushing the first stream event well after the + // user pressed send. Each `async let` starts immediately and is only + // joined when its value is read below. + // + // Snapshot the MainActor flags into locals first: `async let` evaluates + // the right-hand side in a nonisolated autoclosure, so it can't read + // `memoryEnabled` / `memoryInjectEnabled` / `memoryMaxContextItems` + // directly. The local `let` capture solves both isolation and the + // "captured var" warning for `sessionKey`. + let memoryActive = memoryEnabled && memoryInjectEnabled + let memoryLimit = memoryMaxContextItems + let memoryMode = memoryRetrievalMode + let memoryMinScore = memoryInjectionScoreThreshold + let capturedSessionKey = sessionKey + let memoryService = self.memoryService + let marketplace = self.marketplace + let ideMCPServer = self.ideMCPServer + let shouldIncludeIDEMCP = includeIDEMCP + + func logPreflight(_ label: String, detail: String = "") { + let elapsed = Date().timeIntervalSince(streamStart) + if detail.isEmpty { + logger.info("[Stream:UI] preflight \(label, privacy: .public) stream=\(streamId) after=\(String(format: "%.2f", elapsed), privacy: .public)s") + } else { + logger.info("[Stream:UI] preflight \(label, privacy: .public) stream=\(streamId) after=\(String(format: "%.2f", elapsed), privacy: .public)s \(detail, privacy: .public)") + } + } + + async let memoryHitsAsync = memoryActive + ? await memoryService.search( + prompt, + projectId: projectId, + limit: memoryLimit, + minScore: memoryMinScore + ) + : [] + async let currentBranchAsync = GitHelper.currentBranch(at: cwd) + async let idePortAsync: UInt16? = shouldIncludeIDEMCP + ? ideMCPServer.allocate( + sessionKey: capturedSessionKey, + capabilities: agentProvider.staticCapabilities + ) + : nil + async let skillContextAsync: String? = marketplace.promptContext(for: agentProvider) + async let codexSkillOverridesAsync: [String] = shouldIncludeIDEMCP && agentProvider == .codex + ? await marketplace.codexConfigOverrides() + : [] + + let resolvedMemoryContext: String + let memoryHitCount: Int + if memoryActive { + let hits = await memoryHitsAsync + memoryHitCount = hits.count + resolvedMemoryContext = memoryContextSystemPrompt(relatedHits: hits) + } else { + memoryHitCount = 0 + resolvedMemoryContext = "" + } + logPreflight( + "memory", + detail: "enabled=\(memoryActive) mode=\(memoryMode.title) minScore=\(String(format: "%.2f", memoryMinScore)) hits=\(memoryHitCount) contextChars=\(resolvedMemoryContext.count)" + ) + + let branchBriefingContext: String + if let branch = await currentBranchAsync, + let briefing = threadStore.branchBriefingItem(projectId: projectId, branch: branch) { + branchBriefingContext = Self.branchBriefingSystemPrompt( + branch: branch, + briefing: briefing.briefing + ) + logPreflight("branchBriefing", detail: "branch=\(branch) contextChars=\(branchBriefingContext.count)") + } else { + branchBriefingContext = "" + logPreflight("branchBriefing", detail: "contextChars=0") + } + + // The IDE-MCP port is provider-agnostic at allocation time — the + // bridge command is built from the port. Per-backend MCP config + // writes still happen serially after this since they consume the + // bridge, but they no longer block memory/git/skill resolution. + let idePort = await idePortAsync + let bridge = idePort.map { IDEMCPServer.bridgeCommand(forPort: $0) } + logPreflight( + "ideMCP", + detail: "enabled=\(shouldIncludeIDEMCP) port=\(idePort.map(String.init) ?? "")" + ) + + // Session-start hooks fire once, only for a brand-new thread (no resumed + // CLI session). Their stdout is injected into this turn's agent context + // the same way the branch briefing / memory context is, and the status + // cards inserted by UserAddedHook persist via the `.result` save. + // + // Project-new-chat hooks update passive banners and are driven by the + // visible chat views. Keep them out of this preflight path: awaiting + // network-backed banner checks here delays the first backend event. + var hookStartContext = "" + if cliSessionId == nil, let project = projects.first(where: { $0.id == projectId }) { + hookStartContext = await hookManager.dispatchSessionStart( + SessionStartPayload(project: project, sessionKey: sessionKey) + ).combinedOutput + logPreflight("hooksStart", detail: "contextChars=\(hookStartContext.count)") + } + + switch agentProvider { + case .claudeCode: + mcpClaudeConfigPath = await mcp.writeClaudeConfig(projectPath: cwd, bridgeCommand: bridge) + logPreflight("claudeMCP", detail: "hasConfig=\(mcpClaudeConfigPath != nil)") + // Surface the accumulated briefing for the project's current branch + // to the agent as background context via `--append-system-prompt`. + appendExtraSystemPrompt(branchBriefingContext) + appendExtraSystemPrompt(resolvedMemoryContext) + appendExtraSystemPrompt(hookStartContext) + if let skillContext = await skillContextAsync { + logPreflight("skillContext", detail: "chars=\(skillContext.count)") + appendExtraSystemPrompt(skillContext) + } else { + logPreflight("skillContext", detail: "chars=0") + } + case .codex: + mcpCodexOverrides = await mcp.codexConfigOverrides( + projectPath: cwd, + bridgeCommand: bridge, + disableAllServers: !shouldIncludeIDEMCP + ) + if !shouldIncludeIDEMCP { + mcpCodexOverrides += ["--disable", "plugins"] + } + logPreflight( + "codexMCP", + detail: "args=\(mcpCodexOverrides.count) disabledAll=\(!shouldIncludeIDEMCP)" + ) + let codexSkillOverrides = await codexSkillOverridesAsync + logPreflight("codexSkillOverrides", detail: "args=\(codexSkillOverrides.count)") + mcpCodexOverrides += codexSkillOverrides + resolvedPrompt = Self.promptWithBackgroundContext( + [branchBriefingContext, resolvedMemoryContext, hookStartContext], + prompt: resolvedPrompt + ) + if let skillContext = await skillContextAsync { + logPreflight("skillContext", detail: "chars=\(skillContext.count)") + resolvedPrompt = "\(skillContext)\n\nUser request:\n\(resolvedPrompt)" + } else { + logPreflight("skillContext", detail: "chars=0") + } + resolvedSendMode = registerMode + case .acp: + acpMCPServers = await mcp.acpMCPServers( + projectPath: cwd, + bridgeCommand: bridge + ) + logPreflight("acpMCP", detail: "servers=\(acpMCPServers.count)") + resolvedPrompt = Self.promptWithBackgroundContext( + [branchBriefingContext, resolvedMemoryContext, hookStartContext], + prompt: resolvedPrompt + ) + if let skillContext = await skillContextAsync { + logPreflight("skillContext", detail: "chars=\(skillContext.count)") + resolvedPrompt = "\(skillContext)\n\nUser request:\n\(resolvedPrompt)" + } else { + logPreflight("skillContext", detail: "chars=0") + } + // `model` may be a composite `::` key (from the picker) + // or a bare model id (from a per-session override). + let split = acpSelectionParts(for: model) + let resolvedClientId = split?.clientId + ?? sessionStates[sessionKey]?.acpClientId + ?? selectedACPClientId + resolvedModel = split?.model ?? model + resolvedSendMode = registerMode + if let spec = acpClients.first(where: { $0.id == resolvedClientId && $0.enabled }) { + acpSpec = spec + } else { + logger.error("[ACP] no enabled client for id=\(resolvedClientId, privacy: .public)") + earlyStream = AsyncStream { c in + c.yield(.user(UserMessage( + toolUseId: nil, + content: "No ACP client configured. Add one in Settings → ACP Clients.", + isError: true + ))) + c.yield(.result(ResultEvent( + durationMs: nil, totalCostUsd: nil, + sessionId: cliSessionId ?? sessionKey, + isError: true, totalTurns: nil, usage: nil, contextWindow: nil + ))) + c.finish() + } + } + } + + return StreamPreflight( + mcpClaudeConfigPath: mcpClaudeConfigPath, + extraSystemPrompt: extraSystemPrompt, + mcpCodexOverrides: mcpCodexOverrides, + acpMCPServers: acpMCPServers, + acpSpec: acpSpec, + resolvedPrompt: resolvedPrompt, + resolvedModel: resolvedModel, + resolvedSendMode: resolvedSendMode, + earlyStream: earlyStream + ) + } +} diff --git a/RxCode/App/AppState.swift b/RxCode/App/AppState.swift index 0bd78df4..331b005b 100644 --- a/RxCode/App/AppState.swift +++ b/RxCode/App/AppState.swift @@ -35,6 +35,14 @@ struct SessionStreamState { var streamTask: Task? var hasUncheckedCompletion = false + /// Live background-task ids ("backend agents": background shells, subagents) + /// this session has spawned that have not yet reported completion. Recent + /// Claude Code ends the spawning turn with a `result` while these keep + /// running, then autonomously runs a follow-up turn when they finish. While + /// this set is non-empty we keep the CLI process alive and the UI "in + /// progress" instead of tearing the turn down on that yield `result`. + var liveBackgroundTaskIds: Set = [] + /// Last time any stream event (system / assistant / tool result / etc.) arrived /// for the active stream. Updated in `processStream` and polled by the /// inactivity watchdog so a CLI that goes silent without exiting (broken pipe diff --git a/RxCode/App/WorkspaceManager.swift b/RxCode/App/WorkspaceManager.swift index 6406e5f6..eb31837f 100644 --- a/RxCode/App/WorkspaceManager.swift +++ b/RxCode/App/WorkspaceManager.swift @@ -27,6 +27,7 @@ final class WorkspaceManager { /// Whether `MobileSyncService.start()` has run (it is process-global and not /// idempotent, so it must fire exactly once). private var didStartMobileSync = false + @ObservationIgnored private var performanceMonitorTask: Task? init() { let core = AppCore() @@ -47,6 +48,7 @@ final class WorkspaceManager { let workspace = snapshot.all.first { $0.id == workspaceID } ?? snapshot.active let appState = AppState(core: core, workspace: workspace) appStatesByWorkspaceID[workspace.id] = appState + startPerformanceMonitorIfNeeded() return appState } @@ -94,4 +96,79 @@ final class WorkspaceManager { markFrontmost(AppWorkspace.personalID) } } + + private func startPerformanceMonitorIfNeeded() { + guard performanceMonitorTask == nil else { return } + performanceMonitorTask = Task { @MainActor [weak self] in + guard let self else { return } + await PerformanceDiagnosticsService.shared.append( + state: await performanceStateSnapshot(), + mainActorMaxLagMilliseconds: 0 + ) + + let clock = ContinuousClock() + var lastSample = clock.now + var maximumLagMilliseconds = 0.0 + + while !Task.isCancelled { + let expectedWake = clock.now.advanced(by: .seconds(5)) + try? await Task.sleep(for: .seconds(5)) + guard !Task.isCancelled else { return } + + let now = clock.now + let lag = expectedWake.duration(to: now).components + let lagMilliseconds = max( + 0, + Double(lag.seconds) * 1_000 + + Double(lag.attoseconds) / 1_000_000_000_000_000 + ) + maximumLagMilliseconds = max(maximumLagMilliseconds, lagMilliseconds) + + guard lastSample.duration(to: now) >= .seconds(30) else { continue } + let snapshot = await performanceStateSnapshot() + let intervalMaximumLag = maximumLagMilliseconds + lastSample = now + maximumLagMilliseconds = 0 + await PerformanceDiagnosticsService.shared.append( + state: snapshot, + mainActorMaxLagMilliseconds: intervalMaximumLag + ) + } + } + } + + private func performanceStateSnapshot() async -> PerformanceStateSnapshot { + let appStates = Array(appStatesByWorkspaceID.values) + let states = appStates.flatMap { $0.sessionStates.values } + let messageCounts = states.map { $0.messages.count } + var searchIndexedThreadCount = 0 + var searchChunkCount = 0 + for appState in appStates { + let stats = await appState.searchService.diagnosticStats() + searchIndexedThreadCount += stats.threadCount + searchChunkCount += stats.chunkCount + } + return PerformanceStateSnapshot( + workspaceCount: appStatesByWorkspaceID.count, + projectCount: appStatesByWorkspaceID.values.reduce(0) { $0 + $1.projects.count }, + sessionSummaryCount: appStatesByWorkspaceID.values.reduce(0) { $0 + $1.allSessionSummaries.count }, + inMemorySessionCount: states.count, + inMemoryMessageCount: messageCounts.reduce(0, +), + inMemoryBlockCount: states.reduce(0) { total, state in + total + state.messages.reduce(0) { $0 + $1.blocks.count } + }, + largestSessionMessageCount: messageCounts.max() ?? 0, + activeStreamCount: states.count { $0.isStreaming }, + streamTaskCount: states.count { $0.streamTask != nil }, + flushTaskCount: states.count { $0.flushTask != nil }, + editingFileSnapshotCount: states.reduce(0) { $0 + $1.editingFileSnapshots.count }, + sessionRedirectCount: appStatesByWorkspaceID.values.reduce(0) { $0 + $1.sessionIdRedirect.count }, + pendingCompletionCount: appStatesByWorkspaceID.values.reduce(0) { $0 + $1.pendingStreamCompletions.count }, + completionWaiterCount: appStatesByWorkspaceID.values.reduce(0) { $0 + $1.streamCompletionWaiters.count }, + pendingMobileRequestCount: appStatesByWorkspaceID.values.reduce(0) { $0 + $1.mobilePendingRequests.count }, + reconciledSessionCount: appStatesByWorkspaceID.values.reduce(0) { $0 + $1.lastReconciledJsonlSize.count }, + searchIndexedThreadCount: searchIndexedThreadCount, + searchChunkCount: searchChunkCount + ) + } } diff --git a/RxCode/Services/CodexAppServer+Process.swift b/RxCode/Services/CodexAppServer+Process.swift index fc166571..77111b82 100644 --- a/RxCode/Services/CodexAppServer+Process.swift +++ b/RxCode/Services/CodexAppServer+Process.swift @@ -64,9 +64,20 @@ extension CodexAppServer { logger.info("[CodexAppServer] launch start stream=\(streamId) cwd=\(cwd ?? "", privacy: .public) overrides=\(configOverrides.count)") let process = Process() process.executableURL = URL(fileURLWithPath: binary) - process.arguments = ["app-server", "--listen", "stdio://"] + configOverrides + let env = await resolvedEnvironment() + var arguments = ["app-server", "--listen", "stdio://"] + configOverrides + // Recent Codex routes tool execution (terminal commands, apply_patch) + // through a separate `codex-code-mode-host` sidecar binary. The Homebrew + // Cask ships only the main `codex` binary, so the sidecar is missing and + // every edit/command fails with "code-mode host ... No such file". When we + // can't locate the sidecar, disable the feature so execution runs directly. + if !Self.codeModeHostAvailable(binary: binary, path: env["PATH"]) { + arguments += ["-c", "features.code_mode_host=false"] + logger.warning("[CodexAppServer] codex-code-mode-host sidecar not found; disabling code_mode_host feature so tool execution runs directly") + } + process.arguments = arguments if let cwd { process.currentDirectoryURL = URL(fileURLWithPath: cwd) } - process.environment = await resolvedEnvironment() + process.environment = env let stdin = Pipe() let stdout = Pipe() @@ -101,6 +112,39 @@ extension CodexAppServer { stderrBuffers[streamId, default: ""] += text } + /// Whether the `codex-code-mode-host` sidecar (spawned by Codex for tool + /// execution when the `code_mode_host` feature is on) can be located. Mirrors + /// how Codex resolves it: the `CODEX_CODE_MODE_HOST_PATH` env override, a + /// sibling of the `codex` binary (and of its symlink target — the Homebrew + /// symlink dir differs from the Caskroom target dir), or the shell `PATH`. + static func codeModeHostAvailable(binary: String, path: String?) -> Bool { + let fm = FileManager.default + let hostName = "codex-code-mode-host" + + if let explicit = ProcessInfo.processInfo.environment["CODEX_CODE_MODE_HOST_PATH"], + !explicit.isEmpty, fm.isExecutableFile(atPath: explicit) { + return true + } + + let binaryDir = (binary as NSString).deletingLastPathComponent + let resolvedDir = ((binary as NSString).resolvingSymlinksInPath as NSString).deletingLastPathComponent + for dir in [binaryDir, resolvedDir] where !dir.isEmpty { + if fm.isExecutableFile(atPath: (dir as NSString).appendingPathComponent(hostName)) { + return true + } + } + + if let path, !path.isEmpty { + for dir in path.split(separator: ":") where !dir.isEmpty { + if fm.isExecutableFile(atPath: (String(dir) as NSString).appendingPathComponent(hostName)) { + return true + } + } + } + + return false + } + func findNvmCodexBinary(root: String) -> String? { let fm = FileManager.default guard let versions = try? fm.contentsOfDirectory(atPath: root) else { return nil } diff --git a/RxCode/Services/PerformanceDiagnosticsService.swift b/RxCode/Services/PerformanceDiagnosticsService.swift new file mode 100644 index 00000000..db10f53f --- /dev/null +++ b/RxCode/Services/PerformanceDiagnosticsService.swift @@ -0,0 +1,132 @@ +import Darwin +import Foundation +import RxCodeCore +import os + +nonisolated struct PerformanceStateSnapshot: Codable, Sendable { + let workspaceCount: Int + let projectCount: Int + let sessionSummaryCount: Int + let inMemorySessionCount: Int + let inMemoryMessageCount: Int + let inMemoryBlockCount: Int + let largestSessionMessageCount: Int + let activeStreamCount: Int + let streamTaskCount: Int + let flushTaskCount: Int + let editingFileSnapshotCount: Int + let sessionRedirectCount: Int + let pendingCompletionCount: Int + let completionWaiterCount: Int + let pendingMobileRequestCount: Int + let reconciledSessionCount: Int + let searchIndexedThreadCount: Int + let searchChunkCount: Int +} + +private nonisolated struct ProcessPerformanceMetrics: Codable, Sendable { + let residentMemoryMB: Double + let cpuUserSeconds: Double + let cpuSystemSeconds: Double + + static func current() -> ProcessPerformanceMetrics { + var info = mach_task_basic_info_data_t() + var count = mach_msg_type_number_t( + MemoryLayout.size / MemoryLayout.size + ) + let result = withUnsafeMutablePointer(to: &info) { pointer in + pointer.withMemoryRebound(to: integer_t.self, capacity: Int(count)) { + task_info( + mach_task_self_, + task_flavor_t(MACH_TASK_BASIC_INFO), + $0, + &count + ) + } + } + guard result == KERN_SUCCESS else { + return ProcessPerformanceMetrics( + residentMemoryMB: 0, + cpuUserSeconds: 0, + cpuSystemSeconds: 0 + ) + } + return ProcessPerformanceMetrics( + residentMemoryMB: Double(info.resident_size) / 1_048_576, + cpuUserSeconds: Self.seconds(info.user_time), + cpuSystemSeconds: Self.seconds(info.system_time) + ) + } + + private static func seconds(_ value: time_value_t) -> Double { + Double(value.seconds) + Double(value.microseconds) / 1_000_000 + } +} + +private nonisolated struct PerformanceLogRecord: Codable, Sendable { + let timestamp: Date + let uptimeSeconds: TimeInterval + let mainActorMaxLagMilliseconds: Double + let process: ProcessPerformanceMetrics + let state: PerformanceStateSnapshot + let events: PerformanceDiagnostics.Snapshot +} + +/// Writes a compact JSONL sample every 30 seconds. Records contain only counts, +/// timings, and process metrics; no prompts, paths, session ids, or tool data. +actor PerformanceDiagnosticsService { + static let shared = PerformanceDiagnosticsService() + + static let logURL = AppSupport.bundleScopedURL + .appendingPathComponent("diagnostics", isDirectory: true) + .appendingPathComponent("performance.jsonl") + + private static let maximumLogBytes: UInt64 = 5 * 1_024 * 1_024 + private let logger = Logger(subsystem: "com.claudework", category: "PerformanceDiagnostics") + private let encoder: JSONEncoder = { + let encoder = JSONEncoder() + encoder.dateEncodingStrategy = .iso8601 + encoder.outputFormatting = [.sortedKeys] + return encoder + }() + + func append(state: PerformanceStateSnapshot, mainActorMaxLagMilliseconds: Double) { + let record = PerformanceLogRecord( + timestamp: .now, + uptimeSeconds: ProcessInfo.processInfo.systemUptime, + mainActorMaxLagMilliseconds: mainActorMaxLagMilliseconds, + process: .current(), + state: state, + events: PerformanceDiagnostics.drain() + ) + + do { + try prepareLogFile() + var data = try encoder.encode(record) + data.append(0x0A) + if !FileManager.default.fileExists(atPath: Self.logURL.path) { + FileManager.default.createFile(atPath: Self.logURL.path, contents: nil) + } + let handle = try FileHandle(forWritingTo: Self.logURL) + try handle.seekToEnd() + try handle.write(contentsOf: data) + try handle.close() + } catch { + logger.error("Failed to write performance diagnostics: \(error.localizedDescription)") + } + } + + private func prepareLogFile() throws { + let fileManager = FileManager.default + let directory = Self.logURL.deletingLastPathComponent() + try fileManager.createDirectory(at: directory, withIntermediateDirectories: true) + + guard let attributes = try? fileManager.attributesOfItem(atPath: Self.logURL.path), + let size = attributes[.size] as? NSNumber, + size.uint64Value >= Self.maximumLogBytes else { return } + + let rotatedURL = Self.logURL.appendingPathExtension("1") + try? fileManager.removeItem(at: rotatedURL) + try fileManager.moveItem(at: Self.logURL, to: rotatedURL) + } +} diff --git a/RxCode/Services/ThreadSearchService.swift b/RxCode/Services/ThreadSearchService.swift index 0c8ef2fe..08dda8fb 100644 --- a/RxCode/Services/ThreadSearchService.swift +++ b/RxCode/Services/ThreadSearchService.swift @@ -25,6 +25,11 @@ actor ThreadSearchService { let hits: [Hit] } + struct DiagnosticStats: Sendable, Equatable { + let threadCount: Int + let chunkCount: Int + } + /// Literal substring match inside the currently-open thread. Distinct from /// `Hit` because we surface the matched message rather than a cosine-ranked /// chunk, and we keep the surrounding text window so the overlay can @@ -67,6 +72,13 @@ actor ThreadSearchService { init() {} + func diagnosticStats() -> DiagnosticStats { + DiagnosticStats( + threadCount: index.count, + chunkCount: index.values.reduce(0) { $0 + $1.count } + ) + } + func setWorkspaceDefaults(_ defaults: WorkspaceDefaults) async { workspaceDefaults = defaults }