Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 93 additions & 0 deletions apps/swift-ios/App/NativeConversationRewind.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
import Foundation

enum NativeConversationRewind {
/// Use checkpoint counts, not visible message indexes. Pages and steering messages
/// do not each represent one completed provider turn.
static func turnCount(before messageID: String, in thread: OrchestrationThread) -> Int? {
guard let index = thread.messages.firstIndex(where: { $0.id == messageID }),
thread.messages[index].role == "user" else { return nil }
let checkpoints = Dictionary(
thread.checkpoints.compactMap { checkpoint in
checkpoint.assistantMessageId.map { ($0, checkpoint.checkpointTurnCount) }
},
uniquingKeysWith: max
)
for message in thread.messages.dropFirst(index + 1) {
if message.role == "user" { return nil }
if let count = checkpoints[message.id] { return max(0, count - 1) }
}
return nil
}

static func isComplete(_ thread: OrchestrationThread, messageID: String, turnCount: Int) -> Bool {
!thread.messages.contains(where: { $0.id == messageID })
&& thread.checkpoints.allSatisfy { $0.checkpointTurnCount <= turnCount }
&& (turnCount == 0
? thread.latestTurn == nil
: thread.checkpoints.contains { $0.turnId == thread.latestTurn?.turnId })
}

/// Command acceptance precedes provider rollback. Wait for its completion event
/// or an authoritative replacement snapshot, including while another thread is open.
static func waitForCompletion(
batches: AsyncThrowingStream<[ThreadStreamItem], Error>,
threadID: String,
messageID: String,
turnCount: Int,
afterSequence: Int,
previousFailureIDs: Set<String>,
timeout: Duration = .seconds(120)
) async throws -> Int {
try await withThrowingTaskGroup(of: Int.self) { group in
group.addTask {
for try await batch in batches {
for item in batch {
switch item {
case .synchronized:
continue
case let .snapshot(snapshot):
guard snapshot.thread.id == threadID,
snapshot.snapshotSequence > afterSequence else { continue }
if isComplete(snapshot.thread, messageID: messageID, turnCount: turnCount) {
return snapshot.snapshotSequence
}
if let failure = snapshot.thread.activities.last(where: {
$0.kind == "checkpoint.revert.failed" && !previousFailureIDs.contains($0.id)
&& $0.payload["turnCount"] == .number(Double(turnCount))
}) {
throw FeatureConversationRewindError(
message: failure.payload["detail"]?.stringValue ?? failure.summary,
didNotRevert: true
)
}
case let .event(event):
guard event["payload"]?["threadId"]?.stringValue == threadID,
case let .number(sequence)? = event["sequence"],
sequence > Double(afterSequence) else { continue }
if event["type"]?.stringValue == "thread.activity-appended",
let activity = event["payload"]?["activity"],
activity["kind"]?.stringValue == "checkpoint.revert.failed",
activity["payload"]?["turnCount"] == .number(Double(turnCount)) {
throw FeatureConversationRewindError(
message: activity["payload"]?["detail"]?.stringValue
?? activity["summary"]?.stringValue ?? "Conversation rewind failed.",
didNotRevert: true
)
}
if event["type"]?.stringValue == "thread.reverted",
event["payload"]?["turnCount"] == .number(Double(turnCount)) { return Int(sequence) }
}
}
}
throw FeatureConversationRewindError(message: "The connection closed before rewind finished. Reload the thread before trying again.")
}
group.addTask {
try await Task.sleep(for: timeout)
throw FeatureConversationRewindError(message: "Timed out waiting for rewind. Reload the thread before trying again.")
}
defer { group.cancelAll() }
guard let sequence = try await group.next() else { throw CancellationError() }
return sequence
}
}
}
150 changes: 150 additions & 0 deletions apps/swift-ios/App/NativeFeatureClient.swift
Original file line number Diff line number Diff line change
Expand Up @@ -2317,6 +2317,156 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging,
try? await refresh(client: route.client)
}

func canRewindConversation(threadID: String, messageID: String) -> Bool {
guard let route = try? threadRoute(for: threadID),
let thread = activeThreadID == route.uiID
? activeRawThread : threadResumeStates[route.uiID]?.thread,
let provider = serverConfigsByEnvironmentID[route.environmentID]?.providers.first(where: {
$0.instanceId == (thread.session?.providerInstanceId ?? thread.modelSelection.instanceId)
}),
provider.supportsConversationRollback != false,
provider.driver != "cursor", provider.driver != "grok" else { return false }
return NativeConversationRewind.turnCount(before: messageID, in: thread) != nil
}

