Fence Hermes memory work while suspended
This commit is contained in:
1 parent
70c41b9225
commit
568e9b7327
3 files changed
+157
-11
No files matched your search
@@ -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<Void, Never>] = []
|
||||
private var releaseWaiters: [CheckedContinuation<Void, Never>] = []
|
||||
|
||||
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() }
|
||||
|
||||
@@ -33,6 +33,7 @@ public final class MemoryModel {
|
||||
private var indexDirty = false
|
||||
private var reviews: [UUID: Task<Void, Never>] = [:]
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user