From 99ca611ef8bd64e7e5ee8822ecba25d92f7c5cc5 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 13 Sep 2026 21:17:20 -0700 Subject: [PATCH 1/4] perf(swift-ios): apply thread catch-up in bounded batches --- apps/swift-ios/App/NativeFeatureClient.swift | 45 ++++- apps/swift-ios/Core/T3Client.swift | 6 +- apps/swift-ios/Core/WebSocketRPC.swift | 23 ++- .../CoreTests/WebSocketRPCRaceTests.swift | 14 +- .../NativeThreadCatchUpTests.swift | 156 +++++++++++++++++- 5 files changed, 223 insertions(+), 21 deletions(-) diff --git a/apps/swift-ios/App/NativeFeatureClient.swift b/apps/swift-ios/App/NativeFeatureClient.swift index 91277dec5e02..970880ea65fd 100644 --- a/apps/swift-ios/App/NativeFeatureClient.swift +++ b/apps/swift-ios/App/NativeFeatureClient.swift @@ -3898,7 +3898,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, guard !Task.isCancelled, self?.isCurrentDetail(route, generation: streamGeneration) == true, self?.environmentGeneration == sessionGeneration else { return } - let subscription = try await route.client.threadEvents( + let subscription = try await route.client.threadEventBatches( threadID: route.wireID, after: sequence, turnLimit: supportsPagination ? Self.initialThreadUserTurnLimit : nil @@ -3914,8 +3914,8 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, self?.continuation.yield(.threadSync(id: route.uiID, state: .catchingUp)) } self?.activeDetailConnectionID = subscriptionConnectionID - for try await item in subscription.events { - if case .synchronized = item { + for try await items in subscription.events { + if items.contains(where: { if case .synchronized = $0 { true } else { false } }) { let connectionID = await route.client.currentConnectionID() guard !Task.isCancelled, let self, self.isCurrentDetail(route, generation: streamGeneration), @@ -3933,8 +3933,8 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, self.continuation.yield(.threadSync(id: route.uiID, state: .catchingUp)) self.ensureDetailCatchUpFallback(route, generation: streamGeneration) } - self.consumeDetailStreamItem( - item, route: route, subscriptionEpoch: subscriptionEpoch + self.consumeDetailStreamBatch( + items, route: route, subscriptionEpoch: subscriptionEpoch ) } // A thread subscription stays open until its owner leaves. @@ -4070,10 +4070,40 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, ensureDetailCatchUpFallback(route, generation: detailStreamGeneration) } + private func consumeDetailStreamBatch( + _ items: [ThreadStreamItem], + route: NativeThreadRoute, + subscriptionEpoch: Int + ) { + // Keep snapshot replacement and page-watermark merges at their event + // positions. Ordinary replay batches share one legacy sync publication. + let needsIndividualUpdates = items.count == 1 || activeRawThread == nil + || pendingOlderThreadPage != nil || items.contains { item in + switch item { + case .snapshot: true + case let .event(event): + event["type"]?.stringValue == "thread.reverted" + || event["type"]?.stringValue == "thread.deleted" + case .synchronized: false + } + } + for item in items { + consumeDetailStreamItem( + item, route: route, subscriptionEpoch: subscriptionEpoch, + synchronizeLegacy: needsIndividualUpdates + ) + } + if !needsIndividualUpdates, + serverConfigsByEnvironmentID[route.environmentID]?.threadResumeCompletionMarker != true { + markDetailSynchronized(route) + } + } + private func consumeDetailStreamItem( _ item: ThreadStreamItem, route: NativeThreadRoute, - subscriptionEpoch: Int + subscriptionEpoch: Int, + synchronizeLegacy: Bool ) { switch item { case .synchronized: @@ -4144,7 +4174,8 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, scheduleDetailRefresh(threadID: route.uiID, client: route.client, force: true) } } - if serverConfigsByEnvironmentID[route.environmentID]?.threadResumeCompletionMarker != true { + if synchronizeLegacy, + serverConfigsByEnvironmentID[route.environmentID]?.threadResumeCompletionMarker != true { markDetailSynchronized(route) } } diff --git a/apps/swift-ios/Core/T3Client.swift b/apps/swift-ios/Core/T3Client.swift index f1a8c55340b8..881f16d6522d 100644 --- a/apps/swift-ios/Core/T3Client.swift +++ b/apps/swift-ios/Core/T3Client.swift @@ -625,18 +625,18 @@ public actor T3Client { ) } - public func threadEvents( + public func threadEventBatches( threadID: String, after sequence: Int? = nil, turnLimit: Int? = nil - ) async throws -> (events: AsyncThrowingStream, connectionID: UUID) { + ) async throws -> (events: AsyncThrowingStream<[ThreadStreamItem], Error>, connectionID: UUID) { var payload: [String: JSONValue] = [ "threadId": .string(threadID), "requestCompletionMarker": .bool(true), ] if let sequence { payload["afterSequence"] = .number(Double(sequence)) } if let turnLimit { payload["turnLimit"] = .number(Double(turnLimit)) } - return try await rpc.subscribeOnCurrentConnection( + return try await rpc.subscribeBatchesOnCurrentConnection( RPCMethod.subscribeThread.rawValue, payload: .object(payload), as: ThreadStreamItem.self diff --git a/apps/swift-ios/Core/WebSocketRPC.swift b/apps/swift-ios/Core/WebSocketRPC.swift index bd96df45577a..eb53906dabf7 100644 --- a/apps/swift-ios/Core/WebSocketRPC.swift +++ b/apps/swift-ios/Core/WebSocketRPC.swift @@ -494,16 +494,31 @@ public actor WebSocketRPCClient { _ tag: String, payload: JSONValue = .object([:]), as type: Value.Type + ) async throws -> (events: AsyncThrowingStream, connectionID: UUID) { + try await subscribeOnCurrentConnection { + subscribe(tag, payload: payload, reconnect: false, as: type) + } + } + + public func subscribeBatchesOnCurrentConnection( + _ tag: String, + payload: JSONValue = .object([:]), + as type: Value.Type + ) async throws -> (events: AsyncThrowingStream<[Value], Error>, connectionID: UUID) { + try await subscribeOnCurrentConnection { + subscribeBatches(tag, payload: payload, reconnect: false, as: type) + } + } + + private func subscribeOnCurrentConnection( + makeStream: () -> AsyncThrowingStream ) async throws -> (events: AsyncThrowingStream, connectionID: UUID) { try Task.checkCancellation() start() while true { try Task.checkCancellation() if let id = connectionID { - return ( - subscribe(tag, payload: payload, reconnect: false, as: type), - id - ) + return (makeStream(), id) } _ = try await waitForConnection(after: nil) } diff --git a/apps/swift-ios/Tests/CoreTests/WebSocketRPCRaceTests.swift b/apps/swift-ios/Tests/CoreTests/WebSocketRPCRaceTests.swift index 90084a4775ce..67564bea8a36 100644 --- a/apps/swift-ios/Tests/CoreTests/WebSocketRPCRaceTests.swift +++ b/apps/swift-ios/Tests/CoreTests/WebSocketRPCRaceTests.swift @@ -31,11 +31,13 @@ final class WebSocketRPCRaceTests: XCTestCase { connector: SequencedConnector(connections: [connection]), endpointProvider: { URL(string: "wss://studio.example/ws")! } ) - let stream = await client.subscribeBatches("thread.events", reconnect: false, as: Int.self) + let subscription = try await client.subscribeBatchesOnCurrentConnection("thread.events", as: Int.self) + let connectionID = await client.currentConnectionID() + XCTAssertEqual(subscription.connectionID, connectionID) await connection.waitUntilSubscriptionStarted() try await connection.sendSubscriptionValues((0..<600).map { .number(Double($0)) }) - var iterator = stream.makeAsyncIterator() + var iterator = subscription.events.makeAsyncIterator() var received: [Int] = [] var batchSizes: [Int] = [] while received.count < 600, let batch = try await iterator.next() { @@ -60,10 +62,10 @@ final class WebSocketRPCRaceTests: XCTestCase { subscriptionBufferLimit: 2, endpointProvider: { URL(string: "wss://studio.example/ws")! } ) - let stream = await client.subscribeBatches("thread.events", reconnect: false, as: String.self) + let subscription = try await client.subscribeBatchesOnCurrentConnection("thread.events", as: String.self) // A unary request is not needed: the connection closes only after overflow. await connection.waitUntilClosed() - var iterator = stream.makeAsyncIterator() + var iterator = subscription.events.makeAsyncIterator() let buffered = try await iterator.next() XCTAssertEqual(buffered, ["first", "second"]) do { @@ -157,7 +159,7 @@ final class WebSocketRPCRaceTests: XCTestCase { } - func testColdSubscriptionRetainsFailedSocketIdentity() async throws { + func testColdBatchSubscriptionRetainsFailedSocketIdentity() async throws { let connection = SubscriptionTrafficConnection(sendsInvalidSubscriptionValue: true) let connector = GatedConnector(connection: connection) let client = WebSocketRPCClient( @@ -165,7 +167,7 @@ final class WebSocketRPCRaceTests: XCTestCase { endpointProvider: { URL(string: "wss://studio.example/ws")! } ) let pending = Task { - try await client.subscribeOnCurrentConnection("thread.events", as: Int.self) + try await client.subscribeBatchesOnCurrentConnection("thread.events", as: Int.self) } await connector.waitUntilConnectStarted() let beforeConnection = await client.currentConnectionID() diff --git a/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift b/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift index 935404736bbf..cfe546a3ef86 100644 --- a/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift @@ -5,6 +5,143 @@ import XCTest @MainActor @available(iOS 18.0, *) final class NativeThreadCatchUpTests: XCTestCase { + func testLegacyReplayPublishesOncePerReceivedBatchAndResumesAfterAppliedEvents() async throws { + for batchSize in [1, 16, 500] { + let fixture = try await CatchUpFixture.make(completionMarker: false) + defer { fixture.cleanUp() } + var requests = fixture.requests.makeAsyncIterator() + var events = fixture.client.events().makeAsyncIterator() + _ = try await fixture.client.loadThread(id: fixture.firstID) + let stream = try await nextThreadRequest(&requests) + try await stream.synchronize() + _ = await messagesBeforeLive(&events, threadID: fixture.firstID) + + let expected = (0..<500).map { "\($0)," }.joined() + let values = (0..<500).map { index in + stream.messageValue( + text: index == 499 ? expected : "\(index),", sequence: index + 3, + streaming: index < 499 + ) + } + for start in stride(from: 0, to: values.count, by: batchSize) { + try await stream.socket.chunk( + id: stream.id, values: Array(values[start...Continuation @@ -1023,6 +1161,8 @@ private actor CatchUpHTTPTransport: HTTPTransport { func setActivities(_ activities: [OrchestrationActivity]) { self.activities = activities } + func setPage(_ page: OrchestrationThreadDetailPage) { self.page = page } + func data(for request: URLRequest) async throws -> (Data, HTTPURLResponse) { let value: JSONValue switch request.url!.path { @@ -1047,7 +1187,7 @@ private actor CatchUpHTTPTransport: HTTPTransport { var thread = snapshot.thread thread.activities = activities value = try .encode(OrchestrationThreadDetailSnapshot( - snapshotSequence: snapshot.snapshotSequence, thread: thread, page: snapshot.page + snapshotSequence: snapshot.snapshotSequence, thread: thread, page: page ?? snapshot.page )) } let response = (try JSONEncoder.t3.encode(value), HTTPURLResponse( @@ -1108,6 +1248,20 @@ private struct CatchUpRequest: Sendable { let payload: JSONValue let socket: CatchUpSocket + func messageValue(text: String, sequence: Int, streaming: Bool = false) -> JSONValue { + .object([ + "kind": .string("event"), "event": .object([ + "type": .string("thread.message-sent"), "sequence": .number(Double(sequence)), + "occurredAt": .string("2026-09-02T12:00:00Z"), "payload": .object([ + "threadId": payload["threadId"]!, "messageId": .string("answer-0"), + "role": .string("assistant"), "text": .string(text), "streaming": .bool(streaming), + "createdAt": .string("2026-09-02T12:00:00Z"), + "updatedAt": .string("2026-09-02T12:00:00Z"), + ]), + ]), + ]) + } + func terminate(_ failure: CatchUpStreamFailure) async throws { switch failure { case .malformed: From 7ef3fe96487237dd3e884796277d11c5dbf2e068 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 13 Sep 2026 21:18:31 -0700 Subject: [PATCH 2/4] fix(swift-ios): keep stale replay from marking details live --- apps/swift-ios/App/NativeFeatureClient.swift | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/swift-ios/App/NativeFeatureClient.swift b/apps/swift-ios/App/NativeFeatureClient.swift index 970880ea65fd..84b50ec6df7c 100644 --- a/apps/swift-ios/App/NativeFeatureClient.swift +++ b/apps/swift-ios/App/NativeFeatureClient.swift @@ -4077,6 +4077,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, ) { // Keep snapshot replacement and page-watermark merges at their event // positions. Ordinary replay batches share one legacy sync publication. + let previousSequence = activeThreadSequence let needsIndividualUpdates = items.count == 1 || activeRawThread == nil || pendingOlderThreadPage != nil || items.contains { item in switch item { @@ -4093,7 +4094,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, synchronizeLegacy: needsIndividualUpdates ) } - if !needsIndividualUpdates, + if !needsIndividualUpdates, activeThreadSequence != previousSequence, serverConfigsByEnvironmentID[route.environmentID]?.threadResumeCompletionMarker != true { markDetailSynchronized(route) } From fabebcd0bd1b168d7bdc0d565e1123efd80b3c63 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 13 Sep 2026 21:19:02 -0700 Subject: [PATCH 3/4] test(swift-ios): ignore stale batch sync receipts --- .../NativeThreadCatchUpTests.swift | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift b/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift index cfe546a3ef86..c18720dae87a 100644 --- a/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift @@ -75,6 +75,27 @@ final class NativeThreadCatchUpTests: XCTestCase { await fixture.client.disconnect() } + func testLegacyStaleBatchDoesNotPublishLiveBeforeNewMessages() async throws { + let fixture = try await CatchUpFixture.make(completionMarker: false) + defer { fixture.cleanUp() } + var requests = fixture.requests.makeAsyncIterator() + var events = fixture.client.events().makeAsyncIterator() + _ = try await fixture.client.loadThread(id: fixture.firstID) + let stream = try await nextThreadRequest(&requests) + try await stream.synchronize() + _ = await messagesBeforeLive(&events, threadID: fixture.firstID) + try await stream.socket.chunk(id: stream.id, values: [ + stream.messageValue(text: "Stale", sequence: 1), + stream.messageValue(text: "Stale", sequence: 2), + ]) + try await stream.socket.chunk(id: stream.id, values: [ + stream.messageValue(text: "New", sequence: 3), + ]) + let messages = await messagesBeforeLive(&events, threadID: fixture.firstID) + XCTAssertEqual(messages, ["New"], "Stale batches must not emit an earlier live receipt.") + await fixture.client.disconnect() + } + func testBatchKeepsRevertSnapshotAndLaterDeltaInOrder() async throws { let fixture = try await CatchUpFixture.make() defer { fixture.cleanUp() } From 343e3ffb62deaedd5e768515cce270d7ee55646d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 13 Sep 2026 21:20:56 -0700 Subject: [PATCH 4/4] test(swift-ios): drain the legacy cold-open sync receipt --- .../swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift b/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift index c18720dae87a..0ad311be737b 100644 --- a/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/NativeThreadCatchUpTests.swift @@ -82,7 +82,7 @@ final class NativeThreadCatchUpTests: XCTestCase { var events = fixture.client.events().makeAsyncIterator() _ = try await fixture.client.loadThread(id: fixture.firstID) let stream = try await nextThreadRequest(&requests) - try await stream.synchronize() + // Legacy cold opens publish live when their HTTP snapshot arrives. _ = await messagesBeforeLive(&events, threadID: fixture.firstID) try await stream.socket.chunk(id: stream.id, values: [ stream.messageValue(text: "Stale", sequence: 1),