diff --git a/Scripts/Fixtures/test-suite-contract-ledger.tsv b/Scripts/Fixtures/test-suite-contract-ledger.tsv index ded3e2ead..c4422667e 100644 --- a/Scripts/Fixtures/test-suite-contract-ledger.tsv +++ b/Scripts/Fixtures/test-suite-contract-ledger.tsv @@ -2860,6 +2860,13 @@ root/RepoPromptTests.WorkspaceRootTargetSeedPlanManifestTests/testManifestAuthen root/RepoPromptTests.WorkspaceRootTargetSeedPlanManifestTests/testManifestRejectsDuplicateByteExactPathWithoutPublishing root Tests/RepoPromptTests/WorkspaceContext/WorkspaceRootTargetSeedPlanManifestTests.swift RepoPromptTests.WorkspaceRootTargetSeedPlanManifestTests testManifestRejectsDuplicateByteExactPathWithoutPublishing WorkspaceContext workspace_root_target_plan.duplicate_rejection duplicate,byte_exact,transactional_publication protocol_negative root_swiftpm routine 2 WorkspaceRootTargetSeedPlanManifestStore Duplicate paths fail without publication. Duplicate paths could make reconciliation ambiguous. 0.002000 temp_directory;spill_io manifest_store test_case+lease_cleanup retain 0 Ephemeral plan schema with inventory compatibility v4; no adapter. root/RepoPromptTests.WorkspaceRootTargetSeedPlanManifestTests/testManifestScaleStreamsAreBounded root Tests/RepoPromptTests/WorkspaceContext/WorkspaceRootTargetSeedPlanManifestTests.swift RepoPromptTests.WorkspaceRootTargetSeedPlanManifestTests testManifestScaleStreamsAreBounded WorkspaceContext target_seed_plan_manifest.routine_boundary manifest_streaming,seeded_inventory,routine_boundary stress root_swiftpm routine 2 WorkspaceRootTargetSeedPlanManifestStore 10K target seed plans stream sorted overlay records with exact footer/reader counts and bounded buffered/resident path bytes. Target seed plan manifests could regress bounded streaming behavior in routine overlay publication. 1.0 routine_scale temporary_artifacts test_case retain 0 routine 10K boundary contract replacing the default 100K run root/RepoPromptTests.WorkspaceRootTargetSeedPlanManifestTests/testManifestScaleStreamsOneHundredThousandOrMillionWhenEnabled root Tests/RepoPromptTests/WorkspaceContext/WorkspaceRootTargetSeedPlanManifestTests.swift RepoPromptTests.WorkspaceRootTargetSeedPlanManifestTests testManifestScaleStreamsOneHundredThousandOrMillionWhenEnabled WorkspaceContext target_seed_plan_manifest.scale_100k_or_million manifest_streaming,seeded_inventory,scale_gate,million_opt_in stress root_swiftpm scale 3 WorkspaceRootTargetSeedPlanManifestStore 100K target seed plans, or 1M when RPCE_RUN_MILLION_ENTRY_TESTS=1, stream sorted records with exact footer counts and bounded buffered/resident path bytes. Target seed plan manifests could buffer too much data or corrupt counts during large overlay publication. 6.0 scale,opt_in temporary_artifacts test_case retain 0 gated behind RPCE_RUN_SCALE_TESTS=1 for 100K; RPCE_RUN_MILLION_ENTRY_TESTS=1 still selects the million-entry lane +root/RepoPromptTests.WorkspaceSaveCoordinatorTests/testCloseBoundaryFlushesPendingDebouncedSave root Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift RepoPromptTests.WorkspaceSaveCoordinatorTests testCloseBoundaryFlushesPendingDebouncedSave Workspaces workspace_save.close_flushes_debounce close_boundary,debounce,durability async_persistence_integration root_swiftpm integration 1 SyntheticWorkspaceSaveFixture Window-close preparation followed by the explicit boundary flush commits a pending debounced workspace save. Closing during the debounce interval could discard the latest workspace state. 0.300000 filesystem;workspace_manager;async_tasks GlobalCustomStorageURL;WorkspaceDiskWriter test_case+tearDown_writer_reset+defer_storage_cleanup retain 0 PR #469 close durability contract +root/RepoPromptTests.WorkspaceSaveCoordinatorTests/testCloseBoundaryRetriesDirtyVersionAfterEarlierFailure root Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift RepoPromptTests.WorkspaceSaveCoordinatorTests testCloseBoundaryRetriesDirtyVersionAfterEarlierFailure Workspaces workspace_save.close_retries_dirty_failure close_boundary,writer_failure,dirty_version,bounded_retry async_persistence_integration root_swiftpm integration 2 SyntheticWorkspaceSaveFixture,InjectedAtomicWriteFailure A failed save leaves its version dirty, and the later close boundary retries and commits after the writer recovers. Idle coordinator cleanup could otherwise hide an unsaved dirty version from termination durability. 0.300000 filesystem;workspace_manager;async_tasks GlobalCustomStorageURL;WorkspaceDiskWriter test_case+tearDown_writer_reset+defer_storage_cleanup retain 0 PR #469 failed-save close retry contract +root/RepoPromptTests.WorkspaceSaveCoordinatorTests/testFailedAtomicWriteIsObservableAndDoesNotAdvanceSavedVersion root Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift RepoPromptTests.WorkspaceSaveCoordinatorTests testFailedAtomicWriteIsObservableAndDoesNotAdvanceSavedVersion Workspaces workspace_save.failure_does_not_advance_version writer_failure,retry,durability_outcome async_persistence_integration root_swiftpm integration 2 SyntheticWorkspaceSaveFixture,InjectedAtomicWriteFailure Injected atomic-write failure returns failed without advancing the saved version; a healthy explicit retry commits the current state. A swallowed write failure could falsely mark in-memory workspace state durable and suppress retry. 0.100000 filesystem;workspace_manager;async_tasks GlobalCustomStorageURL;WorkspaceDiskWriter test_case+tearDown_writer_reset+defer_storage_cleanup retain 0 PR #469 pre-serialization save coalescing failure contract +root/RepoPromptTests.WorkspaceSaveCoordinatorTests/testFailedThenSuccessfulPayloadsReceiveTheirOwnOutcomes root Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift RepoPromptTests.WorkspaceSaveCoordinatorTests testFailedThenSuccessfulPayloadsReceiveTheirOwnOutcomes Workspaces workspace_writer.failed_then_successful_receipts per_payload_receipt,failed_then_successful,drain_failure async_persistence_integration root_swiftpm integration 2 InjectedAtomicWriteFailure,WorkspaceSavePreparationGate A failed payload reports failed, the following payload reports committed, and the drain barrier preserves the failure. A later successful drain result could mask an earlier caller's failed payload. 0.050000 filesystem;async_tasks WorkspaceDiskWriter test_case+tearDown_writer_reset+defer_storage_cleanup retain 0 PR #469 per-payload writer receipt contract +root/RepoPromptTests.WorkspaceSaveCoordinatorTests/testMutationDuringPreparationProducesOneFollowUpAndFlushCommitsNewestState root Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift RepoPromptTests.WorkspaceSaveCoordinatorTests testMutationDuringPreparationProducesOneFollowUpAndFlushCommitsNewestState Workspaces workspace_save.single_followup_for_newer_state preparation_gate,request_sequence,boundary_flush,latest_state async_persistence_integration root_swiftpm integration 2 SyntheticWorkspaceSaveFixture,WorkspaceSavePreparationGate A newer version arriving during blocked preparation causes exactly one follow-up preparation, and boundary flush returns only after that later request is atomically readable. Version-only coalescing could drop the final save-time prompt or selection snapshot, while recursive retry could amplify preparation. 0.300000 filesystem;workspace_manager;async_tasks GlobalCustomStorageURL;WorkspaceDiskWriter test_case+tearDown_writer_reset+defer_storage_cleanup retain 0 PR #469 pre-serialization save coalescing latest-state loop +root/RepoPromptTests.WorkspaceSaveCoordinatorTests/testRapidRequestsCoalesceBeforeRepresentativeWorkspacePreparation root Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift RepoPromptTests.WorkspaceSaveCoordinatorTests testRapidRequestsCoalesceBeforeRepresentativeWorkspacePreparation Workspaces workspace_save.preparation_single_flight rapid_requests,representative_payload,safe_metrics,pre_serialization bounded_performance_contract root_swiftpm integration 1 SyntheticWorkspaceSaveFixture One hundred deferred requests plus a flush produce one preparation and one atomic write for a 25-tab payload while reporting only structural counts. Late-only coalescing could repeatedly decode and encode a compose-tab-heavy workspace and recreate the CPU amplification. 0.300000 filesystem;workspace_manager;async_tasks GlobalCustomStorageURL;WorkspaceDiskWriter test_case+tearDown_writer_reset+defer_storage_cleanup retain 0 PR #469 pre-serialization save coalescing representative fixture +root/RepoPromptTests.WorkspaceSaveCoordinatorTests/testSuccessfulThenFailedPayloadsReceiveTheirOwnOutcomes root Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift RepoPromptTests.WorkspaceSaveCoordinatorTests testSuccessfulThenFailedPayloadsReceiveTheirOwnOutcomes Workspaces workspace_writer.successful_then_failed_receipts per_payload_receipt,successful_then_failed,drain_failure async_persistence_integration root_swiftpm integration 2 InjectedAtomicWriteFailure,WorkspaceSavePreparationGate A committed payload reports committed, the following payload reports failed, and the drain barrier reports the failure. A final failure could be incorrectly attributed to an earlier successful payload or lost at drain completion. 0.050000 filesystem;async_tasks WorkspaceDiskWriter test_case+tearDown_writer_reset+defer_storage_cleanup retain 0 PR #469 per-payload writer receipt contract root/RepoPromptTests.WorkspaceSearchServiceTests/testPathMatchWarmupPreservesLookupBehavior root Tests/RepoPromptTests/WorkspaceContext/Search/WorkspaceSearchServiceTests.swift RepoPromptTests.WorkspaceSearchServiceTests testPathMatchWarmupPreservesLookupBehavior WorkspaceContext unreviewed unreviewed root_swiftpm routine 1 unreviewed unreviewed 0.003000 unreviewed retain_pending_review 0 initial census source line 311 root/RepoPromptTests.WorkspaceSearchServiceTests/testSearchCatalogGenerationChangesOnRootLoadDeltaAndUnload root Tests/RepoPromptTests/WorkspaceContext/Search/WorkspaceSearchServiceTests.swift RepoPromptTests.WorkspaceSearchServiceTests testSearchCatalogGenerationChangesOnRootLoadDeltaAndUnload WorkspaceContext unreviewed unreviewed root_swiftpm routine 1 unreviewed unreviewed 0.005500 unreviewed retain_pending_review 0 initial census source line 15 root/RepoPromptTests.WorkspaceSearchServiceTests/testWorkspaceSearchServiceCatchesUpWhenEventPrecedesSubscription root Tests/RepoPromptTests/WorkspaceContext/Search/WorkspaceSearchServiceTests.swift RepoPromptTests.WorkspaceSearchServiceTests testWorkspaceSearchServiceCatchesUpWhenEventPrecedesSubscription WorkspaceContext unreviewed unreviewed root_swiftpm routine 1 unreviewed unreviewed 0.016000 unreviewed retain_pending_review 0 initial census source line 227 diff --git a/Sources/RepoPrompt/App/AppDelegate.swift b/Sources/RepoPrompt/App/AppDelegate.swift index a671a2940..438f1b269 100644 --- a/Sources/RepoPrompt/App/AppDelegate.swift +++ b/Sources/RepoPrompt/App/AppDelegate.swift @@ -161,7 +161,13 @@ class AppDelegate: NSObject, ObservableObject, NSApplicationDelegate { await WindowStatesManager.shared.persistWindowSessionImmediately(reason: "appShouldTerminate") } - // 3) Shut down agent processes and MCP tools on the main actor WITHOUT blocking. + // 3) Flush already-captured workspace saves before window teardown can release them. + let workspaceSavesFlushed = await WindowStatesManager.shared.flushAllWorkspaceSavesBeforeTermination() + if !workspaceSavesFlushed { + print("💾 App termination exhausted the bounded workspace-save retry.") + } + + // 4) Shut down agent processes and MCP tools on the main actor WITHOUT blocking. // Kill Claude CLI and Codex app-server processes BEFORE stopping MCP servers, // so child processes are terminated and reaped rather than orphaned on quit. await WindowStatesManager.shared.shutdownAllAgentSessions() diff --git a/Sources/RepoPrompt/App/WindowState.swift b/Sources/RepoPrompt/App/WindowState.swift index 7b36e0679..13546090a 100644 --- a/Sources/RepoPrompt/App/WindowState.swift +++ b/Sources/RepoPrompt/App/WindowState.swift @@ -1325,6 +1325,10 @@ class WindowState: ObservableObject { func tearDown() async { beginClose() + let workspaceSaveOutcome = await workspaceManager.flushPendingWorkspaceSavesBeforeClose() + if workspaceSaveOutcome == .failed || workspaceSaveOutcome == .cancelled { + print("💾 Window close could not flush all workspace saves: \(workspaceSaveOutcome)") + } await promptManager.gitViewModel.shutdownForWindowClose() let isAppTermination = WindowStatesManager.shared.isTerminating diff --git a/Sources/RepoPrompt/App/WindowStateManager.swift b/Sources/RepoPrompt/App/WindowStateManager.swift index 56fcdaab8..509af8dda 100644 --- a/Sources/RepoPrompt/App/WindowStateManager.swift +++ b/Sources/RepoPrompt/App/WindowStateManager.swift @@ -911,6 +911,20 @@ class WindowStatesManager: ObservableObject { // MARK: - MCP Server Coordination + /// Establishes a durability barrier for saves already captured by each window. + /// Windows are flushed serially so two views of the same workspace do not prepare + /// competing termination snapshots concurrently. + func flushAllWorkspaceSavesBeforeTermination() async -> Bool { + var allSucceeded = true + for window in allWindows { + let outcome = await window.workspaceManager.flushPendingWorkspaceSavesBeforeClose() + if outcome == .failed || outcome == .cancelled { + allSucceeded = false + } + } + return allSucceeded + } + /// Stops MCP servers in **all** windows (unconditionally). /// During teardown, "running" state can be stale; we just want tools off. func stopAllServers() async { diff --git a/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceManagerViewModel.swift b/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceManagerViewModel.swift index 7c76f4348..1ad4204fb 100644 --- a/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceManagerViewModel.swift +++ b/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceManagerViewModel.swift @@ -275,6 +275,43 @@ class WorkspaceManagerViewModel: ObservableObject { private var stateVersionByWorkspaceID: [UUID: Int] = [:] private var lastSavedVersionByWorkspaceID: [UUID: Int] = [:] + private struct WorkspaceSaveRequest { + let workspaceID: UUID + let requestedVersion: Int + let source: WorkspaceSaveSource + let urgency: WorkspaceSaveUrgency + let requestedAt: ContinuousClock.Instant + let sequence: UInt64 + } + + private struct WorkspaceSaveCoordinatorState { + var pendingLatest: WorkspaceSaveRequest? + var firstDeferredRequestAt: ContinuousClock.Instant? + var worker: Task? + var waiters: [CheckedContinuation] = [] + var coalescedRequestCount = 0 + var staleVersionRetryCount = 0 + var requiresBoundaryFlush = false + var debounceGeneration: UInt64 = 0 + var debounceWaiter: CheckedContinuation? + } + + private var workspaceSaveStateByWorkspaceID: [UUID: WorkspaceSaveCoordinatorState] = [:] + private var nextWorkspaceSaveRequestSequence: UInt64 = 1 + private static let workspaceSaveQuietDelay = Duration.milliseconds(200) + private static let workspaceSaveMaximumDelay = Duration.seconds(1) + + #if DEBUG + private var workspaceSavePreparationCountByWorkspaceID: [UUID: Int] = [:] + private var workspaceSavePreparationGate: (@Sendable (UUID, Int) async -> Void)? + private var latestWorkspaceSavePerformanceSummaryByWorkspaceID: [UUID: WorkspaceSavePerformanceSummary] = [:] + + private nonisolated static func elapsedMilliseconds(since start: ContinuousClock.Instant) -> Double { + let components = (ContinuousClock.now - start).components + return Double(components.seconds) * 1000 + Double(components.attoseconds) / 1_000_000_000_000_000 + } + #endif + @MainActor private static var nextWorkspaceSelectionRevision: UInt64 = 1 @MainActor @@ -1625,6 +1662,9 @@ class WorkspaceManagerViewModel: ObservableObject { tasks.forEach { $0.cancel() } } postCatalogRootWorkTasks.removeAll() + for state in workspaceSaveStateByWorkspaceID.values { + state.worker?.cancel() + } } func prepareForWindowClose() { @@ -1641,9 +1681,50 @@ class WorkspaceManagerViewModel: ObservableObject { tasks.forEach { $0.cancel() } } postCatalogRootWorkTasks.removeAll() + // Workspace saves are drained explicitly by WindowState/AppDelegate close paths. + // Do not cancel them here: this synchronous hook runs before those async barriers. cancellables.removeAll() } + #if DEBUG + func test_scheduleWorkspaceSave(source: WorkspaceSaveSource = "test.deferred") { + scheduleSave(source: source) + } + + func test_flushWorkspaceSave( + workspaceID explicitWorkspaceID: UUID? = nil, + source: WorkspaceSaveSource = "test.flush" + ) async -> WorkspaceSaveCompletion { + guard let workspaceID = explicitWorkspaceID ?? activeWorkspaceID else { return .skippedIneligible } + return await requestWorkspaceSave( + workspaceID: workspaceID, + requestedVersion: stateVersionByWorkspaceID[workspaceID, default: 0], + source: source, + urgency: .flushBeforeBoundary + ) + } + + func test_workspaceSavePreparationCount(workspaceID: UUID) -> Int { + workspaceSavePreparationCountByWorkspaceID[workspaceID, default: 0] + } + + func test_lastSavedVersion(workspaceID: UUID) -> Int? { + lastSavedVersionByWorkspaceID[workspaceID] + } + + func test_workspaceSavePerformanceSummary(workspaceID: UUID) -> WorkspaceSavePerformanceSummary? { + latestWorkspaceSavePerformanceSummaryByWorkspaceID[workspaceID] + } + + func test_setWorkspaceSavePreparationGate(_ gate: (@Sendable (UUID, Int) async -> Void)?) { + workspaceSavePreparationGate = gate + } + + func test_markWorkspaceDirty(workspaceID: UUID) { + bumpStateVersion(for: workspaceID) + } + #endif + #if DEBUG var test_isPollTimerActive: Bool { pollTimer?.isValid == true @@ -4772,12 +4853,12 @@ class WorkspaceManagerViewModel: ObservableObject { if let snapshot { composeTabSnapshotSubject.send(snapshot) } - await saveWorkspaceAsync(source: source) // see change below to avoid big reassign - if let savedWorkspace = workspace(withID: wsID) { - await WorkspaceDiskWriter.shared.flush(url: workspaceFileURL(for: savedWorkspace)) - } - - lastSavedVersionByWorkspaceID[wsID] = cur + _ = await requestWorkspaceSave( + workspaceID: wsID, + requestedVersion: stateVersionByWorkspaceID[wsID, default: cur], + source: source, + urgency: .flushBeforeBoundary + ) } func restoreWorkspaceState( @@ -6699,12 +6780,21 @@ class WorkspaceManagerViewModel: ObservableObject { /// Shared actor for serialized workspace disk writes across all windows actor WorkspaceDiskWriter { + enum WriteOutcome: Equatable { + case committed + case superseded + case failed + case idle + } + // MARK: internal model private struct Pending { var newestData: Data var newestMetadata: WorkspaceSavePayloadMetadata? var newestLifecycleCorrelation: EditFlowPerf.LifecycleCorrelation? + var newestReceiptID: UUID? + var supersededReceiptIDs: [UUID] = [] var task: Task? } @@ -6723,11 +6813,16 @@ class WorkspaceManagerViewModel: ObservableObject { } private var pendingByURL: [URL: Pending] = [:] - private var waitersByURL: [URL: [CheckedContinuation]] = [:] + private var flushWaitersByURL: [URL: [CheckedContinuation]] = [:] + private var receiptWaitersByID: [UUID: CheckedContinuation] = [:] + private var drainOutcomeByURL: [URL: WriteOutcome] = [:] private var latestSelectionByWorkspaceTab: [WorkspaceTabSelectionKey: LatestSelectionRecord] = [:] private var lastWrittenSelectionRevisionByWorkspaceTab: [WorkspaceTabSelectionKey: UInt64] = [:] #if DEBUG private var atomicWriteGateForTesting: (@Sendable () async -> Void)? + private var failAtomicWriteForTesting = false + private var atomicWriteAttemptCountByURLForTesting: [URL: Int] = [:] + private var failAtomicWriteAttemptForTesting: (url: URL, attempt: Int)? #endif // MARK: public API @@ -6742,7 +6837,12 @@ class WorkspaceManagerViewModel: ObservableObject { enqueue(data: data, url: url, metadata: metadata) } - private func enqueue(data: Data, url: URL, metadata: WorkspaceSavePayloadMetadata?) { + private func enqueue( + data: Data, + url: URL, + metadata: WorkspaceSavePayloadMetadata?, + receiptID: UUID? = nil + ) { let lifecycleCorrelation = EditFlowPerf.currentLifecycleCorrelation WorkspaceSaveTracer.event("workspaceSave.enqueue", metadata: metadata, url: url) recordLatestSelectionIfNeeded(metadata) @@ -6753,18 +6853,30 @@ class WorkspaceManagerViewModel: ObservableObject { let existingMetadata = pending.newestMetadata, metadata.activeSelectionRevision > existingMetadata.activeSelectionRevision { + if let oldReceiptID = pending.newestReceiptID { + pending.supersededReceiptIDs.append(oldReceiptID) + } pending.newestData = data pending.newestMetadata = metadata pending.newestLifecycleCorrelation = lifecycleCorrelation ?? pending.newestLifecycleCorrelation + pending.newestReceiptID = receiptID decision = "replacedExistingNewerSelectionRevision" } else if Self.shouldKeepExistingWorkspacePayload(existing: pending.newestData, incoming: data, url: url) { decision = "keptExistingNewerDate" + if let receiptID { + pending.supersededReceiptIDs.append(receiptID) + } WorkspaceSaveTracer.event("workspaceSave.coalesce", metadata: metadata, url: url, extra: ["decision": decision]) + pendingByURL[url] = pending return } else { + if let oldReceiptID = pending.newestReceiptID { + pending.supersededReceiptIDs.append(oldReceiptID) + } pending.newestData = data pending.newestMetadata = metadata pending.newestLifecycleCorrelation = lifecycleCorrelation ?? pending.newestLifecycleCorrelation + pending.newestReceiptID = receiptID decision = "storedAsNewest" } pendingByURL[url] = pending @@ -6776,14 +6888,31 @@ class WorkspaceManagerViewModel: ObservableObject { newestData: data, newestMetadata: metadata, newestLifecycleCorrelation: lifecycleCorrelation, + newestReceiptID: receiptID, task: nil ) + drainOutcomeByURL[url] = .superseded runNext(for: url) } - func enqueueAndWait(data: Data, url: URL) async { - enqueue(data: data, url: url) - await flush(url: url) + func enqueueAndWait(data: Data, url: URL) async -> WriteOutcome { + await withCheckedContinuation { continuation in + let receiptID = UUID() + receiptWaitersByID[receiptID] = continuation + enqueue(data: data, url: url, metadata: nil, receiptID: receiptID) + } + } + + func enqueueWorkspaceAndWait( + data: Data, + url: URL, + metadata: WorkspaceSavePayloadMetadata + ) async -> WriteOutcome { + await withCheckedContinuation { continuation in + let receiptID = UUID() + receiptWaitersByID[receiptID] = continuation + enqueue(data: data, url: url, metadata: metadata, receiptID: receiptID) + } } func writeNormalizationIfUnchanged( @@ -6819,22 +6948,25 @@ class WorkspaceManagerViewModel: ObservableObject { } } - func flush(url: URL) async { - if pendingByURL[url] == nil { return } + func flush(url: URL) async -> WriteOutcome { + if pendingByURL[url] == nil { + return .idle + } let lifecycleCorrelation = EditFlowPerf.currentLifecycleCorrelation let flushState = EditFlowPerf.begin(EditFlowPerf.Stage.WorkspaceDurability.flushWait) EditFlowPerf.lifecycleEvent( EditFlowPerf.Lifecycle.WorkspaceDurability.flushBegan, correlation: lifecycleCorrelation ) - await withCheckedContinuation { (cont: CheckedContinuation) in - waitersByURL[url, default: []].append(cont) + let outcome = await withCheckedContinuation { (cont: CheckedContinuation) in + flushWaitersByURL[url, default: []].append(cont) } EditFlowPerf.lifecycleEvent( EditFlowPerf.Lifecycle.WorkspaceDurability.flushEnded, correlation: lifecycleCorrelation ) EditFlowPerf.end(EditFlowPerf.Stage.WorkspaceDurability.flushWait, flushState) + return outcome } #if DEBUG @@ -6842,6 +6974,19 @@ class WorkspaceManagerViewModel: ObservableObject { atomicWriteGateForTesting = gate } + func setFailAtomicWriteForTesting(_ fail: Bool) { + failAtomicWriteForTesting = fail + } + + func setFailAtomicWriteForTesting( + url: URL, + afterAdditionalAttempts additionalAttempts: Int? + ) { + failAtomicWriteAttemptForTesting = additionalAttempts.map { + (url, atomicWriteAttemptCountByURLForTesting[url, default: 0] + $0) + } + } + func removeAllForTesting() { for (_, pending) in pendingByURL { pending.task?.cancel() @@ -6850,10 +6995,19 @@ class WorkspaceManagerViewModel: ObservableObject { latestSelectionByWorkspaceTab.removeAll() lastWrittenSelectionRevisionByWorkspaceTab.removeAll() atomicWriteGateForTesting = nil - let allWaiters = waitersByURL.values.flatMap(\.self) - waitersByURL.removeAll() - for waiter in allWaiters { - waiter.resume() + failAtomicWriteForTesting = false + atomicWriteAttemptCountByURLForTesting.removeAll() + failAtomicWriteAttemptForTesting = nil + drainOutcomeByURL.removeAll() + let allFlushWaiters = flushWaitersByURL.values.flatMap(\.self) + flushWaitersByURL.removeAll() + for waiter in allFlushWaiters { + waiter.resume(returning: .failed) + } + let receiptWaiters = Array(receiptWaitersByID.values) + receiptWaitersByID.removeAll() + for waiter in receiptWaiters { + waiter.resume(returning: .failed) } } #endif @@ -6894,7 +7048,6 @@ class WorkspaceManagerViewModel: ObservableObject { "workspaceDiskWriter.skipStaleCoalescedPayload", fields: [ "workspaceID": WorkspaceRestorePerfLog.shortID(incomingWorkspace.id), - "workspaceName": incomingWorkspace.name, "url": url.lastPathComponent ] ) @@ -6949,7 +7102,7 @@ class WorkspaceManagerViewModel: ObservableObject { extra: [ "latestSelectionRevision": "\(latestRevision)", "lastWrittenSelectionRevision": "\(lastWrittenRevision)", - "latestPayloadID": latestMetadata.payloadID.uuidString + "latestPayloadID": WorkspaceRestorePerfLog.shortID(latestMetadata.payloadID) ] ) return EffectiveWritePayload(data: encoded, metadata: latestMetadata, selectionKey: key, effectiveSelectionRevision: latestRevision, shouldWrite: true) @@ -6975,7 +7128,7 @@ class WorkspaceManagerViewModel: ObservableObject { extra: [ "incomingSelectionRevision": "\(incomingRevision)", "latestSelectionRevision": "\(latestRevision)", - "latestPayloadID": latestMetadata.payloadID.uuidString + "latestPayloadID": WorkspaceRestorePerfLog.shortID(latestMetadata.payloadID) ] ) return EffectiveWritePayload(data: encoded, metadata: latestMetadata, selectionKey: key, effectiveSelectionRevision: latestRevision, shouldWrite: true) @@ -7004,14 +7157,25 @@ class WorkspaceManagerViewModel: ObservableObject { let payload = slot.newestData let metadata = slot.newestMetadata let lifecycleCorrelation = slot.newestLifecycleCorrelation + let receiptID = slot.newestReceiptID + let supersededReceiptIDs = slot.supersededReceiptIDs slot.newestData = Data() slot.newestMetadata = nil slot.newestLifecycleCorrelation = nil + slot.newestReceiptID = nil + slot.supersededReceiptIDs = [] pendingByURL[url] = slot let latestRecord = metadata?.selectionKey.flatMap { latestSelectionByWorkspaceTab[$0] } let lastWrittenRevision = metadata?.selectionKey.map { lastWrittenSelectionRevisionByWorkspaceTab[$0, default: 0] } ?? 0 #if DEBUG let atomicWriteGateForTesting = atomicWriteGateForTesting + atomicWriteAttemptCountByURLForTesting[url, default: 0] += 1 + let atomicWriteAttempt = atomicWriteAttemptCountByURLForTesting[url, default: 0] + let failAtomicWriteForTesting = failAtomicWriteForTesting || + ( + failAtomicWriteAttemptForTesting?.url == url && + failAtomicWriteAttemptForTesting?.attempt == atomicWriteAttempt + ) #endif let task = Task.detached(priority: .utility) { [weak self] in @@ -7028,6 +7192,9 @@ class WorkspaceManagerViewModel: ObservableObject { if effective.shouldWrite { #if DEBUG await atomicWriteGateForTesting?() + if failAtomicWriteForTesting { + throw CocoaError(.fileWriteUnknown) + } #endif let writeState = EditFlowPerf.begin(EditFlowPerf.Stage.WorkspaceDurability.atomicWrite) EditFlowPerf.lifecycleEvent( @@ -7055,7 +7222,19 @@ class WorkspaceManagerViewModel: ObservableObject { print("💾 Write failed \(url.lastPathComponent): \(error)") } WorkspaceSaveTracer.event("workspaceSave.write.finish", metadata: effective.metadata, url: url, extra: ["writeSucceeded": "\(writeSucceeded)"]) - await self?.writerFinished(for: url, effective: effective, writeSucceeded: writeSucceeded) + let outcome: WriteOutcome = if effective.shouldWrite { + writeSucceeded ? .committed : .failed + } else { + .superseded + } + await self?.writerFinished( + for: url, + receiptID: receiptID, + supersededReceiptIDs: supersededReceiptIDs, + effective: effective, + writeSucceeded: writeSucceeded, + outcome: outcome + ) } if var current = pendingByURL[url] { current.task = task @@ -7063,7 +7242,24 @@ class WorkspaceManagerViewModel: ObservableObject { } } - private func writerFinished(for url: URL, effective: EffectiveWritePayload, writeSucceeded: Bool) { + private func writerFinished( + for url: URL, + receiptID: UUID?, + supersededReceiptIDs: [UUID], + effective: EffectiveWritePayload, + writeSucceeded: Bool, + outcome: WriteOutcome + ) { + resolveReceipt(receiptID, with: outcome) + let supersededOutcome: WriteOutcome = (outcome == .failed) ? .failed : .superseded + for supersededReceiptID in supersededReceiptIDs { + resolveReceipt(supersededReceiptID, with: supersededOutcome) + } + guard var slot = pendingByURL[url] else { return } + drainOutcomeByURL[url] = Self.combinedDrainOutcome( + drainOutcomeByURL[url] ?? .superseded, + outcome + ) if writeSucceeded, let key = effective.selectionKey, effective.effectiveSelectionRevision > 0 @@ -7073,12 +7269,18 @@ class WorkspaceManagerViewModel: ObservableObject { effective.effectiveSelectionRevision ) } - guard var slot = pendingByURL[url] else { return } if slot.newestData.isEmpty { + // No follow-up write; any receipts that were superseded by the just-finished + // write are resolved here. + for supersededReceiptID in slot.supersededReceiptIDs { + resolveReceipt(supersededReceiptID, with: supersededOutcome) + } + slot.supersededReceiptIDs = [] pendingByURL.removeValue(forKey: url) - if let waiters = waitersByURL.removeValue(forKey: url) { - for w in waiters { - w.resume() + let drainOutcome = drainOutcomeByURL.removeValue(forKey: url) ?? outcome + if let waiters = flushWaitersByURL.removeValue(forKey: url) { + for waiter in waiters { + waiter.resume(returning: drainOutcome) } } } else { @@ -7087,28 +7289,316 @@ class WorkspaceManagerViewModel: ObservableObject { runNext(for: url) } } + + private func resolveReceipt(_ receiptID: UUID?, with outcome: WriteOutcome) { + guard let receiptID, + let waiter = receiptWaitersByID.removeValue(forKey: receiptID) + else { return } + waiter.resume(returning: outcome) + } + + private static func combinedDrainOutcome(_ current: WriteOutcome, _ next: WriteOutcome) -> WriteOutcome { + if current == .failed || next == .failed { + return .failed + } + if current == .committed || next == .committed { + return .committed + } + return .superseded + } } // MARK: - Save/Load Single Workspace + private func requestWorkspaceSave( + workspaceID: UUID, + requestedVersion: Int, + source: WorkspaceSaveSource, + urgency: WorkspaceSaveUrgency + ) async -> WorkspaceSaveCompletion { + await withCheckedContinuation { continuation in + enqueueWorkspaceSave( + workspaceID: workspaceID, + requestedVersion: requestedVersion, + source: source, + urgency: urgency, + boundaryWaiter: continuation + ) + } + } + + private func enqueueWorkspaceSave( + workspaceID: UUID, + requestedVersion: Int, + source: WorkspaceSaveSource, + urgency: WorkspaceSaveUrgency, + boundaryWaiter: CheckedContinuation? = nil + ) { + let now = ContinuousClock.now + let sequence = nextWorkspaceSaveRequestSequence + nextWorkspaceSaveRequestSequence &+= 1 + var state = workspaceSaveStateByWorkspaceID[workspaceID] ?? WorkspaceSaveCoordinatorState() + var debounceWaiterToResume: CheckedContinuation? + if state.pendingLatest != nil || state.worker != nil { + state.coalescedRequestCount += 1 + } + let existing = state.pendingLatest + if urgency == .flushBeforeBoundary { + state.requiresBoundaryFlush = true + debounceWaiterToResume = state.debounceWaiter + state.debounceWaiter = nil + state.debounceGeneration &+= 1 + } + let effectiveUrgency: WorkspaceSaveUrgency = if state.requiresBoundaryFlush || existing?.urgency == .flushBeforeBoundary { + .flushBeforeBoundary + } else { + .deferred + } + if existing == nil || requestedVersion >= existing?.requestedVersion ?? Int.min { + state.pendingLatest = WorkspaceSaveRequest( + workspaceID: workspaceID, + requestedVersion: requestedVersion, + source: source, + urgency: effectiveUrgency, + requestedAt: now, + sequence: sequence + ) + } else if effectiveUrgency == .flushBeforeBoundary, let existing { + state.pendingLatest = WorkspaceSaveRequest( + workspaceID: existing.workspaceID, + requestedVersion: existing.requestedVersion, + source: existing.source, + urgency: .flushBeforeBoundary, + requestedAt: now, + sequence: sequence + ) + } + if state.firstDeferredRequestAt == nil, effectiveUrgency == .deferred { + state.firstDeferredRequestAt = now + } + if let boundaryWaiter { + state.waiters.append(boundaryWaiter) + } + if state.worker == nil { + state.worker = Task { @MainActor [weak self] in + await self?.runWorkspaceSaveLoop(workspaceID: workspaceID) + } + } + workspaceSaveStateByWorkspaceID[workspaceID] = state + debounceWaiterToResume?.resume() + } + + private func resumeWorkspaceSaveDebounce(workspaceID: UUID, generation: UInt64) { + guard var state = workspaceSaveStateByWorkspaceID[workspaceID], + state.debounceGeneration == generation, + let waiter = state.debounceWaiter + else { return } + state.debounceWaiter = nil + workspaceSaveStateByWorkspaceID[workspaceID] = state + waiter.resume() + } + + private func runWorkspaceSaveLoop(workspaceID: UUID) async { + var finalOutcome: WorkspaceSaveCompletion = .cancelled + while !Task.isCancelled { + guard var state = workspaceSaveStateByWorkspaceID[workspaceID], + let request = state.pendingLatest + else { break } + + if request.urgency == .deferred, let firstRequestAt = state.firstDeferredRequestAt { + let now = ContinuousClock.now + let quietRemaining = now.duration(to: request.requestedAt.advanced(by: Self.workspaceSaveQuietDelay)) + let maximumRemaining = now.duration(to: firstRequestAt.advanced(by: Self.workspaceSaveMaximumDelay)) + let delay = min(quietRemaining, maximumRemaining) + if delay > .zero { + let generation = state.debounceGeneration &+ 1 + await withCheckedContinuation { continuation in + state.debounceGeneration = generation + state.debounceWaiter = continuation + workspaceSaveStateByWorkspaceID[workspaceID] = state + Task { @MainActor [weak self] in + try? await Task.sleep(for: delay) + self?.resumeWorkspaceSaveDebounce( + workspaceID: workspaceID, + generation: generation + ) + } + } + continue + } + } + + state.pendingLatest = nil + state.firstDeferredRequestAt = nil + workspaceSaveStateByWorkspaceID[workspaceID] = state + finalOutcome = await performWorkspaceSavePreparation(request) + + guard var latestState = workspaceSaveStateByWorkspaceID[workspaceID] else { break } + switch finalOutcome { + case let .committed(version): + // A durable write of `version` just completed, so record it as the + // newest saved version immediately. This must happen even when a + // newer pending version is re-armed below; otherwise an interrupted + // loop (cancelled worker, or a guard exit before the newer version + // is processed) can resume a boundary waiter with `.committed(version)` + // while `lastSavedVersion` lags behind. That mismatch makes the close + // boundary report a version it never recorded as saved. + lastSavedVersionByWorkspaceID[workspaceID] = max( + lastSavedVersionByWorkspaceID[workspaceID, default: -1], + version + ) + let latestVersion = stateVersionByWorkspaceID[workspaceID, default: request.requestedVersion] + if let pending = latestState.pendingLatest, + pending.requestedVersion < version || + (pending.requestedVersion == version && pending.sequence <= request.sequence) + { + latestState.pendingLatest = nil + latestState.firstDeferredRequestAt = nil + } + if latestVersion > request.requestedVersion, + latestState.pendingLatest?.requestedVersion ?? Int.min < latestVersion + { + latestState.pendingLatest = WorkspaceSaveRequest( + workspaceID: workspaceID, + requestedVersion: latestVersion, + source: request.source, + urgency: latestState.requiresBoundaryFlush ? .flushBeforeBoundary : request.urgency, + requestedAt: ContinuousClock.now, + sequence: nextWorkspaceSaveRequestSequence + ) + nextWorkspaceSaveRequestSequence &+= 1 + } + case let .superseded(version): + let latestVersion = stateVersionByWorkspaceID[workspaceID, default: request.requestedVersion] + if latestVersion > request.requestedVersion { + latestState.staleVersionRetryCount += 1 + } + if latestVersion > request.requestedVersion, + latestState.pendingLatest?.requestedVersion ?? Int.min < latestVersion + { + latestState.pendingLatest = WorkspaceSaveRequest( + workspaceID: workspaceID, + requestedVersion: latestVersion, + source: request.source, + urgency: latestState.requiresBoundaryFlush ? .flushBeforeBoundary : request.urgency, + requestedAt: ContinuousClock.now, + sequence: nextWorkspaceSaveRequestSequence + ) + nextWorkspaceSaveRequestSequence &+= 1 + } else if latestVersion == request.requestedVersion, + latestState.pendingLatest == nil + { + lastSavedVersionByWorkspaceID[workspaceID] = max( + lastSavedVersionByWorkspaceID[workspaceID, default: -1], + version + ) + } + case .skippedIneligible, .failed, .cancelled: + break + } + workspaceSaveStateByWorkspaceID[workspaceID] = latestState + if latestState.pendingLatest != nil { + continue + } + break + } + + guard let state = workspaceSaveStateByWorkspaceID.removeValue(forKey: workspaceID) else { return } + for waiter in state.waiters { + waiter.resume(returning: finalOutcome) + } + } + + func flushPendingWorkspaceSavesBeforeClose() async -> WorkspaceSaveCompletion { + let activeWorkspaceIDs = workspaceSaveStateByWorkspaceID.compactMap { workspaceID, state in + state.pendingLatest != nil || state.worker != nil ? workspaceID : nil + } + let dirtyWorkspaceIDs: [UUID] = stateVersionByWorkspaceID.compactMap { workspaceID, version in + guard workspace(withID: workspaceID) != nil, + lastSavedVersionByWorkspaceID[workspaceID] != version + else { return nil } + return workspaceID + } + let workspaceIDs = Set(activeWorkspaceIDs).union(dirtyWorkspaceIDs) + guard !workspaceIDs.isEmpty else { return .skippedIneligible } + + var result: WorkspaceSaveCompletion = .skippedIneligible + for workspaceID in workspaceIDs { + var outcome: WorkspaceSaveCompletion = .failed + for attempt in 0 ..< 2 { + outcome = await requestWorkspaceSave( + workspaceID: workspaceID, + requestedVersion: stateVersionByWorkspaceID[workspaceID, default: 0], + source: attempt == 0 ? "windowCloseBoundary" : "windowCloseBoundaryRetry", + urgency: .flushBeforeBoundary + ) + if outcome != .failed, outcome != .cancelled { + break + } + } + result = Self.combinedWorkspaceSaveCompletion(result, outcome) + } + return result + } + + private static func combinedWorkspaceSaveCompletion( + _ current: WorkspaceSaveCompletion, + _ next: WorkspaceSaveCompletion + ) -> WorkspaceSaveCompletion { + switch (current, next) { + case (.failed, _), (_, .failed): + .failed + case (.cancelled, _), (_, .cancelled): + .cancelled + case (.skippedIneligible, _): + next + case let (.committed(currentVersion), .committed(nextVersion)): + .committed(version: max(currentVersion, nextVersion)) + case let (.committed(version), _), let (_, .committed(version)): + .committed(version: version) + case let (.superseded(currentVersion), .superseded(nextVersion)): + .superseded(version: max(currentVersion, nextVersion)) + case let (.superseded(version), _), let (_, .superseded(version)): + .superseded(version: version) + case (.skippedIneligible, .skippedIneligible): + .skippedIneligible + } + } + private func saveWorkspaceAsync(source: WorkspaceSaveSource = .saveWorkspaceAsync) async { - guard let active = activeWorkspace, - let idx = workspaces.firstIndex(where: { $0.id == active.id }) - else { + guard let active = activeWorkspace else { print("No active workspace to save.") return } + _ = await requestWorkspaceSave( + workspaceID: active.id, + requestedVersion: stateVersionByWorkspaceID[active.id, default: 0], + source: source, + urgency: .flushBeforeBoundary + ) + } + + private func performWorkspaceSavePreparation(_ request: WorkspaceSaveRequest) async -> WorkspaceSaveCompletion { + guard let idx = workspaces.firstIndex(where: { $0.id == request.workspaceID }) + else { + return .skippedIneligible + } let current = workspaces[idx] - let capturedStateVersion = stateVersionByWorkspaceID[current.id, default: 0] + let capturedStateVersion = request.requestedVersion let baseRoot = currentBaseRoot let customStoragePath = current.customStoragePath let workspaceDirName = directoryName(for: current) let lastSyncedRepoPaths = lastSyncedRepoPathsByWorkspaceID[current.id] - await WorkspaceDiskWriter.shared.flush(url: workspaceFileURL(for: current)) + _ = await WorkspaceDiskWriter.shared.flush(url: workspaceFileURL(for: current)) + #if DEBUG + let preparationStartedAt = ContinuousClock.now + workspaceSavePreparationCountByWorkspaceID[current.id, default: 0] += 1 + await workspaceSavePreparationGate?(current.id, capturedStateVersion) + #endif do { - let (merged, data, indexFieldsChanged, preservedDiskRepoPaths, url) = try await Task.detached(priority: .utility) { + let (merged, data, indexFieldsChanged, preservedDiskRepoPaths, url, diskDecodeDurationMS, encodeDurationMS) = try await Task.detached(priority: .utility) { let workspaceDir: URL = if let customStoragePath { customStoragePath } else { @@ -7134,11 +7624,19 @@ class WorkspaceManagerViewModel: ObservableObject { } let url = workspaceDir.appendingPathComponent("workspace.json") + #if DEBUG + let diskDecodeStartedAt = ContinuousClock.now + #endif let diskWorkspace: WorkspaceModel? = if fm.fileExists(atPath: url.path) { try? Self.loadWorkspaceFromFile(at: url) } else { nil } + #if DEBUG + let diskDecodeDurationMS = Self.elapsedMilliseconds(since: diskDecodeStartedAt) + #else + let diskDecodeDurationMS = 0.0 + #endif let mergeResult = Self.workspaceForSavePreservingDiskRepoPaths( current: current, @@ -7146,14 +7644,22 @@ class WorkspaceManagerViewModel: ObservableObject { lastSyncedRepoPaths: lastSyncedRepoPaths ) let merged = mergeResult.workspace + #if DEBUG + let encodeStartedAt = ContinuousClock.now + #endif let data = try JSONEncoder().encode(merged) + #if DEBUG + let encodeDurationMS = Self.elapsedMilliseconds(since: encodeStartedAt) + #else + let encodeDurationMS = 0.0 + #endif let indexFieldsChanged = (merged.name != current.name) || (merged.customStoragePath != current.customStoragePath) || (merged.isSystemWorkspace != current.isSystemWorkspace) || (merged.isHiddenInMenus != current.isHiddenInMenus) - return (merged, data, indexFieldsChanged, mergeResult.preservedDiskRepoPaths, url) + return (merged, data, indexFieldsChanged, mergeResult.preservedDiskRepoPaths, url, diskDecodeDurationMS, encodeDurationMS) }.value let latestStateVersion = stateVersionByWorkspaceID[current.id, default: 0] @@ -7168,8 +7674,7 @@ class WorkspaceManagerViewModel: ObservableObject { ] ) #endif - await saveWorkspaceAsync(source: source) - return + return .superseded(version: latestStateVersion) } // IMPORTANT: Do NOT assign `workspaces[idx] = merged` here. @@ -7186,25 +7691,75 @@ class WorkspaceManagerViewModel: ObservableObject { if preservedDiskRepoPaths { workspaces[idx].repoPaths = merged.repoPaths } - recordRepoPathBaseline(for: merged) - - let metadata = workspaceSaveMetadata(for: merged, source: source) + let metadata = workspaceSaveMetadata(for: merged, source: request.source) WorkspaceFileDecodeCache.shared.invalidate(url: url) - await WorkspaceDiskWriter.shared.enqueueWorkspace(data: data, url: url, metadata: metadata) + let writeOutcome = await WorkspaceDiskWriter.shared.enqueueWorkspaceAndWait( + data: data, + url: url, + metadata: metadata + ) - if indexFieldsChanged { - await rebuildAndSaveIndexAsync() + // The workspace can change while the actor is awaiting the atomic write. + // A committed receipt only certifies the captured payload; it must not let + // the coordinator declare a newer in-memory version durable. + let postWriteStateVersion = stateVersionByWorkspaceID[current.id, default: capturedStateVersion] + if writeOutcome == .committed, + postWriteStateVersion != capturedStateVersion + { + return .superseded(version: postWriteStateVersion) + } + + if writeOutcome == .committed { + recordRepoPathBaseline(for: merged) + if indexFieldsChanged { + await rebuildAndSaveIndexAsync() + } + } + #if DEBUG + let selection = metadata.selectionSummary + let coordinatorState = workspaceSaveStateByWorkspaceID[current.id] + let summary = WorkspaceSavePerformanceSummary( + source: request.source, + payloadByteCount: data.count, + composeTabCount: merged.composeTabs.count, + selectedPathCount: selection.selectedPaths, + sliceFileCount: selection.sliceFiles, + sliceRangeCount: selection.sliceRanges, + preparationDurationMS: Self.elapsedMilliseconds(since: preparationStartedAt), + diskDecodeDurationMS: diskDecodeDurationMS, + encodeDurationMS: encodeDurationMS, + staleVersionRetryCount: coordinatorState?.staleVersionRetryCount ?? 0, + coalescedRequestCount: coordinatorState?.coalescedRequestCount ?? 0, + atomicWriteCount: writeOutcome == .committed ? 1 : 0 + ) + latestWorkspaceSavePerformanceSummaryByWorkspaceID[current.id] = summary + WorkspaceRestorePerfLog.event("workspaceSave.performance", fields: summary.traceFields) + #endif + switch writeOutcome { + case .committed: + return .committed(version: capturedStateVersion) + case .superseded: + return .superseded(version: capturedStateVersion) + case .failed: + return .failed + case .idle: + return .failed } } catch { print("💾 Failed to serialize workspace: \(error)") + return error is CancellationError ? .cancelled : .failed } } private func scheduleSave(source: WorkspaceSaveSource) { - // Use async version - Task { - await saveWorkspaceAsync(source: source) - } + guard let active = activeWorkspace else { return } + let workspaceID = active.id + enqueueWorkspaceSave( + workspaceID: workspaceID, + requestedVersion: stateVersionByWorkspaceID[workspaceID, default: 0], + source: source, + urgency: .deferred + ) } // File I/O queue removed - now using async/await for non-blocking operations @@ -8060,8 +8615,8 @@ class WorkspaceManagerViewModel: ObservableObject { return } workspaces[index].currentPromptText = newText - scheduleSave(source: .updatePromptText) bumpStateVersion(for: active.id) + scheduleSave(source: .updatePromptText) } func updateSelectedMetaPromptIDs(_ newIDs: [UUID]) { @@ -8071,8 +8626,8 @@ class WorkspaceManagerViewModel: ObservableObject { return } workspaces[index].selectedMetaPromptIDs = newIDs - scheduleSave(source: .updateSelectedMetaPromptIDs) bumpStateVersion(for: active.id) + scheduleSave(source: .updateSelectedMetaPromptIDs) } /// Sets a workspace's ephemeral property by ID diff --git a/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceSaveDiagnostics.swift b/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceSaveDiagnostics.swift index c6f8cb835..11debf300 100644 --- a/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceSaveDiagnostics.swift +++ b/Sources/RepoPrompt/Features/Workspaces/ViewModels/WorkspaceSaveDiagnostics.swift @@ -52,6 +52,53 @@ struct WorkspaceSaveSource: Equatable, Hashable, ExpressibleByStringLiteral, Cus static let directUnknown = WorkspaceSaveSource("directUnknown") } +enum WorkspaceSaveUrgency: Equatable { + case deferred + case flushBeforeBoundary +} + +enum WorkspaceSaveCompletion: Equatable { + case committed(version: Int) + case superseded(version: Int) + case skippedIneligible + case failed + case cancelled +} + +#if DEBUG + struct WorkspaceSavePerformanceSummary: Equatable { + let source: WorkspaceSaveSource + let payloadByteCount: Int + let composeTabCount: Int + let selectedPathCount: Int + let sliceFileCount: Int + let sliceRangeCount: Int + let preparationDurationMS: Double + let diskDecodeDurationMS: Double + let encodeDurationMS: Double + let staleVersionRetryCount: Int + let coalescedRequestCount: Int + let atomicWriteCount: Int + + var traceFields: [String: String] { + [ + "source": source.rawValue, + "payloadByteCount": "\(payloadByteCount)", + "composeTabCount": "\(composeTabCount)", + "selectedPathCount": "\(selectedPathCount)", + "sliceFileCount": "\(sliceFileCount)", + "sliceRangeCount": "\(sliceRangeCount)", + "preparationDurationMS": String(format: "%.3f", preparationDurationMS), + "diskDecodeDurationMS": String(format: "%.3f", diskDecodeDurationMS), + "encodeDurationMS": String(format: "%.3f", encodeDurationMS), + "staleVersionRetryCount": "\(staleVersionRetryCount)", + "coalescedRequestCount": "\(coalescedRequestCount)", + "atomicWriteCount": "\(atomicWriteCount)" + ] + } + } +#endif + struct WorkspaceSaveOwner: Equatable, Hashable { let windowID: Int? let managerID: UUID? @@ -189,7 +236,6 @@ enum WorkspaceSaveTracer { "windowID": metadata.owner.windowID.map(String.init) ?? "", "managerID": metadata.owner.managerID.map { WorkspaceRestorePerfLog.shortID($0) } ?? "", "workspaceID": WorkspaceRestorePerfLog.shortID(metadata.workspaceID), - "workspaceName": metadata.workspaceName, "workspaceDateModified": String(format: "%.6f", metadata.workspaceDateModified.timeIntervalSince1970), "activeTabID": metadata.activeTabID.map { WorkspaceRestorePerfLog.shortID($0) } ?? "", "activeSelectionRevision": "\(metadata.activeSelectionRevision)", diff --git a/Tests/RepoPromptTests/MCP/Control/PersistentAgentModeMCPReadFileConnectionTests.swift b/Tests/RepoPromptTests/MCP/Control/PersistentAgentModeMCPReadFileConnectionTests.swift index 53f3a8b99..7405bde62 100644 --- a/Tests/RepoPromptTests/MCP/Control/PersistentAgentModeMCPReadFileConnectionTests.swift +++ b/Tests/RepoPromptTests/MCP/Control/PersistentAgentModeMCPReadFileConnectionTests.swift @@ -515,6 +515,11 @@ final class PersistentAgentModeMCPReadFileConnectionTests: XCTestCase { userPath: physicalURL.path, fallbackScope: .allLoaded ) + try await AsyncTestWait.waitUntil("slice rebase is registered after applied ingress") { + fixture.window.workspaceManager + .composeTab(with: Fixture.tabID)? + .selection.slices[targetLogicalPath] != originalRanges + } let fence = await fixture.window.workspaceFilesViewModel.waitForPendingSliceRebasesAndCaptureFence( affectingCandidatePaths: [physicalURL.path] ) diff --git a/Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift b/Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift new file mode 100644 index 000000000..955132e83 --- /dev/null +++ b/Tests/RepoPromptTests/Workspaces/WorkspaceSaveCoordinatorTests.swift @@ -0,0 +1,392 @@ +import Foundation +@testable import RepoPromptApp +import XCTest + +@MainActor +final class WorkspaceSaveCoordinatorTests: XCTestCase { + override func tearDown() async throws { + await WorkspaceManagerViewModel.WorkspaceDiskWriter.shared.removeAllForTesting() + try await super.tearDown() + } + + func testFailedThenSuccessfulPayloadsReceiveTheirOwnOutcomes() async throws { + let storageRoot = try makeTestDirectory(named: "FailedThenSuccessful") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let url = storageRoot.appendingPathComponent("payload.json") + let writer = WorkspaceManagerViewModel.WorkspaceDiskWriter.shared + let gate = WorkspaceSavePreparationGate() + await writer.setAtomicWriteGateForTesting { + await gate.pauseFirstPreparation() + } + await writer.setFailAtomicWriteForTesting(url: url, afterAdditionalAttempts: 1) + + let failedTask = Task { await writer.enqueueAndWait(data: Data("failed".utf8), url: url) } + await gate.waitUntilPaused() + let successfulTask = Task { await writer.enqueueAndWait(data: Data("successful".utf8), url: url) } + let flushTask = Task { await writer.flush(url: url) } + await drainMainActorTasks() + await gate.release() + + let failedOutcome = await failedTask.value + let successfulOutcome = await successfulTask.value + let flushOutcome = await flushTask.value + XCTAssertEqual(failedOutcome, .failed) + XCTAssertEqual(successfulOutcome, .committed) + XCTAssertEqual(flushOutcome, .failed) + XCTAssertEqual(try Data(contentsOf: url), Data("successful".utf8)) + } + + func testSupersededPayloadReportsReplacementFailure() async throws { + let storageRoot = try makeTestDirectory(named: "SupersededReplacementFailure") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let url = storageRoot.appendingPathComponent("payload.json") + let writer = WorkspaceManagerViewModel.WorkspaceDiskWriter.shared + let gate = WorkspaceSavePreparationGate() + await writer.setAtomicWriteGateForTesting { + await gate.pauseFirstPreparation() + } + await writer.setFailAtomicWriteForTesting(true) + + let workspaceID = UUID() + let olderWorkspace = WorkspaceModel( + id: workspaceID, + dateModified: Date(timeIntervalSince1970: 1000), + name: "Workspace", + repoPaths: [] + ) + let newerWorkspace = WorkspaceModel( + id: workspaceID, + dateModified: Date(timeIntervalSince1970: 2000), + name: "Workspace", + repoPaths: [] + ) + let olderData = try JSONEncoder().encode(olderWorkspace) + let newerData = try JSONEncoder().encode(newerWorkspace) + let firstData = Data("first".utf8) + + let firstTask = Task { await writer.enqueueAndWait(data: firstData, url: url) } + await gate.waitUntilPaused() + let newerTask = Task { await writer.enqueueAndWait(data: newerData, url: url) } + let olderTask = Task { await writer.enqueueAndWait(data: olderData, url: url) } + let flushTask = Task { await writer.flush(url: url) } + await drainMainActorTasks() + await gate.release() + + let firstOutcome = await firstTask.value + let newerOutcome = await newerTask.value + let olderOutcome = await olderTask.value + let flushOutcome = await flushTask.value + XCTAssertEqual(firstOutcome, .failed) + XCTAssertEqual(newerOutcome, .failed) + XCTAssertEqual(olderOutcome, .failed) + XCTAssertEqual(flushOutcome, .failed) + XCTAssertFalse(FileManager.default.fileExists(atPath: url.path)) + } + + func testSuccessfulThenFailedPayloadsReceiveTheirOwnOutcomes() async throws { + let storageRoot = try makeTestDirectory(named: "SuccessfulThenFailed") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let url = storageRoot.appendingPathComponent("payload.json") + let writer = WorkspaceManagerViewModel.WorkspaceDiskWriter.shared + let gate = WorkspaceSavePreparationGate() + await writer.setAtomicWriteGateForTesting { + await gate.pauseFirstPreparation() + } + await writer.setFailAtomicWriteForTesting(url: url, afterAdditionalAttempts: 2) + + let successfulTask = Task { await writer.enqueueAndWait(data: Data("successful".utf8), url: url) } + await gate.waitUntilPaused() + let failedTask = Task { await writer.enqueueAndWait(data: Data("failed".utf8), url: url) } + let flushTask = Task { await writer.flush(url: url) } + await drainMainActorTasks() + await gate.release() + + let successfulOutcome = await successfulTask.value + let failedOutcome = await failedTask.value + let flushOutcome = await flushTask.value + XCTAssertEqual(successfulOutcome, .committed) + XCTAssertEqual(failedOutcome, .failed) + XCTAssertEqual(flushOutcome, .failed) + XCTAssertEqual(try Data(contentsOf: url), Data("successful".utf8)) + } + + func testCloseBoundaryFlushesPendingDebouncedSave() async throws { + let storageRoot = try makeTestDirectory(named: "CloseDuringDebounce") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let manager = makeManager(storageRoot: storageRoot) + let workspace = representativeWorkspace(storageRoot: storageRoot) + manager.workspaces = [workspace] + manager.activeWorkspace = workspace + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + + manager.test_scheduleWorkspaceSave(source: "test.close.deferred") + manager.prepareForWindowClose() + let outcome = await manager.flushPendingWorkspaceSavesBeforeClose() + + let committedVersion = try unwrapCommittedVersion(outcome) + XCTAssertEqual(manager.test_lastSavedVersion(workspaceID: workspace.id), committedVersion) + let fileURL = try XCTUnwrap(workspace.customStoragePath?.appendingPathComponent("workspace.json")) + XCTAssertTrue(FileManager.default.fileExists(atPath: fileURL.path)) + let decoded = try WorkspaceManagerViewModel.loadWorkspaceFromFile(at: fileURL) + XCTAssertEqual(decoded.id, workspace.id) + } + + func testCloseBoundaryRetriesDirtyVersionAfterEarlierFailure() async throws { + let storageRoot = try makeTestDirectory(named: "CloseRetryAfterFailure") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let manager = makeManager(storageRoot: storageRoot) + defer { manager.prepareForWindowClose() } + let workspace = representativeWorkspace(storageRoot: storageRoot) + manager.workspaces = [workspace] + manager.activeWorkspace = workspace + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + let writer = WorkspaceManagerViewModel.WorkspaceDiskWriter.shared + await writer.setFailAtomicWriteForTesting(true) + + manager.test_scheduleWorkspaceSave(source: "test.preCloseFailure") + let failed = await manager.test_flushWorkspaceSave( + workspaceID: workspace.id, + source: "test.observePreCloseFailure" + ) + XCTAssertEqual(failed, .failed) + XCTAssertNil(manager.test_lastSavedVersion(workspaceID: workspace.id)) + + await writer.setFailAtomicWriteForTesting(false) + let closeOutcome = await manager.flushPendingWorkspaceSavesBeforeClose() + _ = try unwrapCommittedVersion(closeOutcome) + XCTAssertNotNil(manager.test_lastSavedVersion(workspaceID: workspace.id)) + let fileURL = try XCTUnwrap(workspace.customStoragePath?.appendingPathComponent("workspace.json")) + XCTAssertTrue(FileManager.default.fileExists(atPath: fileURL.path)) + } + + func testRapidRequestsCoalesceBeforeRepresentativeWorkspacePreparation() async throws { + let storageRoot = try makeTestDirectory(named: "RapidRequests") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let manager = makeManager(storageRoot: storageRoot) + defer { manager.prepareForWindowClose() } + let workspace = representativeWorkspace(storageRoot: storageRoot) + manager.workspaces = [workspace] + manager.activeWorkspace = workspace + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + + for _ in 0 ..< 100 { + manager.test_scheduleWorkspaceSave(source: "test.rapid") + } + await drainMainActorTasks() + let outcome = await manager.test_flushWorkspaceSave(workspaceID: workspace.id, source: "test.rapid.flush") + + let committedVersion = try unwrapCommittedVersion(outcome) + XCTAssertEqual(manager.test_lastSavedVersion(workspaceID: workspace.id), committedVersion) + XCTAssertEqual(manager.test_workspaceSavePreparationCount(workspaceID: workspace.id), 1) + let summary = try XCTUnwrap(manager.test_workspaceSavePerformanceSummary(workspaceID: workspace.id)) + XCTAssertEqual(summary.composeTabCount, 25) + XCTAssertGreaterThanOrEqual(summary.payloadByteCount, 100_000) + XCTAssertEqual(summary.selectedPathCount, 180) + XCTAssertEqual(summary.sliceFileCount, 130) + XCTAssertEqual(summary.sliceRangeCount, 260) + XCTAssertGreaterThanOrEqual(summary.coalescedRequestCount, 100) + XCTAssertEqual(summary.atomicWriteCount, 1) + } + + func testMutationDuringPreparationProducesOneFollowUpAndFlushCommitsNewestState() async throws { + let storageRoot = try makeTestDirectory(named: "MutationDuringPreparation") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let manager = makeManager(storageRoot: storageRoot) + defer { manager.prepareForWindowClose() } + var workspace = representativeWorkspace(storageRoot: storageRoot) + workspace.currentPromptText = "before" + manager.workspaces = [workspace] + manager.activeWorkspace = workspace + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + let gate = WorkspaceSavePreparationGate() + manager.test_setWorkspaceSavePreparationGate { _, _ in + await gate.pauseFirstPreparation() + } + + manager.test_scheduleWorkspaceSave(source: "test.blockedPreparation") + await gate.waitUntilPaused() + manager.workspaces[0].currentPromptText = "after" + manager.workspaces[0].dateModified = Date() + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + for _ in 0 ..< 20 { + manager.test_scheduleWorkspaceSave(source: "test.newerWhileBlocked") + } + await drainMainActorTasks() + await gate.release() + let outcome = await manager.test_flushWorkspaceSave(workspaceID: workspace.id, source: "test.boundaryFlush") + manager.test_setWorkspaceSavePreparationGate(nil) + + let committedVersion = try unwrapCommittedVersion(outcome) + XCTAssertEqual(manager.test_workspaceSavePreparationCount(workspaceID: workspace.id), 2) + let fileURL = try XCTUnwrap(workspace.customStoragePath?.appendingPathComponent("workspace.json")) + let decoded = try WorkspaceManagerViewModel.loadWorkspaceFromFile(at: fileURL) + XCTAssertEqual(decoded.currentPromptText, "after") + XCTAssertEqual(manager.test_lastSavedVersion(workspaceID: workspace.id), committedVersion) + } + + func testMutationDuringAtomicWriteFlushesNewestStateBeforeCompleting() async throws { + let storageRoot = try makeTestDirectory(named: "MutationDuringAtomicWrite") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let manager = makeManager(storageRoot: storageRoot) + defer { manager.prepareForWindowClose() } + var workspace = representativeWorkspace(storageRoot: storageRoot) + workspace.currentPromptText = "before" + manager.workspaces = [workspace] + manager.activeWorkspace = workspace + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + let gate = WorkspaceSavePreparationGate() + await WorkspaceManagerViewModel.WorkspaceDiskWriter.shared.setAtomicWriteGateForTesting { + await gate.pauseFirstPreparation() + } + + manager.test_scheduleWorkspaceSave(source: "test.blockedAtomicWrite") + let flushTask = Task { + await manager.test_flushWorkspaceSave(workspaceID: workspace.id, source: "test.boundaryFlush") + } + await gate.waitUntilPaused() + manager.workspaces[0].currentPromptText = "after" + manager.workspaces[0].dateModified = Date() + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + await gate.release() + + let committedVersion = try await unwrapCommittedVersion(flushTask.value) + let fileURL = try XCTUnwrap(workspace.customStoragePath?.appendingPathComponent("workspace.json")) + let decoded = try WorkspaceManagerViewModel.loadWorkspaceFromFile(at: fileURL) + XCTAssertEqual(decoded.currentPromptText, "after") + XCTAssertEqual(manager.test_lastSavedVersion(workspaceID: workspace.id), committedVersion) + XCTAssertEqual(manager.test_workspaceSavePreparationCount(workspaceID: workspace.id), 2) + } + + func testFailedAtomicWriteIsObservableAndDoesNotAdvanceSavedVersion() async throws { + let storageRoot = try makeTestDirectory(named: "FailedWrite") + defer { try? FileManager.default.removeItem(at: storageRoot) } + let manager = makeManager(storageRoot: storageRoot) + defer { manager.prepareForWindowClose() } + var workspace = representativeWorkspace(storageRoot: storageRoot) + workspace.currentPromptText = "state to retry" + manager.workspaces = [workspace] + manager.activeWorkspace = workspace + manager.test_markWorkspaceDirty(workspaceID: workspace.id) + let writer = WorkspaceManagerViewModel.WorkspaceDiskWriter.shared + let fileURL = try XCTUnwrap(workspace.customStoragePath?.appendingPathComponent("workspace.json")) + await writer.setFailAtomicWriteForTesting(true) + + manager.test_scheduleWorkspaceSave(source: "test.failingSnapshot") + let failed = await manager.test_flushWorkspaceSave(workspaceID: workspace.id, source: "test.injectedFailure") + + XCTAssertEqual(failed, .failed) + XCTAssertNil(manager.test_lastSavedVersion(workspaceID: workspace.id)) + + await writer.setFailAtomicWriteForTesting(false) + let retried = await manager.test_flushWorkspaceSave(workspaceID: workspace.id, source: "test.retryAfterFailure") + let committedVersion = try unwrapCommittedVersion(retried) + XCTAssertEqual(manager.test_lastSavedVersion(workspaceID: workspace.id), committedVersion) + let decoded = try WorkspaceManagerViewModel.loadWorkspaceFromFile(at: fileURL) + XCTAssertEqual(decoded.currentPromptText, "state to retry") + } + + private func unwrapCommittedVersion(_ outcome: WorkspaceSaveCompletion) throws -> Int { + guard case let .committed(version) = outcome else { + XCTFail("Expected committed workspace save, received \(outcome)") + throw UnexpectedWorkspaceSaveOutcome() + } + return version + } + + private func makeManager(storageRoot: URL) -> WorkspaceManagerViewModel { + let previousAutoStart = GlobalSettingsStore.shared.mcpAutoStart() + let defaults = UserDefaults.standard + let previousStoragePath = defaults.string(forKey: "GlobalCustomStorageURL") + GlobalSettingsStore.shared.setMCPAutoStart(false, commit: false) + defaults.set(storageRoot.path, forKey: "GlobalCustomStorageURL") + defer { + GlobalSettingsStore.shared.setMCPAutoStart(previousAutoStart, commit: false) + if let previousStoragePath { + defaults.set(previousStoragePath, forKey: "GlobalCustomStorageURL") + } else { + defaults.removeObject(forKey: "GlobalCustomStorageURL") + } + } + return WindowStateCompositionFactory.make( + windowID: -940 - Int.random(in: 1 ... 40), + deferredInitialAgentSystemWorkspaceRefresh: true, + sharedMCPService: MCPService() + ).workspaceManager + } + + private func representativeWorkspace(storageRoot: URL) -> WorkspaceModel { + let selectedPaths = (0 ..< 180).map { "/synthetic/root/File\($0).swift" } + let slices = Dictionary(uniqueKeysWithValues: (0 ..< 130).map { index in + ( + "/synthetic/root/Slice\(index).swift", + [ + LineRange(start: 1, end: 5), + LineRange(start: 20, end: 30) + ] + ) + }) + let selection = StoredSelection( + selectedPaths: selectedPaths, + slices: slices, + codemapAutoEnabled: false + ) + let tabs = (0 ..< 25).map { index in + ComposeTabState( + name: "T\(index + 1)", + selection: index == 0 ? selection : .init(), + promptText: String(repeating: "p", count: 4000) + ) + } + return WorkspaceModel( + name: "Synthetic Save Fixture", + repoPaths: ["/synthetic/root"], + customStoragePath: storageRoot.appendingPathComponent("workspace", isDirectory: true), + composeTabs: tabs, + activeComposeTabID: tabs[0].id + ) + } + + private func makeTestDirectory(named name: String) throws -> URL { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("WorkspaceSaveCoordinatorTests-\(UUID().uuidString)", isDirectory: true) + .appendingPathComponent(name, isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + return directory + } + + private func drainMainActorTasks() async { + for _ in 0 ..< 5 { + await Task.yield() + } + } +} + +private struct UnexpectedWorkspaceSaveOutcome: Error {} + +private actor WorkspaceSavePreparationGate { + private var didPause = false + private var pauseWaiters: [CheckedContinuation] = [] + private var releaseWaiters: [CheckedContinuation] = [] + + func pauseFirstPreparation() async { + guard !didPause else { return } + didPause = true + let waiters = pauseWaiters + pauseWaiters.removeAll() + waiters.forEach { $0.resume() } + await withCheckedContinuation { releaseWaiters.append($0) } + } + + func waitUntilPaused() async { + if didPause { + return + } + await withCheckedContinuation { pauseWaiters.append($0) } + } + + func release() { + let waiters = releaseWaiters + releaseWaiters.removeAll() + waiters.forEach { $0.resume() } + } +}