From 568e9b73272d45f70e041acc08ca797f611b06bf Mon Sep 17 00:00:00 2001 From: Kaede Akino Date: Wed, 23 Sep 2026 16:48:07 +0800 Subject: [PATCH] Fence Hermes memory work while suspended --- FrameworkTests/HermesMemoryTests.swift | 139 ++++++++++++++++++ .../FeatureModels/MemoryModel.swift | 28 ++-- TFI/App/MobileAppModel.swift | 1 + 3 files changed, 157 insertions(+), 11 deletions(-) diff --git a/FrameworkTests/HermesMemoryTests.swift b/FrameworkTests/HermesMemoryTests.swift index 27c13fa6..cbc5a2cd 100644 --- a/FrameworkTests/HermesMemoryTests.swift +++ b/FrameworkTests/HermesMemoryTests.swift @@ -113,6 +113,66 @@ struct HermesMemoryTests { await model.setEnabled(false) } + @Test func suspendedMemoryDefersTerminalHistoryIndexUntilResume() async throws { + let store = MemoryLifecycleStore() + let model = MemoryModel(storage: store) + let history = InMemoryAgentRunRepository() + let repository = MemoryIndexedAgentRepository(base: history, memory: model) + await model.setEnabled(true) + try await waitUntil { await store.historyReplacementCount == 1 && !model.isIndexing } + + model.suspend() + let conversationID = UUID() + try await repository.save( + .init( + conversationID: conversationID, state: .completed, permissionMode: .askForApproval, + messages: [.init(role: .user, text: "Saved after suspension")])) + + #expect(!model.isIndexing) + #expect(await store.historyReplacementCount == 1) + + model.resume() + try await waitUntil { await store.historyReplacementCount == 2 && !model.isIndexing } + #expect(await store.lastHistory?.contains(where: { $0.id == conversationID }) == true) + } + + @Test func suspendInvalidatesExistingMemoryAccessAcrossResume() async throws { + let store = MemoryLifecycleStore() + let model = MemoryModel(storage: store) + await model.setEnabled(true) + let conversationID = UUID() + let original = try #require(await model.session(conversationID)) + + model.suspend() + #expect(await model.session(conversationID) == nil) + model.resume() + + await #expect(throws: CancellationError.self) { try await original.access.memoryState() } + let resumed = try #require(await model.session(conversationID)) + _ = try await resumed.access.memoryState() + } + + @Test func suspendPreventsStaleReviewFromApplyingAfterResume() async throws { + let store = MemoryLifecycleStore() + let model = MemoryModel(storage: store) + await model.setEnabled(true) + let reviewModel = BlockingMemoryReviewModel() + let review = Task { + await model.review( + conversationID: UUID(), messages: [.init(role: .user, text: "Remember concise replies")], + model: reviewModel, wait: true) + } + await reviewModel.waitUntilEntered() + + model.suspend() + model.resume() + await reviewModel.release() + await review.value + + #expect(await store.reviewApplyCount == 0) + #expect(model.state.entries.isEmpty) + } + @Test func skillPackagesPersistWithoutChangingLegacySkillRows() throws { let container = try TermindPersistence.makeContainer(inMemory: true) let persistence = SkillPersistence(context: container.mainContext) @@ -183,6 +243,85 @@ struct HermesMemoryTests { } } +private actor MemoryLifecycleStore: MemoryPersisting { + private var memory = AgentMemoryState() + private var historyReplacements: [[StoredAgentConversation]] = [] + private(set) var reviewApplyCount = 0 + + var historyReplacementCount: Int { historyReplacements.count } + var lastHistory: [StoredAgentConversation]? { historyReplacements.last } + + func rebuildHistoryIndex() {} + func memoryState() -> AgentMemoryState { memory } + func applyMemory(_ operations: [AgentMemoryOperation]) throws -> AgentMemoryState { + memory = try memory.applying(operations) + return memory + } + func snapshot(for id: UUID) -> String { memory.prompt } + func replaceHistory(_ conversations: [StoredAgentConversation]) { historyReplacements.append(conversations) } + func searchSessions(_ request: AgentSessionSearchRequest) -> AgentJSONValue { .array([]) } + func lastReviewedMessage(_ id: UUID) -> UUID? { nil } + func markReviewed(_ messageID: UUID, conversationID: UUID) {} + func applyReview( + _ operations: [AgentMemoryOperation], expected: AgentMemoryState, messageID: UUID, conversationID: UUID + ) throws -> AgentMemoryState? { + reviewApplyCount += 1 + guard memory == expected else { return nil } + memory = try memory.applying(operations) + return memory + } + func replaceTarget(_ target: AgentMemoryTarget, text: String) -> AgentMemoryState { memory } + func close() {} + func destroy() { memory = .init() } +} + +private actor BlockingMemoryReviewModel: AgentModelCompleting { + private var entered = false + private var released = false + private var entryWaiters: [CheckedContinuation] = [] + private var releaseWaiters: [CheckedContinuation] = [] + + func waitUntilEntered() async { + if entered { return } + await withCheckedContinuation { entryWaiters.append($0) } + } + + func release() { + released = true + let waiters = releaseWaiters + releaseWaiters.removeAll() + for waiter in waiters { waiter.resume() } + } + + func complete(_ request: AgentModelRequest) async throws -> [AgentModelStreamEvent] { + entered = true + let waiters = entryWaiters + entryWaiters.removeAll() + for waiter in waiters { waiter.resume() } + if !released { + await withCheckedContinuation { releaseWaiters.append($0) } + } + return [ + .textDelta(#"{"operations":[{"action":"add","target":"user","content":"Likes concise replies"}]}"#), + .finished(.completed), + ] + } +} + +@MainActor +private func waitUntil( + timeout: Duration = .seconds(2), _ condition: @escaping @MainActor () async -> Bool +) async throws { + let clock = ContinuousClock() + let deadline = clock.now.advanced(by: timeout) + while !(await condition()) { + guard clock.now < deadline else { throw MemoryLifecycleTestTimeout() } + try await Task.sleep(for: .milliseconds(1)) + } +} + +private struct MemoryLifecycleTestTimeout: Error {} + private struct FailingDestroyStore: MemoryPersisting { let base: GRDBMemoryStore func rebuildHistoryIndex() async throws { try await base.rebuildHistoryIndex() } diff --git a/Frameworks/TermindApplication/FeatureModels/MemoryModel.swift b/Frameworks/TermindApplication/FeatureModels/MemoryModel.swift index c884a2a0..c6b661a9 100644 --- a/Frameworks/TermindApplication/FeatureModels/MemoryModel.swift +++ b/Frameworks/TermindApplication/FeatureModels/MemoryModel.swift @@ -33,6 +33,7 @@ public final class MemoryModel { private var indexDirty = false private var reviews: [UUID: Task] = [:] private var pendingReviews: [UUID: (@Sendable () async -> Void)] = [:] + private var isSuspended = false public init(storage: (any MemoryPersisting)? = nil, defaults: UserDefaults? = nil) { self.storage = storage @@ -75,7 +76,7 @@ public final class MemoryModel { } catch { self.error = error.localizedDescription } } public func refresh() async { - guard isEnabled, let storage else { return } + guard isEnabled, !isSuspended, let storage else { return } let token = generation do { let value = try await storage.memoryState() @@ -86,7 +87,7 @@ public final class MemoryModel { } catch { if token == generation { self.error = error.localizedDescription } } } public func edit(_ target: AgentMemoryTarget, text: String) async { - guard isEnabled, let storage else { return } + guard isEnabled, !isSuspended, let storage else { return } let token = generation do { let updated = try await storage.replaceTarget(target, text: text) @@ -96,7 +97,7 @@ public final class MemoryModel { } catch { if token == generation { self.error = error.localizedDescription } } } public func session(_ id: UUID) async -> (context: AgentMemoryContext, access: any AgentMemoryAccessing)? { - guard isEnabled, let storage else { return nil } + guard isEnabled, !isSuspended, let storage else { return nil } let token = generation do { let context: AgentMemoryContext @@ -116,7 +117,7 @@ public final class MemoryModel { } private func require(_ token: UUID) throws -> any MemoryPersisting { try Task.checkCancellation() - guard isEnabled, token == generation, !isBusy, let storage else { throw CancellationError() } + guard isEnabled, !isSuspended, token == generation, !isBusy, let storage else { throw CancellationError() } return storage } fileprivate func read(token: UUID) async throws -> AgentMemoryState { @@ -140,7 +141,7 @@ public final class MemoryModel { return result } public func rebuildHistoryIndex() async { - guard isEnabled else { return } + guard isEnabled, !isSuspended else { return } let token = generation historyChanged() await indexTask?.value @@ -151,7 +152,7 @@ public final class MemoryModel { } public func historyChanged() { - guard isEnabled, let storage, let history else { return } + guard isEnabled, !isSuspended, let storage, let history else { return } indexDirty = true guard indexTask == nil else { return } let token = generation @@ -161,7 +162,7 @@ public final class MemoryModel { defer { indexTask = nil isIndexing = false - if indexDirty, isEnabled { historyChanged() } + if indexDirty, isEnabled, !isSuspended { historyChanged() } } do { while indexDirty, token == generation, isEnabled, !Task.isCancelled { @@ -183,7 +184,7 @@ public final class MemoryModel { conversationID: UUID, messages: [AgentTranscriptMessage], model: any AgentModelCompleting, wait: Bool = false ) async { - guard isEnabled, reviewEnabled, let storage, let last = messages.last?.id else { return } + guard isEnabled, !isSuspended, reviewEnabled, let storage, let last = messages.last?.id else { return } let token = generation if let running = reviews[conversationID] { if wait { @@ -249,9 +250,14 @@ public final class MemoryModel { isBusy = false } public func suspend() { - indexTask?.cancel() - for task in reviews.values { task.cancel() } - pendingReviews.removeAll() + guard !isSuspended else { return } + isSuspended = true + invalidate() + } + public func resume() { + guard isSuspended else { return } + isSuspended = false + historyChanged() } } diff --git a/TFI/App/MobileAppModel.swift b/TFI/App/MobileAppModel.swift index 896d8ad6..8c4f86df 100644 --- a/TFI/App/MobileAppModel.swift +++ b/TFI/App/MobileAppModel.swift @@ -566,6 +566,7 @@ final class MobileAppModel { } else if phase == .active { devices.updateCurrentDeviceName(Device.current.safeDescription) sftpEdits.setMonitoring(true) + memory.resume() Task { await sftpEdits.recheck() } Task { [weak self] in await self?.pro.refresh() } background.finish()