func rewindConversation(
threadID: String, messageID: String,
prepareRecovery: @MainActor (FeatureRevertedMessage) async throws -> Void
) async throws {
let route = try threadRoute(for: threadID)
let generation = environmentGeneration
// The visible page can omit checkpoints. Validate against the whole thread.
let snapshot = try await route.client.threadSnapshot(id: route.wireID)
guard let provider = serverConfigsByEnvironmentID[route.environmentID]?.providers.first(where: {
$0.instanceId == (snapshot.thread.session?.providerInstanceId ?? snapshot.thread.modelSelection.instanceId)
}), provider.supportsConversationRollback != false,
provider.driver != "cursor", provider.driver != "grok" else {
throw FeatureCapabilityUnavailable("Conversation rewind for this provider")
}
guard snapshot.thread.session?.status != "running",
snapshot.thread.session?.status != "starting",
let turnCount = NativeConversationRewind.turnCount(before: messageID, in: snapshot.thread),
let original = snapshot.thread.messages.first(where: { $0.id == messageID }) else {
throw FeatureConversationRewindError(message: "Wait for this turn to finish before rewinding.")
}
let message = mapMessage(original, environmentID: route.environmentID)
let fileStore = ManagedAttachmentFileStore()
var attachments: [FeatureDraftAttachment] = []
var recoveryIsStored = false
defer {
if !recoveryIsStored {
for attachment in attachments {
if let file = attachment.ownedFile {
try? fileStore.removeOwnedFile(fileName: file.fileName)
}
}
}
}
// Rewind removes the old server assets, even when workspace files are kept.
for attachment in message.attachments {
let url = try await attachmentAssetURL(threadID: route.uiID, attachment: attachment)
let (temporaryURL, response) = try await URLSession.shared.download(from: url)
defer { try? FileManager.default.removeItem(at: temporaryURL) }
guard let response = response as? HTTPURLResponse,
(200..<300).contains(response.statusCode) else {
throw FeatureConversationRewindError(message: "Could not save \(attachment.name) before rewind.")
}
let id = UUID()
let file = try await Task.detached {
try fileStore.copyOwnedFile(
from: temporaryURL, attachmentID: id, originalFileName: attachment.name
)
}.value
attachments.append(FeatureDraftAttachment(
id: id, ownedFile: file, filename: attachment.name, mimeType: attachment.mimeType,
source: attachment.source
))
}
try await prepareRecovery(FeatureRevertedMessage(message: message, attachments: attachments))
recoveryIsStored = true
let subscription: (events: AsyncThrowingStream<[ThreadStreamItem], Error>, connectionID: UUID)
do {
try Task.checkCancellation()
guard isKnownClient(route.client, environmentID: route.environmentID, generation: generation) else {
throw CancellationError()
}
subscription = try await route.client.threadEventBatches(
threadID: route.wireID, after: snapshot.snapshotSequence
)
} catch {
throw FeatureConversationRewindError(message: error.localizedDescription, didNotRevert: true)
}
// Keep events that arrive before dispatch replies. The pump owns socket
// cancellation even when dispatch rejects before completion tracking starts.
let buffered = AsyncThrowingStream<[ThreadStreamItem], Error>.makeStream(bufferingPolicy: .bufferingOldest(64))
let pump = Task {
do {
for try await batch in subscription.events {
if case .dropped = buffered.continuation.yield(batch) {
throw FeatureConversationRewindError(message: "Rewind updates fell behind. Reload the thread to check its history.")
}
}
buffered.continuation.finish()
} catch {
buffered.continuation.finish(throwing: error)
}
}
defer {
pump.cancel()
buffered.continuation.finish()
}
do {
let accepted: DispatchResult
do {
accepted = try await route.client.dispatch(OrchestrationCommands.revertConversation(
threadID: route.wireID, turnCount: turnCount
))
} catch let error as RPCError {
if case .remote = error {
throw FeatureConversationRewindError(message: error.localizedDescription, didNotRevert: true)
}
throw error
} catch let error as HTTPError {
if case let .status(code, _, _) = error, (400..<500).contains(code) {
throw FeatureConversationRewindError(message: error.localizedDescription, didNotRevert: true)
}
throw error
}
let receiptSequence = try await NativeConversationRewind.waitForCompletion(
batches: buffered.stream,
threadID: route.wireID,
messageID: messageID,
turnCount: turnCount,
afterSequence: accepted.sequence,
previousFailureIDs: Set(snapshot.thread.activities.filter {
$0.kind == "checkpoint.revert.failed"
}.map(\.id))
)
let current = try await route.client.threadSnapshot(id: route.wireID)
guard current.snapshotSequence >= receiptSequence,
NativeConversationRewind.isComplete(current.thread, messageID: messageID, turnCount: turnCount) else {
throw FeatureConversationRewindError(message: "Could not confirm the final rewind state. The prompt remains saved for recovery.")
}
} catch {
if (error as? FeatureConversationRewindError)?.didNotRevert == true { throw error }
// A lost socket does not mean rollback failed. An HTTP read can
// confirm completion without submitting the destructive command again.
guard let current = try? await route.client.threadSnapshot(id: route.wireID),
current.snapshotSequence > snapshot.snapshotSequence,
NativeConversationRewind.isComplete(current.thread, messageID: messageID, turnCount: turnCount) else {
throw error
}
}
// Do not turn a failed refresh into a failed rewind. The receipt confirms
// that history changed, so the recovered prompt must still reach the draft.
threadResumeStates[route.uiID] = nil
do {
try await refreshThread(id: route.uiID, client: route.client)
} catch {
continuation.yield(.threadSync(id: route.uiID, state: .failed(error.localizedDescription)))
}
}

