diff --git a/Docs/architecture.md b/Docs/architecture.md index b5a99909..e85c7799 100644 --- a/Docs/architecture.md +++ b/Docs/architecture.md @@ -107,6 +107,7 @@ flowchart LR - Process-bound routing, upstream readiness, DocumentationSearch discovery, startup summaries, and permission-dialog automation read the same cached snapshot. Permission automation covers the known helper applications exposed by `NSWorkspace`; it does not claim an inventory of every OS process. - The runtime does not run `pgrep`, enumerate all PIDs, or periodically rescan Xcode membership. Inventory membership changes only from the initial/KVO snapshots. Recovery may re-run route reconciliation against the monitor's cached snapshot, but it never re-reads OS process inventory. - Every compatible Xcode PID in the cached inventory owns an independent route. As soon as membership is established, the runtime starts the route's primary bridge and sends that route's protocol `initialize` without waiting for another Xcode's response, permission decision, initialized notification, or catalog. `CanonicalHandshakeState` accepts concurrent proof-bound participants; the first participant to complete its initialized-notification and health commit publishes the canonical semantic result, and results that differ only in `serverInfo` join it. Each joined route loads its own catalog, after which workspace and tab ownership select the matching PID. A meaningful result mismatch terminates the current activation attempt for that route; the route may retry after its cooldown and must pass compatibility again. +- Client-facing `tools/list` availability does not wait for every route catalog to converge. It may return the union of the currently committed route catalogs as an available, non-canonical surface while missing routes continue under their activation and background-refresh owners. The canonical catalog is published only when every currently required route catalog is present; later route commits publish `notifications/tools/list_changed` when the client-visible surface changes. - Canonical initialize support is bound to the exact upstream topology proof, including slot generation. Raw compatible initialize evidence is retained separately from current health eligibility. When the publishing proof is quarantined or exits, the exposed source and raw result rebind to an eligible compatible survivor. If every raw supporter is quarantined, the initialize result is hidden from clients while the retained semantic baseline still rejects incompatible offers. Only a validated health probe or validated `tools/list` response restores eligibility; ordinary request success does not. Detaching the last raw supporter clears the semantic baseline, and a delayed clear from an old slot generation cannot remove its replacement. - Quarantine, detach, and slot replacement evict the catalog and abandon catalog-attempt resources owned by the affected exact proof. A recovered proof can re-expose its retained initialize result after validation, but it is not routable for tools until a fresh `tools/list` load commits a new catalog. - If a downstream `initialize` is pending while only quarantined raw supporters remain, the initialize owner arms one generation-fenced timer for the earliest exact proof and quarantine deadline. The callback validates the topology proof, health-probe generation, deadline, and pending waiter before probing; failure re-arms from the current health state, while publication, removal of the last waiter, debug reset, and shutdown cancel the timer. This recovery reads existing topology and never rescans Xcode processes. diff --git a/Sources/XcodeMCPProxyRuntime/Session/ControlPlane/ProcessControlPlaneAuthority.swift b/Sources/XcodeMCPProxyRuntime/Session/ControlPlane/ProcessControlPlaneAuthority.swift index 633a7974..a86e62c0 100644 --- a/Sources/XcodeMCPProxyRuntime/Session/ControlPlane/ProcessControlPlaneAuthority.swift +++ b/Sources/XcodeMCPProxyRuntime/Session/ControlPlane/ProcessControlPlaneAuthority.swift @@ -1507,7 +1507,8 @@ final class ProcessControlPlaneAuthority: Sendable { func beginCatalogAttempt( routeID: ProcessRouteID, preferredUpstreamProof: UpstreamTopologyProof, - nowUptimeNanoseconds: UInt64 + nowUptimeNanoseconds: UInt64, + allowsConcurrentLoad: Bool = true ) -> (CatalogLease, ProcessControlPlaneTransition)? { state.withLockedValue { state in state.nowUptimeNs = max(state.nowUptimeNs, nowUptimeNanoseconds) @@ -1518,6 +1519,9 @@ final class ProcessControlPlaneAuthority: Sendable { ) else { return nil } if var attempt = record.attempt, [.pending, .attaching, .initialized, .loadingCatalog].contains(attempt.phase) { + guard allowsConcurrentLoad || attempt.loads.isEmpty else { + return nil + } let effects = attempt.readinessToken.map { [ProcessControlPlaneEffect.cancelReadinessWaiter($0)] } ?? [] diff --git a/Sources/XcodeMCPProxyRuntime/Session/Runtime/RuntimeCoordinator+ControlPlane.swift b/Sources/XcodeMCPProxyRuntime/Session/Runtime/RuntimeCoordinator+ControlPlane.swift index e78618a6..5033c410 100644 --- a/Sources/XcodeMCPProxyRuntime/Session/Runtime/RuntimeCoordinator+ControlPlane.swift +++ b/Sources/XcodeMCPProxyRuntime/Session/Runtime/RuntimeCoordinator+ControlPlane.swift @@ -228,6 +228,21 @@ extension RuntimeCoordinator { guard uncachedExposures.isEmpty == false else { throw UpstreamSlotScheduler.AcquisitionError.unavailable } + let uncachedProcessIDs = Set(uncachedExposures.map(\.route.target.processID)) + if let surface = currentSurface, + let sourceProof = surface.sourceProof { + refreshMissingProcessToolsCatalogsIfNeeded( + reason: "foreground_partial_catalog", + processIDs: uncachedProcessIDs + ) + return CanonicalToolsCatalogLoadResult( + rawResult: surface.rawResult, + sourceProof: sourceProof, + durationMilliseconds: elapsedMilliseconds( + sinceUptimeNanoseconds: startedAt + ) + ) + } let routes = uncachedExposures.compactMap { exposure -> AvailableToolsCatalogRoute? in guard let preferred = exposure.usableUpstreamIDs.first, let preferredProof = upstreamTopology.operationLease(for: preferred)?.proof, @@ -254,14 +269,19 @@ extension RuntimeCoordinator { throw UpstreamSlotScheduler.AcquisitionError.unavailable } - return try await loadAvailableToolsCatalogsInBatch( + let result = try await loadAvailableToolsCatalogsInBatch( routes, requestTimeout: requestTimeout, deadlineUptimeNs: deadlineUptimeNs, startedAt: startedAt, exposedProcessIDs: exposedProcessIDs, - returnAfterFirstSuccess: false + returnAfterFirstSuccess: true + ) + refreshMissingProcessToolsCatalogsIfNeeded( + reason: "foreground_first_catalog", + processIDs: uncachedProcessIDs ) + return result } private func loadAvailableToolsCatalogsInBatch( @@ -439,7 +459,8 @@ extension RuntimeCoordinator { let preferredProof = upstreamTopology.operationLease(for: preferred)?.proof, let (lease, transition) = beginProcessCatalogAttemptIfRunning( routeID: exposure.route.id, - preferredUpstreamProof: preferredProof + preferredUpstreamProof: preferredProof, + allowsConcurrentLoad: false ) else { return nil } applyProcessControlPlaneTransition(transition) return AvailableToolsCatalogRoute( @@ -480,7 +501,8 @@ extension RuntimeCoordinator { func beginProcessCatalogAttemptIfRunning( routeID: ProcessRouteID, - preferredUpstreamProof: UpstreamTopologyProof + preferredUpstreamProof: UpstreamTopologyProof, + allowsConcurrentLoad: Bool = true ) -> (CatalogLease, ProcessControlPlaneTransition)? { let nowUptimeNanoseconds = nowUptimeNanoseconds() var attempt: (CatalogLease, ProcessControlPlaneTransition)? @@ -488,7 +510,8 @@ extension RuntimeCoordinator { attempt = processControlPlane.beginCatalogAttempt( routeID: routeID, preferredUpstreamProof: preferredUpstreamProof, - nowUptimeNanoseconds: nowUptimeNanoseconds + nowUptimeNanoseconds: nowUptimeNanoseconds, + allowsConcurrentLoad: allowsConcurrentLoad ) }) else { return nil diff --git a/Tests/XcodeMCPProxyRuntimeTests/ControlPlaneAuthorityTests.swift b/Tests/XcodeMCPProxyRuntimeTests/ControlPlaneAuthorityTests.swift index d86c3b5c..9a9ae61d 100644 --- a/Tests/XcodeMCPProxyRuntimeTests/ControlPlaneAuthorityTests.swift +++ b/Tests/XcodeMCPProxyRuntimeTests/ControlPlaneAuthorityTests.swift @@ -957,6 +957,33 @@ struct ControlPlaneAuthorityTests { #expect(toolNames(authority.canonicalToolsCatalogRaw()) == ["Foreground"]) } + @Test func backgroundCatalogAdmissionCoalescesWhileForegroundCanJoin() throws { + let target = xcodeProcessTarget(processID: 41050, xcodeVersion: "27.0") + let authority = makeAuthority([(target, [0])]) + let route = try #require(authority.route(forProcessID: target.processID)) + let (background, _) = try #require(authority.beginCatalogAttempt( + routeID: route.id, + preferredUpstreamProof: testTopologyProof(0), + nowUptimeNanoseconds: 1, + allowsConcurrentLoad: false + )) + + #expect(authority.beginCatalogAttempt( + routeID: route.id, + preferredUpstreamProof: testTopologyProof(0), + nowUptimeNanoseconds: 2, + allowsConcurrentLoad: false + ) == nil) + + let (foreground, _) = try #require(authority.beginCatalogAttempt( + routeID: route.id, + preferredUpstreamProof: testTopologyProof(0), + nowUptimeNanoseconds: 3 + )) + #expect(foreground.attempt == background.attempt) + #expect(foreground != background) + } + @Test func catalogCommitUsesActualFallbackResponseProof() throws { let group = borrowSharedTestEventLoopGroup() defer { shutdownAndWait(group) } diff --git a/Tests/XcodeMCPProxyRuntimeTests/RuntimeCoordinatorTests.swift b/Tests/XcodeMCPProxyRuntimeTests/RuntimeCoordinatorTests.swift index ef82e6f0..55969f2a 100644 --- a/Tests/XcodeMCPProxyRuntimeTests/RuntimeCoordinatorTests.swift +++ b/Tests/XcodeMCPProxyRuntimeTests/RuntimeCoordinatorTests.swift @@ -2470,6 +2470,10 @@ struct RuntimeCoordinatorProcessRoutingTests { let secondInitialize = try await secondUpstream.nextSent( matching: { methodName(from: $0) == "initialize" } ) + let pendingInitialize = fixture.registerInitialize( + requestID: 27021, + sessionID: "session-compatible-sibling-first-success" + ) await firstUpstream.yield( .message( @@ -2490,6 +2494,15 @@ struct RuntimeCoordinatorProcessRoutingTests { _ = try await secondUpstream.nextSent( matching: { methodName(from: $0) == "notifications/initialized" } ) + let downstreamInitialize = try decodeJSON( + from: try await waitWithTimeout( + "waiting for compatible sibling to complete downstream initialize" + ) { + try await pendingInitialize.get() + } + ) + #expect(downstreamInitialize["result"] != nil) + #expect(fixture.manager.canonicalHandshakeState.initializeSourceUpstream() == 1) let secondTools = try await secondUpstream.nextSent( matching: { methodName(from: $0) == "tools/list" } ) @@ -7508,7 +7521,7 @@ struct RuntimeCoordinatorRecoveryTests { } } - @Test func sessionManagerToolsListWaitsForCompleteCatalogDespiteKnownOwner() + @Test func sessionManagerToolsListReturnsAvailableCatalogDespiteKnownOwner() async throws { let group = borrowSharedTestEventLoopGroup() @@ -7519,7 +7532,7 @@ struct RuntimeCoordinatorRecoveryTests { let latestTarget = xcodeProcessTarget(processID: 80422, xcodeVersion: "27.0") let olderTarget = xcodeProcessTarget(processID: 66333, xcodeVersion: "26.6") let manager = RuntimeCoordinator( - config: makeConfig(requestTimeout: 5), + config: makeConfig(requestTimeout: 0), eventLoop: eventLoop, upstreams: [olderUpstream, latestUpstream], xcodeProcessRoutes: [ @@ -7531,6 +7544,15 @@ struct RuntimeCoordinatorRecoveryTests { defer { manager.shutdownAndWait() } manager.markUpstreamInitialized(upstreamIndex: 0) manager.markUpstreamInitialized(upstreamIndex: 1) + seedCanonicalInitializeForTesting( + on: manager, + result: try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "known-owner-first-success"], + ]), + sourceUpstream: 1 + ) #expect( manager.recordXcodeWindowOwners( from: try jsonValue([ @@ -7545,7 +7567,7 @@ struct RuntimeCoordinatorRecoveryTests { let task = Task { try await manager.sharedToolsList( sessionID: "session-process-catalog-available-owner", - requestTimeoutOverride: .seconds(5) + requestTimeoutOverride: nil ) } @@ -7564,23 +7586,37 @@ struct RuntimeCoordinatorRecoveryTests { ) ) ) + let partial = try await waitWithTimeout("waiting for available process catalog") { + try await task.value + } + #expect(Set(toolNames(in: partial)) == Set(["Only27", "SharedTool"])) + #expect(toolDescription(in: partial, name: "SharedTool") == "from-27") #expect(manager.cachedToolsListResult() == nil) + let cancellation = try await olderUpstream.nextSent( + startingAt: 1, + matching: { methodName(from: $0) == "notifications/cancelled" } + ) + #expect( + try extractCancellationRequestID(from: cancellation) + == extractUpstreamID(from: olderRequest) + ) + let backgroundRequest = try await olderUpstream.nextSent( + startingAt: 2, + matching: { methodName(from: $0) == "tools/list" } + ) await olderUpstream.yield( .message( try makeDocumentationToolsListResponse( - id: try extractUpstreamID(from: olderRequest), + id: try extractUpstreamID(from: backgroundRequest), tools: [ toolDescriptor(name: "Only26", description: "old-only") ] ) ) ) - - let result = try await waitWithTimeout("waiting for process-routed tools/list") { - try await task.value + _ = try await waitWithTimeout("waiting for complete process catalog") { + await manager.drainRuntimeTasksForTesting() } - #expect(Set(toolNames(in: result)) == Set(["Only26", "Only27", "SharedTool"])) - #expect(toolDescription(in: result, name: "SharedTool") == "from-27") #expect(manager.debugSnapshot().controlPlane?.canonicalToolsSourceUpstream == 1) #expect( Set(toolNames(in: manager.cachedToolsListResult() ?? .null)) @@ -7592,7 +7628,7 @@ struct RuntimeCoordinatorRecoveryTests { "Only27", "SharedTool", ])) - #expect(await olderUpstream.sentCount() == 1) + #expect(await olderUpstream.sentCount() == 3) #expect(manager.debugSnapshot().upstreams[0].activeCorrelatedRequestCount == 0) let catalogs = manager.debugSnapshot().processToolCatalogs @@ -7629,6 +7665,15 @@ struct RuntimeCoordinatorRecoveryTests { ) defer { manager.shutdownAndWait() } manager.markUpstreamInitialized(upstreamIndex: 0) + seedCanonicalInitializeForTesting( + on: manager, + result: try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "fallback-owner"], + ]), + sourceUpstream: 0 + ) let fallbackTask = Task { try await manager.sharedToolsList( @@ -7668,15 +7713,17 @@ struct RuntimeCoordinatorRecoveryTests { #expect(toolNames(in: manager.cachedToolsListResult() ?? .null) == ["FallbackOnly"]) manager.markUpstreamInitialized(upstreamIndex: 1) #expect(manager.cachedToolsListResult() == nil) - - let ownerTask = Task { - try await manager.sharedToolsList( - sessionID: "session-process-catalog-after-owner", - requestTimeoutOverride: .seconds(5) - ) - } + manager.refreshMissingProcessToolsCatalogsIfNeeded( + reason: "test_owner_catalog_background_refresh", + processIDs: [ownerTarget.processID] + ) let ownerRequest = try await sentValue(from: ownerUpstream, at: 0, timeout: .seconds(2)) #expect(methodName(from: ownerRequest) == "tools/list") + let ownerResult = try await manager.sharedToolsList( + sessionID: "session-process-catalog-after-owner", + requestTimeoutOverride: nil + ) + #expect(toolNames(in: ownerResult) == ["FallbackOnly"]) await ownerUpstream.yield( .message( try makeDocumentationToolsListResponse( @@ -7687,10 +7734,9 @@ struct RuntimeCoordinatorRecoveryTests { ) ) ) - let ownerResult = try await waitWithTimeout("waiting for owner tools/list") { - try await ownerTask.value + _ = try await waitWithTimeout("waiting for owner catalog completion") { + await manager.drainRuntimeTasksForTesting() } - #expect(Set(toolNames(in: ownerResult)) == Set(["FallbackOnly", "OwnerOnly"])) #expect( Set(toolNames(in: manager.cachedToolsListResult() ?? .null)) == Set([ @@ -7701,7 +7747,7 @@ struct RuntimeCoordinatorRecoveryTests { #expect(await ownerUpstream.sentCount() == 1) } - @Test func sessionManagerToolsListWaitsForStalledRouteBeforePublishingCompleteCatalog() + @Test func sessionManagerToolsListReturnsFirstCatalogAndCompletesMissingRouteInBackground() async throws { let group = borrowSharedTestEventLoopGroup() @@ -7711,24 +7757,44 @@ struct RuntimeCoordinatorRecoveryTests { let newerUpstream = TestUpstreamClient() let olderTarget = xcodeProcessTarget(processID: 66338, xcodeVersion: "26.6") let newerTarget = xcodeProcessTarget(processID: 80425, xcodeVersion: "27.0") + let catalogCommits = LockedRecordedValues<(pid_t, Int)>() let manager = RuntimeCoordinator( - config: makeConfig(requestTimeout: 5), + config: makeConfig(requestTimeout: 0), eventLoop: eventLoop, upstreams: [olderUpstream, newerUpstream], xcodeProcessRoutes: [ XcodeProcessRoute(target: olderTarget, upstreamIndices: [0]), XcodeProcessRoute(target: newerTarget, upstreamIndices: [1]), ], + testHooks: RuntimeCoordinatorTestHooks( + processRouteCatalogCommitted: { catalogCommits.append(($0, $1)) } + ), startImmediately: false ) defer { manager.shutdownAndWait() } manager.markUpstreamInitialized(upstreamIndex: 0) manager.markUpstreamInitialized(upstreamIndex: 1) + seedCanonicalInitializeForTesting( + on: manager, + result: try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "catalog-first-success"], + ]), + sourceUpstream: 1 + ) + let notificationSessionID = "session-process-catalog-first-success-notifications" + let notificationSession = manager.session(id: notificationSessionID) + manager.sessionRegistry.markInitialized( + id: notificationSessionID, + negotiatedProtocolVersion: MCP.ProtocolVersion.current + ) + _ = notificationSession.router.drainBufferedNotifications() let task = Task { try await manager.sharedToolsList( sessionID: "session-process-catalog-later-usable-route", - requestTimeoutOverride: .seconds(5) + requestTimeoutOverride: nil ) } @@ -7741,33 +7807,113 @@ struct RuntimeCoordinatorRecoveryTests { try makeDocumentationToolsListResponse( id: try extractUpstreamID(from: newerRequest), tools: [ - toolDescriptor(name: "NewerRouteOnly") + toolDescriptor(name: "NewerRouteOnly"), + toolDescriptor(name: "XcodeListWindows"), ] ) ) ) + let partial = try await waitWithTimeout( + "waiting for first usable process catalog" + ) { + try await task.value + } + #expect(Set(toolNames(in: partial)) == Set(["NewerRouteOnly", "XcodeListWindows"])) #expect(manager.cachedToolsListResult() == nil) - #expect(await olderUpstream.sentCount() == 1) - #expect(await newerUpstream.sentCount() == 1) + #expect( + Set(toolNames(in: manager.cachedToolsListResult(forUpstreamIndex: 1) ?? .null)) + == Set(["NewerRouteOnly", "XcodeListWindows"]) + ) #expect(manager.debugSnapshot().controlPlane?.canonicalToolsSourceUpstream == nil) + let cancellation = try await olderUpstream.nextSent( + startingAt: 1, + matching: { methodName(from: $0) == "notifications/cancelled" } + ) + #expect( + try extractCancellationRequestID(from: cancellation) + == extractUpstreamID(from: olderRequest) + ) + let backgroundRequest = try await olderUpstream.nextSent( + startingAt: 2, + matching: { methodName(from: $0) == "tools/list" } + ) + #expect(try extractUpstreamID(from: backgroundRequest) != extractUpstreamID(from: olderRequest)) + #expect(await olderUpstream.sentCount() == 3) + #expect(await newerUpstream.sentCount() == 1) + _ = notificationSession.router.drainBufferedNotifications() + + let windowsTask = Task { + try await manager.liveXcodeListWindowsResult( + route: .anyHealthy, + requestTimeoutOverride: nil + ) + } + let windowsRequest = try await newerUpstream.nextSent( + startingAt: 1, + matching: { + methodName(from: $0) == "tools/call" + && toolCallName(from: $0) == "XcodeListWindows" + } + ) + #expect(await olderUpstream.sentCount() == 3) + await newerUpstream.yield( + .message( + try makeXcodeListWindowsResponse( + id: try extractUpstreamID(from: windowsRequest), + message: "* tabIdentifier: tab-newer, workspacePath: /Work/Newer.xcworkspace" + ) + ) + ) + _ = try await waitWithTimeout("waiting for cataloged-route window discovery") { + try await windowsTask.value + } + + let repeatedPartial = try await manager.sharedToolsList( + sessionID: "session-process-catalog-existing-partial", + requestTimeoutOverride: nil + ) + #expect( + Set(toolNames(in: repeatedPartial)) + == Set(["NewerRouteOnly", "XcodeListWindows"]) + ) + #expect(await olderUpstream.sentCount() == 3) + let abandonedLease = try #require( + manager.debugSnapshot().leases.first { + $0.label == "tools/list" + && $0.upstreamIndex == 0 + && $0.state == .abandoned + && $0.releaseReason == "clientDisconnected" + } + ) + #expect(abandonedLease.requestIDKey != nil) + #expect(manager.debugSnapshot().upstreams[0].activeCorrelatedRequestCount == 1) await olderUpstream.yield( .message( try makeDocumentationToolsListResponse( - id: try extractUpstreamID(from: olderRequest), + id: try extractUpstreamID(from: backgroundRequest), tools: [ toolDescriptor(name: "OlderRouteOnly") ] ) ) ) - let result = try await waitWithTimeout("waiting for complete process catalog") { - try await task.value + let backgroundCommit = try await nextRecordedValue(catalogCommits, at: 1) + #expect(backgroundCommit.0 == olderTarget.processID) + #expect(backgroundCommit.1 == 0) + _ = try await waitWithTimeout("waiting for background catalog completion") { + await manager.drainRuntimeTasksForTesting() } - #expect(Set(toolNames(in: result)) == Set(["NewerRouteOnly", "OlderRouteOnly"])) #expect( - Set(toolNames(in: manager.cachedToolsListResult() ?? .null)) == Set(["NewerRouteOnly", "OlderRouteOnly"])) + Set(toolNames(in: manager.cachedToolsListResult() ?? .null)) + == Set(["NewerRouteOnly", "OlderRouteOnly", "XcodeListWindows"]) + ) #expect(manager.debugSnapshot().upstreams[0].activeCorrelatedRequestCount == 0) + let notificationMethods = notificationSession.router.drainBufferedNotifications().compactMap { + methodName(from: $0) + } + #expect(notificationMethods == ["notifications/tools/list_changed"]) + #expect(notificationSession.router.drainBufferedNotifications().isEmpty) } @Test func sessionManagerToolsListCompletesCachedProcessCatalogWithFreshRoutes() @@ -7797,22 +7943,33 @@ struct RuntimeCoordinatorRecoveryTests { manager.markUpstreamInitialized(upstreamIndex: 0) manager.markUpstreamInitialized(upstreamIndex: 1) manager.markUpstreamInitialized(upstreamIndex: 2) + seedCanonicalInitializeForTesting( + on: manager, + result: try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "cached-fresh-routes"], + ]), + sourceUpstream: 0 + ) try seedProcessToolCatalogs( on: manager, entries: [(olderTarget, 0, [toolDescriptor(name: "OlderRouteOnly")])] ) - - let task = Task { - try await manager.sharedToolsList( - sessionID: "session-process-catalog-cached-union", - requestTimeoutOverride: .seconds(5) - ) - } + manager.refreshMissingProcessToolsCatalogsIfNeeded( + reason: "test_cached_fresh_routes_background_refresh", + processIDs: [middleTarget.processID, latestTarget.processID] + ) let middleRequest = try await sentValue(from: middleUpstream, at: 0, timeout: .seconds(2)) #expect(methodName(from: middleRequest) == "tools/list") let latestRequest = try await sentValue(from: latestUpstream, at: 0, timeout: .seconds(2)) #expect(methodName(from: latestRequest) == "tools/list") + let partial = try await manager.sharedToolsList( + sessionID: "session-process-catalog-cached-union", + requestTimeoutOverride: nil + ) + #expect(toolNames(in: partial) == ["OlderRouteOnly"]) await latestUpstream.yield( .message( try makeDocumentationToolsListResponse( @@ -7833,18 +7990,9 @@ struct RuntimeCoordinatorRecoveryTests { ) ) ) - - let result = try await waitWithTimeout("waiting for cached process catalog surface") { - try await task.value + _ = try await waitWithTimeout("waiting for complete cached process catalog") { + await manager.drainRuntimeTasksForTesting() } - - #expect( - Set(toolNames(in: result)) - == Set([ - "LatestRouteOnly", - "MiddleRouteOnly", - "OlderRouteOnly", - ])) #expect(manager.cachedToolsListResult() != nil) #expect(await olderUpstream.sentCount() == 0) #expect(await middleUpstream.sentCount() == 1) @@ -7887,22 +8035,33 @@ struct RuntimeCoordinatorRecoveryTests { defer { manager.shutdownAndWait() } manager.markUpstreamInitialized(upstreamIndex: 0) manager.markUpstreamInitialized(upstreamIndex: 1) + seedCanonicalInitializeForTesting( + on: manager, + result: try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "cached-fresh-failure"], + ]), + sourceUpstream: 0 + ) try seedProcessToolCatalogs( on: manager, entries: [ (olderTarget, 0, [toolDescriptor(name: "OlderRouteOnly")]) ] ) - - let task = Task { - try await manager.sharedToolsList( - sessionID: "session-process-catalog-cached-fresh-fails", - requestTimeoutOverride: .seconds(5) - ) - } + manager.refreshMissingProcessToolsCatalogsIfNeeded( + reason: "test_cached_fresh_failure_background_refresh", + processIDs: [latestTarget.processID] + ) let latestRequest = try await sentValue(from: latestUpstream, at: 0, timeout: .seconds(2)) #expect(methodName(from: latestRequest) == "tools/list") + let result = try await manager.sharedToolsList( + sessionID: "session-process-catalog-cached-fresh-fails", + requestTimeoutOverride: nil + ) + #expect(toolNames(in: result) == ["OlderRouteOnly"]) await latestUpstream.yield( .message( try JSONSerialization.data( @@ -7917,12 +8076,9 @@ struct RuntimeCoordinatorRecoveryTests { ) ) ) - - let result = try await waitWithTimeout("waiting for cached process catalog fallback") { - try await task.value + _ = try await waitWithTimeout("waiting for failed background catalog cleanup") { + await manager.drainRuntimeTasksForTesting() } - - #expect(toolNames(in: result) == ["OlderRouteOnly"]) #expect(manager.cachedToolsListResult() == nil) #expect(toolNames(in: manager.cachedToolsListResult(forUpstreamIndex: 0) ?? .null) == ["OlderRouteOnly"]) #expect(await olderUpstream.sentCount() == 0) @@ -8296,7 +8452,7 @@ struct RuntimeCoordinatorRecoveryTests { await manager.drainRuntimeTasksForTesting() } - @Test func sessionManagerToolsListWaitsWhileCachedProcessCatalogIsIncomplete() + @Test func sessionManagerToolsListReturnsCachedPartialWhileBackgroundCatalogIsIncomplete() async throws { let group = borrowSharedTestEventLoopGroup() @@ -8308,7 +8464,7 @@ struct RuntimeCoordinatorRecoveryTests { let latestTarget = xcodeProcessTarget(processID: 80428, xcodeVersion: "27.0") let toolsListRefreshes = NIOLockedValueBox<[String]>([]) let manager = RuntimeCoordinator( - config: makeConfig(requestTimeout: 5), + config: makeConfig(requestTimeout: 0), eventLoop: eventLoop, upstreams: [olderUpstream, latestUpstream], xcodeProcessRoutes: [ @@ -8327,19 +8483,25 @@ struct RuntimeCoordinatorRecoveryTests { defer { manager.shutdownAndWait() } manager.markUpstreamInitialized(upstreamIndex: 0) manager.markUpstreamInitialized(upstreamIndex: 1) + seedCanonicalInitializeForTesting( + on: manager, + result: try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "cached-partial"], + ]), + sourceUpstream: 0 + ) try seedProcessToolCatalogs( on: manager, entries: [ (olderTarget, 0, [toolDescriptor(name: "OlderRouteOnly")]) ] ) - - let task = Task { - try await manager.sharedToolsList( - sessionID: "session-process-catalog-cached-fresh-cancelled", - requestTimeoutOverride: .seconds(5) - ) - } + manager.refreshMissingProcessToolsCatalogsIfNeeded( + reason: "test_cached_partial_background_refresh", + processIDs: [latestTarget.processID] + ) let latestRequest = try await sentValue(from: latestUpstream, at: 0, timeout: .seconds(2)) #expect(methodName(from: latestRequest) == "tools/list") @@ -8350,6 +8512,17 @@ struct RuntimeCoordinatorRecoveryTests { manager.debugSnapshot().processToolCatalogs.map(\.processID) == [ olderTarget.processID ]) + let partial = try await manager.sharedToolsList( + sessionID: "session-process-catalog-cached-partial", + requestTimeoutOverride: nil + ) + #expect(toolNames(in: partial) == ["OlderRouteOnly"]) + let repeatedPartial = try await manager.sharedToolsList( + sessionID: "session-process-catalog-cached-partial-repeat", + requestTimeoutOverride: nil + ) + #expect(toolNames(in: repeatedPartial) == ["OlderRouteOnly"]) + #expect(await latestUpstream.sentCount() == 1) await latestUpstream.yield( .message( @@ -8361,10 +8534,9 @@ struct RuntimeCoordinatorRecoveryTests { ) ) ) - let result = try await waitWithTimeout("waiting for complete process catalog") { - try await task.value + _ = try await waitWithTimeout("waiting for complete background process catalog") { + await manager.drainRuntimeTasksForTesting() } - #expect(Set(toolNames(in: result)) == Set(["LatestRouteOnly", "OlderRouteOnly"])) #expect( Set(toolNames(in: manager.cachedToolsListResult() ?? .null)) == Set(["LatestRouteOnly", "OlderRouteOnly"]) @@ -9297,9 +9469,10 @@ struct RuntimeCoordinatorRecoveryTests { upstreamProof: manager.operationLeaseForTest(upstreamIndex: 0).proof ) - manager.refreshMissingProcessToolsCatalogsIfNeeded( - reason: "test_overlap_background_refresh", - processIDs: [target.processID] + manager.refreshProcessRouteToolsCatalog( + route: route, + upstreamProof: manager.operationLeaseForTest(upstreamIndex: 0).proof, + reason: "test_overlap_background_refresh" ) let backgroundRequest = try await sentValue( from: upstream, @@ -9873,6 +10046,21 @@ struct RuntimeCoordinatorCatalogTests { manager.markUpstreamInitialized(upstreamIndex: 0) manager.markUpstreamInitialized(upstreamIndex: 1) manager.markUpstreamInitialized(upstreamIndex: 2) + let initializeResult = try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "catalog-source"], + ]) + seedCanonicalInitializeForTesting( + on: manager, + result: initializeResult, + sourceUpstream: 0 + ) + seedCanonicalInitializeForTesting( + on: manager, + result: initializeResult, + sourceUpstream: 1 + ) try seedProcessToolCatalogs( on: manager, entries: [ @@ -9891,18 +10079,20 @@ struct RuntimeCoordinatorCatalogTests { == [remainingTarget.processID] ) #expect(manager.processControlPlane.canonicalSourceUpstream() == nil) - - let freshCatalogTask = Task { - try await manager.sharedToolsList( - sessionID: "session-resync-cleared-process-catalog", - requestTimeoutOverride: .seconds(5) - ) - } + manager.refreshMissingProcessToolsCatalogsIfNeeded( + reason: "test_cleared_source_background_refresh", + processIDs: [clearedTarget.processID] + ) let freshRequest = try await sentValue( from: clearedSibling, at: 0, timeout: .seconds(2) ) + let available = try await manager.sharedToolsList( + sessionID: "session-resync-cleared-process-catalog", + requestTimeoutOverride: nil + ) + #expect(toolNames(in: available) == ["RemainingOnlyTool"]) await clearedSibling.yield( .message( try makeDocumentationToolsListResponse( @@ -9910,11 +10100,11 @@ struct RuntimeCoordinatorCatalogTests { tools: [toolDescriptor(name: "ClearedFreshTool")] )) ) - let refreshed = try await waitWithTimeout("waiting for exact-source catalog reload") { - try await freshCatalogTask.value + _ = try await waitWithTimeout("waiting for exact-source catalog reload") { + await manager.drainRuntimeTasksForTesting() } #expect( - Set(toolNames(in: refreshed)) + Set(toolNames(in: manager.cachedToolsListResult() ?? .null)) == Set([ "ClearedFreshTool", "RemainingOnlyTool", @@ -9988,6 +10178,15 @@ struct RuntimeCoordinatorCatalogTests { ) defer { manager.shutdownAndWait() } manager.markUpstreamInitialized(upstreamIndex: 1) + seedCanonicalInitializeForTesting( + on: manager, + result: try jsonValue([ + "protocolVersion": MCP.ProtocolVersion.current, + "capabilities": [String: Any](), + "serverInfo": ["name": "warm-route"], + ]), + sourceUpstream: 1 + ) let task = Task { try await manager.sharedToolsList( @@ -10021,13 +10220,16 @@ struct RuntimeCoordinatorCatalogTests { manager.markUpstreamInitialized(upstreamIndex: 0) #expect(manager.cachedToolsListResult() == nil) - let completeTask = Task { - try await manager.sharedToolsList( - sessionID: "session-process-catalog-after-cold-warms", - requestTimeoutOverride: .seconds(5) - ) - } + manager.refreshMissingProcessToolsCatalogsIfNeeded( + reason: "test_cold_route_background_refresh", + processIDs: [coldTarget.processID] + ) let coldRequest = try await sentValue(from: coldUpstream, at: 0, timeout: .seconds(2)) + let stillAvailable = try await manager.sharedToolsList( + sessionID: "session-process-catalog-after-cold-warms", + requestTimeoutOverride: nil + ) + #expect(toolNames(in: stillAvailable) == ["WarmOnlyTool"]) await coldUpstream.yield( .message( try makeDocumentationToolsListResponse( @@ -10036,10 +10238,13 @@ struct RuntimeCoordinatorCatalogTests { ) ) ) - let complete = try await waitWithTimeout("waiting for newly warm process catalog") { - try await completeTask.value + _ = try await waitWithTimeout("waiting for newly warm process catalog") { + await manager.drainRuntimeTasksForTesting() } - #expect(Set(toolNames(in: complete)) == Set(["ColdOnlyTool", "WarmOnlyTool"])) + #expect( + Set(toolNames(in: manager.cachedToolsListResult() ?? .null)) + == Set(["ColdOnlyTool", "WarmOnlyTool"]) + ) #expect(await coldUpstream.sentCount() == 1) #expect(await warmUpstream.sentCount() == 1) }