func resolveApproval(id: String, decision: FeatureApprovalDecision) async throws {
guard let request = approvalRoutes[id] else {
throw NativeFeatureClientError.approvalNotFound
Expand Down
1 change: 1 addition & 0 deletions apps/swift-ios/Core/ServerConfigModels.swift
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,7 @@ public struct ServerProviderSnapshot: Codable, Identifiable, Equatable, Sendable
public var workspaceSnapshots: [ServerProviderWorkspaceSnapshot]? = nil
public var setup: ProviderSetupCapabilities? = nil
public var usageLimits: ServerProviderUsageLimits? = nil
public var supportsConversationRollback: Bool? = nil
}

public enum ServerThreadEnvironmentMode: String, Codable, Equatable, Sendable {
Expand Down
16 changes: 16 additions & 0 deletions apps/swift-ios/Core/T3Client.swift
Original file line number Diff line number Diff line change
Expand Up @@ -2065,6 +2065,22 @@ public enum RPCMethod: String, Sendable {
}

public enum OrchestrationCommands {
/// A distinct command makes older servers reject this action without restoring files.
public static func revertConversation(
threadID: String,
turnCount: Int,
commandID: String = UUID().uuidString,
createdAt: String = now()
) -> JSONValue {
.object([
"type": .string("thread.conversation.revert"),
"commandId": .string(commandID),
"threadId": .string(threadID),
"turnCount": .number(Double(turnCount)),
"createdAt": .string(createdAt),
])
}

public static func createThread(
threadID: String = UUID().uuidString,
projectID: String,
Expand Down
11 changes: 9 additions & 2 deletions apps/swift-ios/Features/Chat/FeatureComposerView.swift
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ struct FeatureComposerUploadStatus {

struct FeatureComposerView: View {
@SwiftUI.Environment(\.scenePhase) private var scenePhase
@SwiftUI.Environment(\.isEnabled) private var isEnabled
@State private var isManuallyExpanded = false
@State private var isAttachmentFlowActive = false
@State private var isModelPickerPresented = false
Expand Down Expand Up @@ -61,6 +62,7 @@ struct FeatureComposerView: View {
private let onRefreshModels: (() async throws -> Void)?
private let draftSaveError: String?
private let onRetryDraftSave: (() -> Void)?
private let onInputPreparationChange: ((Bool) -> Void)?
private let threadSelection: FeatureSelection?
private let materializesDefaultSelection: Bool
private let isSending: Bool
Expand Down Expand Up @@ -112,7 +114,8 @@ struct FeatureComposerView: View {
onRefreshModels: (() async throws -> Void)? = nil,
draftSaveError: String? = nil,
onRetryDraftSave: (() -> Void)? = nil,
context: Binding<OrchestrationMessageContext?> = .constant(nil)
context: Binding<OrchestrationMessageContext?> = .constant(nil),
onInputPreparationChange: ((Bool) -> Void)? = nil
) {
_text = text
_selection = selection
Expand All @@ -124,6 +127,7 @@ struct FeatureComposerView: View {
self.environmentIsConnected = environmentIsConnected
self.attachmentUploads = attachmentUploads
self.attachmentPreferences = attachmentPreferences
self.onInputPreparationChange = onInputPreparationChange
self.onRefreshModels = onRefreshModels
self.draftSaveError = draftSaveError
self.onRetryDraftSave = onRetryDraftSave
Expand Down Expand Up @@ -302,6 +306,9 @@ struct FeatureComposerView: View {
.stroke(T3Colors.inputBorder, lineWidth: 1)
}
.clipShape(composerShape)
.onChange(of: attachmentPreparation.isPreparing || isAttachmentFlowActive || voiceInputController.isBusy, initial: true) { _, busy in
onInputPreparationChange?(busy)
}
.modifier(
FeatureComposerImageDrop(
isEnabled: imagesAllowed && !voiceInputController.isBusy,
Expand Down Expand Up @@ -365,7 +372,7 @@ struct FeatureComposerView: View {
focused: $focused,
placeholder: composerPlaceholder,
acceptsImages: imagesAllowed,
isReadOnly: voiceInputController.isBusy,
isReadOnly: voiceInputController.isBusy || !isEnabled,
skills: powerFeatures.enabledSkills,
selectionRequest: textSelectionRequest,
onSelectionChange: handleTextSelectionChange,
Expand Down
Loading
Loading