diff --git a/CHANGELOG.md b/CHANGELOG.md index 3280414..e1314ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,12 @@ All notable changes to Engram are documented in this file. ## [Unreleased] +## [0.14.10] - 2026-09-24 + ### Fixed +- Report memory updates as successful only after a checked transaction commits on the database connection that owns the selected memory, including synced and group memories. +- Recognize update attempts that stop before a busy database transaction begins as verified no-write failures; preserve reconciliation for uncertain writes. +- Check edge restoration during undelete and propagate failures without claiming that an already committed memory update wrote nothing. - Show Codex session learners only the write options their restricted gateway accepts, including exact memory IDs and preserved privacy for updates. - Instruct Codex session learners to stop making tool calls after an MCP error, avoiding follow-up writes after a rejected call; conservative completion and reconciliation checks remain unchanged. diff --git a/Package.swift b/Package.swift index f22d4c1..92c1eb6 100644 --- a/Package.swift +++ b/Package.swift @@ -240,6 +240,7 @@ let package = Package( "EngramMemoryCore", "EngramMemoryContract", .product(name: "Lattice", package: "lattice"), + .product(name: "SwiftLM", package: "SwiftLM"), ], swiftSettings: [ .interoperabilityMode(.Cxx), diff --git a/Sources/EngramKit/MemoryTools+Core.swift b/Sources/EngramKit/MemoryTools+Core.swift index 37c68f5..2095f10 100644 --- a/Sources/EngramKit/MemoryTools+Core.swift +++ b/Sources/EngramKit/MemoryTools+Core.swift @@ -999,11 +999,13 @@ extension MemoryTools { // 5. Locate memory let mem: Memory + let writeLattice: Lattice if let gid = a.id?.value { guard let found = findMemory(id: gid) else { return CallTool.Result(content: [.text("Memory with id \(gid.uuidString) not found.")], isError: true) } mem = found.memory + writeLattice = found.lattice } else { let query = a.query! let db = readLattice(for: a.project) @@ -1020,6 +1022,10 @@ extension MemoryTools { return CallTool.Result(content: [.text("No matching memory found to update.")], isError: false) } mem = match.object + // Keep the connection that hydrated the selected row. A union + // query can return an attached-store row whose setters must be + // covered by this same connection's checked transaction. + writeLattice = db } // 5b. Tombstone gate: a soft-deleted memory only accepts undelete — @@ -1078,79 +1084,106 @@ extension MemoryTools { } } - try localLattice.transaction { - if let content = a.content { - mem.content = content - contentChanged = true - } else if let append = a.append { - mem.content += "\n" + append - contentChanged = true - } else if let prepend = a.prepend { - mem.content = prepend + "\n" + mem.content - contentChanged = true - } else if let find = a.find { - let replace = a.replace! - mem.content = mem.content.replacingOccurrences(of: find, with: replace) - contentChanged = true - } + var transactionBodyEntered = false + do { + try writeLattice.withTransaction { + transactionBodyEntered = true + if let content = a.content { + mem.content = content + contentChanged = true + } else if let append = a.append { + mem.content += "\n" + append + contentChanged = true + } else if let prepend = a.prepend { + mem.content = prepend + "\n" + mem.content + contentChanged = true + } else if let find = a.find { + let replace = a.replace! + mem.content = mem.content.replacingOccurrences(of: find, with: replace) + contentChanged = true + } - if contentChanged { - changes.append("content: \(oldContent.prefix(60))... → \(mem.content.prefix(60))...") - } + if contentChanged { + changes.append("content: \(oldContent.prefix(60))... → \(mem.content.prefix(60))...") + } - if let project = a.setProject { - let old = mem.project - mem.project = project - changes.append("project: \(old) → \(project)") - } - if let topic = a.topic { - let old = mem.topic - mem.topic = topic - changes.append("topic: \(old) → \(topic)") - } - if let source = a.source { - let old = mem.source - mem.source = source - changes.append("source: \(old) → \(source)") - } - if let days = a.expiresInDays?.value { - let oldExpires = mem.expiresAt == .distantFuture ? "permanent" : Self.dateFormatter.string(from: mem.expiresAt) - if days == 0 { - mem.expiresAt = .distantFuture - changes.append("expires: \(oldExpires) → permanent") - } else { - mem.expiresAt = Date().addingTimeInterval(Double(days) * 86400) - changes.append("expires: \(oldExpires) → \(Self.dateFormatter.string(from: mem.expiresAt))") + if let project = a.setProject { + let old = mem.project + mem.project = project + changes.append("project: \(old) → \(project)") } - } - if let imp = a.importance?.value { - let old = mem.importance - mem.importance = imp - changes.append("importance: \(old) → \(imp)") - } - if let priv = a.isPrivate { - let old = mem.isPrivate - mem.isPrivate = priv - changes.append("private: \(old) → \(priv)") - if priv && !old && isGroupShared(mem) { - changes.append("⚠️ retracted from the group: the group's copy (including any teammate edits) is removed for all members") + if let topic = a.topic { + let old = mem.topic + mem.topic = topic + changes.append("topic: \(old) → \(topic)") + } + if let source = a.source { + let old = mem.source + mem.source = source + changes.append("source: \(old) → \(source)") + } + if let days = a.expiresInDays?.value { + let oldExpires = mem.expiresAt == .distantFuture ? "permanent" : Self.dateFormatter.string(from: mem.expiresAt) + if days == 0 { + mem.expiresAt = .distantFuture + changes.append("expires: \(oldExpires) → permanent") + } else { + mem.expiresAt = Date().addingTimeInterval(Double(days) * 86400) + changes.append("expires: \(oldExpires) → \(Self.dateFormatter.string(from: mem.expiresAt))") + } + } + if let imp = a.importance?.value { + let old = mem.importance + mem.importance = imp + changes.append("importance: \(old) → \(imp)") + } + if let priv = a.isPrivate { + let old = mem.isPrivate + mem.isPrivate = priv + changes.append("private: \(old) → \(priv)") + if priv && !old && isGroupShared(mem) { + changes.append("⚠️ retracted from the group: the group's copy (including any teammate edits) is removed for all members") + } + } + if a.undelete == true, mem.deletedAt != nil { + mem.deletedAt = nil + mem.deletedBy = nil + didUndelete = true + changes.append("undeleted (restored for all members)") } - } - if a.undelete == true, mem.deletedAt != nil { - mem.deletedAt = nil - mem.deletedBy = nil - didUndelete = true - changes.append("undeleted (restored for all members)") - } - if contentChanged, let emb = newEmbedding { - mem.embedding = Vector(emb) - } + if contentChanged, let emb = newEmbedding { + mem.embedding = Vector(emb) + } - mem.lastAccessedAt = Date() - // authorUserId is NEVER touched by edits — attribution follows - // the original author, and the sync firewall keys on it. - mem.modifiedAt = Date() + mem.lastAccessedAt = Date() + // authorUserId is NEVER touched by edits — attribution follows + // the original author, and the sync firewall keys on it. + mem.modifiedAt = Date() + } + } catch { + // Only a failed BEGIN proves that this attempt wrote nothing. + // Setter, COMMIT, rollback and notification failures retain an + // unknown outcome; never turn those into a safe-retry receipt. + guard !transactionBodyEntered, + case LatticeError.transactionError(let detail) = error, + ["database is locked", "database is busy", "database table is locked", + "database schema is locked"].contains(where: { + detail == "Failed to begin transaction: " + $0 + }) else { throw error } + return CallTool.Result( + content: [.text("Memory was not updated: the database was busy before the write transaction started. Retry on a later turn.")], + structuredContent: .object([ + "engram_write_receipt": .object([ + "schema_version": .int(1), + "tool": .string("update"), + "write_outcome": .string("not_stored_transaction_not_started"), + "reason": .string("database_busy"), + "memory_ids": .array([]), + ]), + ]), + isError: true + ) } // Restore graph connectivity alongside the memory: edges tombstoned @@ -1159,7 +1192,9 @@ extension MemoryTools { // (didUndelete, not a mem.deletedAt re-read — the materialized // snapshot can serve the stale pre-transaction value.) if didUndelete, let gid = mem.globalId { - let revived = reviveEdgesForMemory(gid) + // The memory is already committed. A later graph failure must + // propagate as an uncertain partial outcome, never as no write. + let revived = try reviveEdgesForMemory(gid) if revived > 0 { changes.append("revived \(revived) edge(s)") } } diff --git a/Sources/EngramKit/MemoryTools+Graph.swift b/Sources/EngramKit/MemoryTools+Graph.swift index 25eddae..4076408 100644 --- a/Sources/EngramKit/MemoryTools+Graph.swift +++ b/Sources/EngramKit/MemoryTools+Graph.swift @@ -324,20 +324,25 @@ extension MemoryTools { /// tombstoned, so restoring one memory never resurrects links into /// removed content. @discardableResult - func reviveEdgesForMemory(_ gid: UUID) -> Int { + func reviveEdgesForMemory(_ gid: UUID) throws -> Int { var total = 0 var lattices: [Lattice] = [localLattice] if let syncedLattice { lattices.append(syncedLattice) } for lattice in lattices { - let edges = lattice.objects(Edge.self) - .where { ($0.sourceGlobalId == gid || $0.targetGlobalId == gid) && $0.deletedAt != nil } - .snapshot() - for edge in edges { - let otherGid = edge.sourceGlobalId == gid ? edge.targetGlobalId : edge.sourceGlobalId - guard let (other, _) = findMemory(id: otherGid), other.deletedAt == nil else { continue } - edge.deletedAt = nil - total += 1 + let revived = try lattice.withTransaction { + let edges = lattice.objects(Edge.self) + .where { ($0.sourceGlobalId == gid || $0.targetGlobalId == gid) && $0.deletedAt != nil } + .snapshot() + var count = 0 + for edge in edges { + let otherGid = edge.sourceGlobalId == gid ? edge.targetGlobalId : edge.sourceGlobalId + guard let (other, _) = findMemory(id: otherGid), other.deletedAt == nil else { continue } + edge.deletedAt = nil + count += 1 + } + return count } + total += revived } return total } diff --git a/Tests/EngramTests/TestHelpers.swift b/Tests/EngramTests/TestHelpers.swift index 3106d3d..2509152 100644 --- a/Tests/EngramTests/TestHelpers.swift +++ b/Tests/EngramTests/TestHelpers.swift @@ -1,8 +1,10 @@ import Testing import EngramKit +import EngramMemoryCore import Lattice import MCP import Foundation +import SwiftLM // Lattice isn't Sendable but tests are sequential — safe for test inspection. extension Lattice: @retroactive @unchecked Sendable {} @@ -21,15 +23,86 @@ let sharedEmbedder: EmbeddingService = { return e }() +private enum FixtureEmbeddingError: Error { + case invalidComputeUnits(String) + case missingResource(String) + case invalidDimension(Int) + case invalidVector +} + +/// The same real MiniLM model, restricted to CPU execution for isolated runners. +private actor CPUFixtureEmbedder: Embedder { + private let model: CoreMLEmbeddingModel + + init(model: CoreMLEmbeddingModel) { self.model = model } + + var dimension: Int { model.embeddingDimension } + + func embed(text: String) async throws -> [Float]? { + let vector = try await model.embed(text: text) + guard vector.count == 384, vector.allSatisfy({ $0.isFinite }) else { + throw FixtureEmbeddingError.invalidVector + } + return vector + } +} + +// Global initialization is lazy; ordinary test runs keep using sharedEmbedder. +// One task shares successful loading or its error across concurrent fixtures. +private let cpuFixtureEmbedder = Task { + let bundle: Bundle + if let path = ProcessInfo.processInfo.environment["ENGRAM_TEST_RESOURCE_BUNDLE"] { + guard let configured = Bundle(path: path) else { + throw FixtureEmbeddingError.missingResource(path) + } + bundle = configured + } else { + bundle = engramKitResourceBundle + } + guard let modelURL = bundle.url(forResource: "paraphrase-MiniLM-L6-v2_Embedding", + withExtension: "mlmodelc"), + let tokenizerURL = bundle.url(forResource: "paraphrase-MiniLM-L6-v2_tokenizer", + withExtension: nil) else { + throw FixtureEmbeddingError.missingResource("MiniLM model and tokenizer") + } + let model = try await CoreMLEmbeddingModel.loadCompiled( + url: modelURL, tokenizerDirectory: tokenizerURL, computeUnits: .cpuOnly) + guard model.embeddingDimension == 384 else { + throw FixtureEmbeddingError.invalidDimension(model.embeddingDimension) + } + return CPUFixtureEmbedder(model: model) +} + +/// Opt into CPU-only real embeddings without changing production/default tests. +func loadedFixtureEmbedder() async throws -> any Embedder { + switch ProcessInfo.processInfo.environment["ENGRAM_TEST_COMPUTE_UNITS"] { + case "cpuOnly": + return try await cpuFixtureEmbedder.value + case nil, "": + if await !sharedEmbedder.isLoaded { + await sharedEmbedder.load() + } + return sharedEmbedder + case let value?: + throw FixtureEmbeddingError.invalidComputeUnits(value) + } +} + +/// Allow isolated test runners to choose a writable fixture directory. +func testFixtureDirectory() throws -> URL { + let directory = ProcessInfo.processInfo.environment["ENGRAM_TEST_ROOT"] + .map { URL(fileURLWithPath: $0, isDirectory: true) } + ?? FileManager.default.temporaryDirectory + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + return directory +} + /// Create a MemoryTools with an isolated temp database and the real embedding model. func makeTools() async throws -> MemoryTools { - let path = FileManager.default.temporaryDirectory + let path = try testFixtureDirectory() .appending(path: "claude-memory-test-\(UUID().uuidString).sqlite") let lattice = try Lattice(Memory.self, Edge.self, Checkpoint.self, HookState.self, SyncConfig.self, configuration: .init(fileURL: path)) - let embedder = sharedEmbedder - if await !embedder.isLoaded { - await embedder.load() - } + let embedder = try await loadedFixtureEmbedder() return MemoryTools(localRef: lattice.sendableReference, syncedRef: nil, embedder: embedder) } @@ -42,18 +115,16 @@ struct DualDBContext { /// Create a MemoryTools with separate local and synced databases for testing dual-DB routing. func makeDualDBTools() async throws -> DualDBContext { - let localPath = FileManager.default.temporaryDirectory + let directory = try testFixtureDirectory() + let localPath = directory .appending(path: "claude-memory-test-local-\(UUID().uuidString).sqlite") - let syncedPath = FileManager.default.temporaryDirectory + let syncedPath = directory .appending(path: "claude-memory-test-synced-\(UUID().uuidString).sqlite") let localLattice = try Lattice(Memory.self, Edge.self, Checkpoint.self, HookState.self, SyncConfig.self, configuration: .init(fileURL: localPath)) let syncedLattice = try Lattice(Memory.self, Edge.self, SyncConfig.self, configuration: .init(fileURL: syncedPath)) - let embedder = sharedEmbedder - if await !embedder.isLoaded { - await embedder.load() - } + let embedder = try await loadedFixtureEmbedder() let tools = MemoryTools( localRef: localLattice.sendableReference, diff --git a/Tests/EngramTests/TombstoneTests.swift b/Tests/EngramTests/TombstoneTests.swift index 8e23ea6..6b14d9b 100644 --- a/Tests/EngramTests/TombstoneTests.swift +++ b/Tests/EngramTests/TombstoneTests.swift @@ -1,5 +1,6 @@ import Testing import EngramKit +import EngramMemoryCore import EngramModels import Lattice import MCP @@ -14,17 +15,17 @@ import Foundation private struct ToolsContext { let tools: MemoryTools let lattice: Lattice + let embedder: any Embedder } private func makeToolsWithLattice() async throws -> ToolsContext { - let path = FileManager.default.temporaryDirectory + let path = try testFixtureDirectory() .appending(path: "engram-tombstone-test-\(UUID().uuidString).sqlite") let lattice = try Lattice(Memory.self, Edge.self, Checkpoint.self, HookState.self, SyncConfig.self, configuration: .init(fileURL: path)) - let embedder = sharedEmbedder - if await !embedder.isLoaded { await embedder.load() } + let embedder = try await loadedFixtureEmbedder() return ToolsContext( tools: MemoryTools(localRef: lattice.sendableReference, syncedRef: nil, embedder: embedder), - lattice: lattice) + lattice: lattice, embedder: embedder) } /// Marks a project as exposed to a group (what setGroupExposure will do). @@ -145,7 +146,7 @@ private func remember(_ tools: MemoryTools, _ content: String, project: String) let mem = Memory( content: "Teammate insight: the flaky test is timezone-dependent", topic: "debugging", project: "badge-proj", - embedding: Vector(try await sharedEmbedder.embed(text: "Teammate insight: the flaky test is timezone-dependent")!), + embedding: Vector(try await ctx.embedder.embed(text: "Teammate insight: the flaky test is timezone-dependent")!), authorUserId: foreignAuthor) try ctx.lattice.add(mem) _ = try await remember(ctx.tools, "My own note about test flakiness", project: "badge-proj") @@ -176,7 +177,7 @@ private func remember(_ tools: MemoryTools, _ content: String, project: String) let mem = Memory( content: "Foreign-authored group memory about CI caching", topic: "ci", project: "authz-proj", - embedding: Vector(try await sharedEmbedder.embed(text: "Foreign-authored group memory about CI caching")!), + embedding: Vector(try await ctx.embedder.embed(text: "Foreign-authored group memory about CI caching")!), authorUserId: foreignAuthor) try ctx.lattice.add(mem) let gid = mem.globalId! @@ -200,7 +201,7 @@ private func remember(_ tools: MemoryTools, _ content: String, project: String) let foreign = Memory( content: "Teammate memory about connection pooling limits", topic: "general", project: "cons-proj", - embedding: Vector(try await sharedEmbedder.embed(text: "Teammate memory about connection pooling limits")!), + embedding: Vector(try await ctx.embedder.embed(text: "Teammate memory about connection pooling limits")!), authorUserId: UUID()) try ctx.lattice.add(foreign) let foreignGid = foreign.globalId! diff --git a/Tests/EngramTests/UpdatePersistenceTests.swift b/Tests/EngramTests/UpdatePersistenceTests.swift new file mode 100644 index 0000000..87cb804 --- /dev/null +++ b/Tests/EngramTests/UpdatePersistenceTests.swift @@ -0,0 +1,317 @@ +@testable import EngramKit +import EngramMemoryCore +import EngramModels +import Foundation +import Lattice +import MCP +import SQLite3 +import Testing + +// Synthetic stores only. Exercise the actual handler, detached setters and +// checked transactions; a separate SQLite reader verifies committed bytes. +// No executable, provider, embedding model, groups.json or live store is used. +@Suite("Update durable acknowledgements", .serialized) +struct UpdatePersistenceTests { + private static let oldDate = Date(timeIntervalSince1970: 946_684_800) + private static let project = "UpdatePersistence" + private static let routes = ["local-id", "synced-id", "group-id", + "local-query", "synced-query", "group-query"] + + private enum Owner: String, CaseIterable, Sendable { + case local, synced, group + } + + private struct FixedEmbedder: Embedder { + var dimension: Int { 384 } + func embed(text: String) async throws -> [Float]? { + var vector = [Float](repeating: 0, count: dimension) + vector[0] = 1 + return vector + } + } + + private struct SavedMemory: Equatable { + let topic: String + let source: String + let importance: Int + let accessedAt: Double + let modifiedAt: Double? + let deletedAt: Double? + } + + private struct SavedEdge: Equatable { + let deletedAt: Double? + } + + private struct SQLiteFailure: Error { let message: String } + + private final class SQL: @unchecked Sendable { + private let db: OpaquePointer + + init(_ path: URL, readOnly: Bool = false) throws { + var pointer: OpaquePointer? + let flags = (readOnly ? SQLITE_OPEN_READONLY : SQLITE_OPEN_READWRITE) | SQLITE_OPEN_FULLMUTEX + let status = sqlite3_open_v2(path.path, &pointer, flags, nil) + guard status == SQLITE_OK, let pointer else { + let message = pointer.map { String(cString: sqlite3_errmsg($0)) } ?? "sqlite open failed" + if let pointer { sqlite3_close(pointer) } + throw SQLiteFailure(message: message) + } + db = pointer + sqlite3_busy_timeout(db, 100) + } + + deinit { sqlite3_close(db) } + + func execute(_ sql: String) throws { + guard sqlite3_exec(db, sql, nil, nil, nil) == SQLITE_OK else { + throw SQLiteFailure(message: String(cString: sqlite3_errmsg(db))) + } + } + + private func prepare(_ sql: String) throws -> OpaquePointer { + var statement: OpaquePointer? + guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK, + let statement else { + throw SQLiteFailure(message: String(cString: sqlite3_errmsg(db))) + } + return statement + } + + private func optionalDate(_ statement: OpaquePointer, _ column: Int32) -> Double? { + sqlite3_column_type(statement, column) == SQLITE_NULL ? nil : sqlite3_column_double(statement, column) + } + + func memories() throws -> [UUID: SavedMemory] { + let statement = try prepare("SELECT globalId, topic, source, importance, lastAccessedAt, modifiedAt, deletedAt FROM Memory") + defer { sqlite3_finalize(statement) } + var rows: [UUID: SavedMemory] = [:] + var status = sqlite3_step(statement) + while status == SQLITE_ROW { + guard let idText = sqlite3_column_text(statement, 0), + let id = UUID(uuidString: String(cString: idText)), + let topic = sqlite3_column_text(statement, 1), + let source = sqlite3_column_text(statement, 2) else { + throw SQLiteFailure(message: "invalid synthetic memory metadata") + } + rows[id] = SavedMemory(topic: String(cString: topic), source: String(cString: source), + importance: Int(sqlite3_column_int64(statement, 3)), + accessedAt: sqlite3_column_double(statement, 4), + modifiedAt: optionalDate(statement, 5), + deletedAt: optionalDate(statement, 6)) + status = sqlite3_step(statement) + } + guard status == SQLITE_DONE else { + throw SQLiteFailure(message: String(cString: sqlite3_errmsg(db))) + } + return rows + } + + func edges() throws -> [UUID: SavedEdge] { + let statement = try prepare("SELECT globalId, deletedAt FROM Edge") + defer { sqlite3_finalize(statement) } + var rows: [UUID: SavedEdge] = [:] + var status = sqlite3_step(statement) + while status == SQLITE_ROW { + guard let idText = sqlite3_column_text(statement, 0), + let id = UUID(uuidString: String(cString: idText)) else { + throw SQLiteFailure(message: "invalid synthetic edge identity") + } + rows[id] = SavedEdge(deletedAt: optionalDate(statement, 1)) + status = sqlite3_step(statement) + } + guard status == SQLITE_DONE else { + throw SQLiteFailure(message: String(cString: sqlite3_errmsg(db))) + } + return rows + } + } + + private struct Fixture { + let tools: MemoryTools + let writers: [Owner: Lattice] + let paths: [Owner: URL] + let ids: [Owner: UUID] + let owner: Owner + let query: Bool + + var targetID: UUID { ids[owner]! } + var ownerPath: URL { paths[owner]! } + + func snapshot() throws -> [Owner: [UUID: SavedMemory]] { + try Dictionary(uniqueKeysWithValues: Owner.allCases.map { + ($0, try SQL(paths[$0]!, readOnly: true).memories()) + }) + } + + func request(topic: String = "after", importance: Int = 5) -> CallTool.Parameters { + var arguments: [String: Value] = ["topic": .string(topic), "importance": .int(importance)] + if query { + arguments["query"] = .string("synthetic exact unit vector") + arguments["project"] = .string(UpdatePersistenceTests.project) + } else { + arguments["id"] = .string(targetID.uuidString) + } + return CallTool.Parameters(name: "update", arguments: arguments) + } + } + + private func fixture(_ route: String, tombstoned: Bool = false) throws -> Fixture { + let pieces = route.split(separator: "-") + let owner = try #require(Owner(rawValue: String(pieces[0]))) + let query = pieces[1] == "query" + let environment = ProcessInfo.processInfo.environment + let configured = environment["ENGRAM_UPDATE_PERSISTENCE_TEST_ROOT"] + ?? environment["ENGRAM_TEST_ROOT"] + let root = configured.map { URL(fileURLWithPath: $0, isDirectory: true) } + ?? FileManager.default.temporaryDirectory + let directory = root.appendingPathComponent("update-\(route)-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true, + attributes: [.posixPermissions: 0o700]) + var writers: [Owner: Lattice] = [:] + var paths: [Owner: URL] = [:] + var ids: [Owner: UUID] = [:] + // Each store gets rowid 1 with a different globalId. Query tests select + // exactly one owner through a real local+synced+group union, so borrowed + // rowids and writes to a selection handle cannot pass the readback. + for store in Owner.allCases { + let path = directory.appendingPathComponent("\(store.rawValue).sqlite") + let writer = try Lattice(Memory.self, Edge.self, SyncConfig.self, GroupProjectMap.self, + configuration: .init(fileURL: path, busyTimeoutMs: 100)) + var vector = [Float](repeating: 0, count: 384) + vector[0] = store == owner ? 1 : -1 + let memory = Memory(content: "synthetic \(store.rawValue) update fixture", + topic: "before", project: Self.project, source: "synthetic-fixture", + embedding: Vector(vector), createdAt: Self.oldDate, + lastAccessedAt: Self.oldDate, importance: 1, + deletedAt: store == owner && tombstoned ? Self.oldDate : nil, + modifiedAt: Self.oldDate) + try writer.add(memory) + writers[store] = writer + paths[store] = path + ids[store] = try #require(memory.globalId) + } + let local = writers[.local]! + try local.add(SyncConfig(project: Self.project, policy: .sync)) + let group = MemoryTools.GroupSpokeRef(groupId: UUID(), path: paths[.group]!.path, + ref: writers[.group]!.sendableReference) + let tools = MemoryTools(localRef: local.sendableReference, + syncedRef: writers[.synced]!.sendableReference, + groupRefs: [group], embedder: FixedEmbedder(), + identity: StaticIdentityProvider(.anonymous)) + return Fixture(tools: tools, writers: writers, paths: paths, ids: ids, + owner: owner, query: query) + } + + private func assertCommitted(_ fixture: Fixture, before: [Owner: [UUID: SavedMemory]], + since: Date) throws { + let after = try fixture.snapshot() + let target = try #require(after[fixture.owner]?[fixture.targetID]) + #expect(target.topic == "after") + #expect(target.importance == 5) + let modifiedAt = try #require(target.modifiedAt) + #expect(modifiedAt >= since.timeIntervalSince1970 - 0.001) + #expect(modifiedAt <= Date().timeIntervalSince1970 + 0.001) + #expect(target.accessedAt >= since.timeIntervalSince1970 - 0.001) + for store in Owner.allCases { + #expect(Set(after[store]!.keys) == Set(before[store]!.keys)) + if store != fixture.owner { #expect(after[store] == before[store]) } + } + } + + @Test(arguments: UpdatePersistenceTests.routes) + func metadataSuccessCommitsToExactOwner(_ route: String) async throws { + let fixture = try fixture(route) + let before = try fixture.snapshot() + let started = Date() + let result = try await fixture.tools.handle(fixture.request()) + #expect(result.isError != true) + #expect(text(from: result).contains("Updated memory")) + #expect(text(from: result).contains(fixture.targetID.uuidString)) + try assertCommitted(fixture, before: before, since: started) + } + + @Test(arguments: UpdatePersistenceTests.routes) + func busyBeforeBeginReturnsExactNoWriteReceipt(_ route: String) async throws { + let fixture = try fixture(route) + // Warm a query union before taking the storage lock. This also proves + // the same exact route is writable when no external writer owns it. + let control = try await fixture.tools.handle(fixture.request(topic: "before", importance: 1)) + #expect(control.isError != true) + let before = try fixture.snapshot() + let blocker = try SQL(fixture.ownerPath) + try blocker.execute("BEGIN IMMEDIATE") + defer { try? blocker.execute("ROLLBACK") } + let started = Date() + let result = try await fixture.tools.handle(fixture.request()) + #expect(Date().timeIntervalSince(started) < 5) + #expect(result.isError == true) + #expect(result.structuredContent == .object([ + "engram_write_receipt": .object([ + "schema_version": .int(1), "tool": .string("update"), + "write_outcome": .string("not_stored_transaction_not_started"), + "reason": .string("database_busy"), "memory_ids": .array([]), + ]), + ])) + #expect(text(from: result) == "Memory was not updated: the database was busy before the write transaction started. Retry on a later turn.") + #expect(try fixture.snapshot() == before) + try blocker.execute("ROLLBACK") + let recoveredAt = Date() + let recovered = try await fixture.tools.handle(fixture.request()) + #expect(recovered.isError != true) + try assertCommitted(fixture, before: before, since: recoveredAt) + } + + @Test(arguments: UpdatePersistenceTests.routes, ["abort", "rollback", "begin-busy-text"]) + func laterSetterFailureThrowsAndRollsBackEarlierEdits(_ route: String, _ fault: String) async throws { + let fixture = try fixture(route) + let before = try fixture.snapshot() + let blocker = try SQL(fixture.ownerPath) + let action = fault == "rollback" ? "ROLLBACK" : "ABORT" + let detail = fault == "begin-busy-text" ? "Failed to begin transaction: database is locked" : "synthetic-update-body-fault" + // modifiedAt is assigned after topic, importance and lastAccessedAt. + // The failed setter must roll all earlier assignments back as well. + try blocker.execute("CREATE TRIGGER update_metadata_fault BEFORE UPDATE OF modifiedAt ON Memory BEGIN SELECT RAISE(\(action), '\(detail)'); END") + defer { try? blocker.execute("DROP TRIGGER IF EXISTS update_metadata_fault") } + await #expect(throws: (any Error).self) { + _ = try await fixture.tools.handle(fixture.request()) + } + #expect(try fixture.snapshot() == before) + try blocker.execute("DROP TRIGGER update_metadata_fault") + let recoveredAt = Date() + let recovered = try await fixture.tools.handle(fixture.request()) + #expect(recovered.isError != true) + try assertCommitted(fixture, before: before, since: recoveredAt) + } + + @Test(arguments: ["edge-fault", "begin-busy-text"]) + func undeleteGraphFailureDoesNotAttestNoWriteAfterPrimaryCommit(_ fault: String) async throws { + let fixture = try fixture("local-id", tombstoned: true) + let edge = Edge(sourceGlobalId: fixture.targetID, targetGlobalId: fixture.ids[.synced]!, + relation: .relatesTo, deletedAt: Self.oldDate) + try fixture.writers[.local]!.add(edge) + let before = try fixture.snapshot() + let edgesBefore = try SQL(fixture.ownerPath, readOnly: true).edges() + let blocker = try SQL(fixture.ownerPath) + let detail = fault == "begin-busy-text" ? "Failed to begin transaction: database is locked" : "synthetic-edge-restore-fault" + try blocker.execute("CREATE TRIGGER update_edge_fault BEFORE UPDATE OF deletedAt ON Edge BEGIN SELECT RAISE(ABORT, '\(detail)'); END") + defer { try? blocker.execute("DROP TRIGGER IF EXISTS update_edge_fault") } + let started = Date() + // A later graph failure is a partial outcome, even if its text mimics + // a BEGIN failure. It must throw, not report Updated or known no-write. + await #expect(throws: (any Error).self) { + _ = try await fixture.tools.handle(CallTool.Parameters(name: "update", arguments: [ + "id": .string(fixture.targetID.uuidString), "undelete": .bool(true), + ])) + } + let after = try fixture.snapshot() + let restored = try #require(after[.local]?[fixture.targetID]) + #expect(restored.deletedAt == nil) + let modifiedAt = try #require(restored.modifiedAt) + #expect(modifiedAt >= started.timeIntervalSince1970 - 0.001) + #expect(restored.topic == "before") + #expect(try SQL(fixture.ownerPath, readOnly: true).edges() == edgesBefore) + #expect(after[.synced] == before[.synced]) + #expect(after[.group] == before[.group]) + } +} diff --git a/Tests/codex_plugin/test_receipts.py b/Tests/codex_plugin/test_receipts.py index dc7060a..9f1aadc 100644 --- a/Tests/codex_plugin/test_receipts.py +++ b/Tests/codex_plugin/test_receipts.py @@ -118,10 +118,13 @@ def test_malformed_no_write_receipts_fail_both_completion_and_reconciliation(sel (path / 'mcp-audit.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in rows)) self.assertIsNotNone(F.RUNNER.failure_reconciliation(path, True)) - def begin_busy(self): - return {'isError': True, 'content': [{'type': 'text', 'text': PROXY.BEGIN_BUSY_TEXT}], + def begin_busy(self, tool='remember'): + text = ('Memory was not updated: the database was busy before the write transaction started. Retry on a later turn.' + if tool == 'update' else + 'Memory was not stored: the database was busy before the write transaction started. Retry on a later turn.') + return {'isError': True, 'content': [{'type': 'text', 'text': text}], 'structuredContent': {'engram_write_receipt': { - 'schema_version': 1, 'tool': 'remember', + 'schema_version': 1, 'tool': tool, 'write_outcome': 'not_stored_transaction_not_started', 'reason': 'database_busy', 'memory_ids': []}}} @@ -144,6 +147,20 @@ def test_begin_busy_is_failed_but_replay_has_no_unverified_write(self): _, denied = self.call(2, 'remember', {'content': 'Retry'}) self.assertTrue(denied['result']['isError']) + def test_update_begin_busy_is_failed_without_a_reconciliation_gate(self): + self.call(1, 'update', {'id': ID, 'importance': 3}) + output, _ = self.busy_reply(1, self.begin_busy('update')) + self.assertIs(output['result']['isError'], True) + self.assertEqual(self.audit.rows[-1], { + 'event': 'tool_result', 'id': 1, 'tool': 'update', 'ok': False, + 'forwarded': True, 'memory_ids': [], + 'write_outcome': 'not_stored_transaction_not_started', 'write_outcome_version': 1}) + self.assertEqual(self.audit_result(), {'tool_calls': 1, 'write_calls': 0, + 'writes': [], 'tool_errors': 1}) + self.assertIsNone(self.reconciliation([*self.timeout_prefix(), *self.audit.rows])) + _, denied = self.call(2, 'update', {'id': ID, 'importance': 4}) + self.assertTrue(denied['result']['isError']) + def test_begin_busy_contract_rejects_text_spoofs_and_contradictions(self): base = self.begin_busy() candidates = [] @@ -186,7 +203,7 @@ def test_begin_busy_receipt_mutations_remain_uncertain(self): mutations = [{'write_outcome_version': True}, {'write_outcome_version': 1.0}, {'write_outcome_version': 2}, {'write_outcome': 'unknown'}, {'ok': True}, {'ok': 0}, {'forwarded': False}, {'forwarded': 1}, - {'memory_ids': [ID]}, {'memory_ids': None}, {'tool': 'update'}, {'extra': True}] + {'memory_ids': [ID]}, {'memory_ids': None}, {'tool': 'connect'}, {'extra': True}] candidates = [dict(valid, **change) for change in mutations] candidates.extend({k: v for k, v in valid.items() if k != removed} for removed in valid) for candidate in candidates: @@ -194,24 +211,46 @@ def test_begin_busy_receipt_mutations_remain_uncertain(self): self.assertFalse(PROXY.verified_no_write_receipt(candidate)) self.assertIsNotNone(self.reconciliation([*self.timeout_prefix(), self.audit.rows[0], candidate])) + def test_begin_busy_audit_tool_must_match_the_forwarded_call(self): + for tool, other in (('remember', 'update'), ('update', 'remember')): + with self.subTest(tool=tool): + self.audit = MemoryAudit(); self.policy = PROXY.Policy(self.audit, 'codex-session:SOURCE', 3, 1) + self.call(1, tool, {'content': 'Synthetic'} if tool == 'remember' else {'id': ID, 'importance': 3}) + self.busy_reply(1, self.begin_busy(tool)) + self.audit.rows[-1]['tool'] = other + self.assertIsNotNone(self.reconciliation([*self.timeout_prefix(), *self.audit.rows])) + + def test_update_uncertain_body_or_commit_failure_still_requires_reconciliation(self): + self.call(1, 'update', {'id': ID, 'importance': 3}) + self.reply(1, 'Storage failed after transaction entry; write outcome uncertain.', error=True) + self.assertNotIn('write_outcome', self.audit.rows[-1]) + self.assertEqual(self.reconciliation([*self.timeout_prefix(), *self.audit.rows]), + {'reason': 'successful_or_unverified_write', 'memory_ids': []}) + def test_safe_begin_failure_cannot_erase_prior_success_or_uncertainty(self): - for prior in (None, {'ok': False, 'memory_ids': []}, {'ok': True, 'memory_ids': [ID]}): - with self.subTest(prior=prior): - self.audit = MemoryAudit(); self.policy = PROXY.Policy(self.audit, 'codex-session:SOURCE', 3, 2) - self.call(2, 'remember', {'content': 'Synthetic'}); self.busy_reply(2) - rows = [*self.timeout_prefix(), {'event': 'tool_call', 'id': 1, 'tool': 'remember'}] - if prior is not None: - rows.append({'event': 'tool_result', 'id': 1, 'tool': 'remember', - 'forwarded': True, **prior}) - rows.extend(self.audit.rows) - gate = self.reconciliation(rows) - self.assertEqual(gate['reason'], 'successful_or_unverified_write') - self.assertEqual(gate['memory_ids'], [ID] if prior and prior['ok'] else []) + for tool in ('remember', 'update'): + for prior in (None, {'ok': False, 'memory_ids': []}, {'ok': True, 'memory_ids': [ID]}): + with self.subTest(tool=tool, prior=prior): + self.audit = MemoryAudit(); self.policy = PROXY.Policy(self.audit, 'codex-session:SOURCE', 3, 2) + self.call(2, tool, {'content': 'Synthetic'} if tool == 'remember' else {'id': ID, 'importance': 3}) + self.busy_reply(2, self.begin_busy(tool)) + rows = [*self.timeout_prefix(), {'event': 'tool_call', 'id': 1, 'tool': 'remember'}] + if prior is not None: + rows.append({'event': 'tool_result', 'id': 1, 'tool': 'remember', + 'forwarded': True, **prior}) + rows.extend(self.audit.rows) + gate = self.reconciliation(rows) + self.assertEqual(gate['reason'], 'successful_or_unverified_write') + self.assertEqual(gate['memory_ids'], [ID] if prior and prior['ok'] else []) def test_post_timeout_cleanup_cannot_claim_native_begin_receipt(self): - self.call(1, 'remember', {'content': 'Synthetic'}); self.busy_reply(1) - rows = [*self.timeout_prefix(), self.audit.rows[0], {'event': 'request_timeout'}, self.audit.rows[1]] - self.assertEqual(self.reconciliation(rows)['reason'], 'write_status_unknown') + for tool in ('remember', 'update'): + with self.subTest(tool=tool): + self.audit = MemoryAudit(); self.policy = PROXY.Policy(self.audit, 'codex-session:SOURCE', 3, 1) + self.call(1, tool, {'content': 'Synthetic'} if tool == 'remember' else {'id': ID, 'importance': 3}) + self.busy_reply(1, self.begin_busy(tool)) + rows = [*self.timeout_prefix(), self.audit.rows[0], {'event': 'request_timeout'}, self.audit.rows[1]] + self.assertEqual(self.reconciliation(rows)['reason'], 'write_status_unknown') def test_update_receipt_requires_requested_exact_uuid(self): self.call(1, 'update', {'id': ID, 'content': 'Synthetic'}) diff --git a/codex/plugins/engram/.codex-plugin/plugin.json b/codex/plugins/engram/.codex-plugin/plugin.json index cb7e9ff..fd93ce4 100644 --- a/codex/plugins/engram/.codex-plugin/plugin.json +++ b/codex/plugins/engram/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "engram", - "version": "0.14.9", + "version": "0.14.10", "description": "Engram memory tools, skills, contextual advice, and session learning for Codex.", "author": { "name": "Engram" diff --git a/codex/plugins/engram/scripts/codex_learner/memory_proxy.py b/codex/plugins/engram/scripts/codex_learner/memory_proxy.py index 988d33d..af0fffe 100644 --- a/codex/plugins/engram/scripts/codex_learner/memory_proxy.py +++ b/codex/plugins/engram/scripts/codex_learner/memory_proxy.py @@ -144,6 +144,7 @@ def rpc_error(request_id: Any, message: str) -> dict[str, Any]: BEGIN_BUSY_OUTCOME = "not_stored_transaction_not_started" BEGIN_BUSY_VERSION = 1 BEGIN_BUSY_TEXT = "Memory was not stored: the database was busy before the write transaction started. Retry on a later turn." +BEGIN_BUSY_UPDATE_TEXT = "Memory was not updated: the database was busy before the write transaction started. Retry on a later turn." NEAR_DUPLICATE_PREFIX = "⚠️ Near-duplicate memory detected. The new memory was NOT stored.\n\nExisting similar memories:" NEAR_DUPLICATE_SUFFIX = ('\n\nTo resolve:' '\n - Use `update(id: "UUID", ...)` to modify the existing memory' @@ -168,16 +169,17 @@ def verified_no_write_response(tool: str, result: Any) -> bool: def verified_begin_busy_response(tool: str, result: Any) -> bool: - """Accept only native remember's versioned pre-BEGIN no-write contract. + """Accept only native remember/update versioned pre-BEGIN no-write contracts. Text alone, nested quoted content, and any unknown or contradictory field cannot establish no-write. The native producer checks body entry, not just an error string; post-entry/commit failures never produce this receipt. """ - if (tool != "remember" or not isinstance(result, dict) + expected_text = BEGIN_BUSY_UPDATE_TEXT if tool == "update" else BEGIN_BUSY_TEXT + if (tool not in ("remember", "update") or not isinstance(result, dict) or set(result) != {"isError", "content", "structuredContent"} or result["isError"] is not True - or result["content"] != [{"type": "text", "text": BEGIN_BUSY_TEXT}]): + or result["content"] != [{"type": "text", "text": expected_text}]): return False structured = result["structuredContent"] if not isinstance(structured, dict) or set(structured) != {"engram_write_receipt"}: @@ -186,7 +188,7 @@ def verified_begin_busy_response(tool: str, result: Any) -> bool: return (isinstance(receipt, dict) and set(receipt) == {"schema_version", "tool", "write_outcome", "reason", "memory_ids"} and type(receipt["schema_version"]) is int and receipt["schema_version"] == BEGIN_BUSY_VERSION - and receipt["tool"] == "remember" and receipt["write_outcome"] == BEGIN_BUSY_OUTCOME + and receipt["tool"] == tool and receipt["write_outcome"] == BEGIN_BUSY_OUTCOME and receipt["reason"] == "database_busy" and isinstance(receipt["memory_ids"], list) and receipt["memory_ids"] == []) @@ -195,13 +197,14 @@ def verified_no_write_receipt(entry: Any) -> bool: """Exact gateway metadata shared by completion and failed-run reconciliation.""" fields = {"event", "id", "tool", "ok", "memory_ids", "forwarded", "write_outcome"} if (not isinstance(entry, dict) or entry.get("event") != "tool_result" - or not valid_id(entry.get("id")) or entry.get("tool") != "remember" + or not valid_id(entry.get("id")) or entry.get("forwarded") is not True or not isinstance(entry.get("memory_ids"), list) or entry["memory_ids"] != []): return False if entry.get("write_outcome") == NO_WRITE_OUTCOME: - return set(entry) == fields and entry.get("ok") is True - return (set(entry) == fields | {"write_outcome_version"} + return set(entry) == fields and entry.get("tool") == "remember" and entry.get("ok") is True + return (entry.get("tool") in ("remember", "update") + and set(entry) == fields | {"write_outcome_version"} and entry.get("ok") is False and entry.get("write_outcome") == BEGIN_BUSY_OUTCOME and type(entry.get("write_outcome_version")) is int and entry["write_outcome_version"] == BEGIN_BUSY_VERSION) diff --git a/scripts/codex_learner/memory_proxy.py b/scripts/codex_learner/memory_proxy.py index 7f4df0b..33f6d64 100644 --- a/scripts/codex_learner/memory_proxy.py +++ b/scripts/codex_learner/memory_proxy.py @@ -141,6 +141,10 @@ def rpc_error(request_id: Any, message: str) -> dict[str, Any]: # branch in MemoryTools+Core.swift. Match the outer native receipt, never a # phrase or UUID quoted inside remembered content. NO_WRITE_OUTCOME = "not_stored_near_duplicate" +BEGIN_BUSY_OUTCOME = "not_stored_transaction_not_started" +BEGIN_BUSY_VERSION = 1 +BEGIN_BUSY_TEXT = "Memory was not stored: the database was busy before the write transaction started. Retry on a later turn." +BEGIN_BUSY_UPDATE_TEXT = "Memory was not updated: the database was busy before the write transaction started. Retry on a later turn." NEAR_DUPLICATE_PREFIX = "⚠️ Near-duplicate memory detected. The new memory was NOT stored.\n\nExisting similar memories:" NEAR_DUPLICATE_SUFFIX = ('\n\nTo resolve:' '\n - Use `update(id: "UUID", ...)` to modify the existing memory' @@ -164,15 +168,46 @@ def verified_no_write_response(tool: str, result: Any) -> bool: and text.endswith(NEAR_DUPLICATE_SUFFIX)) +def verified_begin_busy_response(tool: str, result: Any) -> bool: + """Accept only native remember/update versioned pre-BEGIN no-write contracts. + + Text alone, nested quoted content, and any unknown or contradictory field + cannot establish no-write. The native producer checks body entry, not just + an error string; post-entry/commit failures never produce this receipt. + """ + expected_text = BEGIN_BUSY_UPDATE_TEXT if tool == "update" else BEGIN_BUSY_TEXT + if (tool not in ("remember", "update") or not isinstance(result, dict) + or set(result) != {"isError", "content", "structuredContent"} + or result["isError"] is not True + or result["content"] != [{"type": "text", "text": expected_text}]): + return False + structured = result["structuredContent"] + if not isinstance(structured, dict) or set(structured) != {"engram_write_receipt"}: + return False + receipt = structured["engram_write_receipt"] + return (isinstance(receipt, dict) + and set(receipt) == {"schema_version", "tool", "write_outcome", "reason", "memory_ids"} + and type(receipt["schema_version"]) is int and receipt["schema_version"] == BEGIN_BUSY_VERSION + and receipt["tool"] == tool and receipt["write_outcome"] == BEGIN_BUSY_OUTCOME + and receipt["reason"] == "database_busy" + and isinstance(receipt["memory_ids"], list) and receipt["memory_ids"] == []) + + def verified_no_write_receipt(entry: Any) -> bool: """Exact gateway metadata shared by completion and failed-run reconciliation.""" - return (isinstance(entry, dict) - and set(entry) == {"event", "id", "tool", "ok", "memory_ids", "forwarded", "write_outcome"} - and entry.get("event") == "tool_result" and valid_id(entry.get("id")) - and entry.get("tool") == "remember" and entry.get("ok") is True - and entry.get("forwarded") is True - and isinstance(entry.get("memory_ids"), list) and entry["memory_ids"] == [] - and entry.get("write_outcome") == NO_WRITE_OUTCOME) + fields = {"event", "id", "tool", "ok", "memory_ids", "forwarded", "write_outcome"} + if (not isinstance(entry, dict) or entry.get("event") != "tool_result" + or not valid_id(entry.get("id")) + or entry.get("forwarded") is not True + or not isinstance(entry.get("memory_ids"), list) or entry["memory_ids"] != []): + return False + if entry.get("write_outcome") == NO_WRITE_OUTCOME: + return set(entry) == fields and entry.get("tool") == "remember" and entry.get("ok") is True + return (entry.get("tool") in ("remember", "update") + and set(entry) == fields | {"write_outcome_version"} + and entry.get("ok") is False and entry.get("write_outcome") == BEGIN_BUSY_OUTCOME + and type(entry.get("write_outcome_version")) is int + and entry["write_outcome_version"] == BEGIN_BUSY_VERSION) def verified_write_ids(tool: str, arguments: dict[str, Any], result: dict[str, Any]) -> list[str]: @@ -209,10 +244,12 @@ def __init__(self, audit: Audit, provenance: str, max_tool_calls: int, max_write self.pending: dict[tuple[type, Any], dict[str, Any]] = {} self.used_ids: set[tuple[type, Any]] = set() - def record_result(self, request_id: Any, tool: str, ok: bool, memory_ids: list[str] | None = None, *, write_outcome: str | None = None) -> None: + def record_result(self, request_id: Any, tool: str, ok: bool, memory_ids: list[str] | None = None, *, write_outcome: str | None = None, write_outcome_version: int | None = None) -> None: entry = {"event": "tool_result", "id": request_id, "tool": tool, "ok": ok, "memory_ids": memory_ids or []} if write_outcome is not None: entry.update(write_outcome=write_outcome, forwarded=True) + if write_outcome_version is not None: + entry["write_outcome_version"] = write_outcome_version self.audit.write(entry) def deny_tool(self, request_id: Any, tool: str, reason: str) -> tuple[None, dict[str, Any]]: @@ -307,11 +344,13 @@ def server_message(self, message: Any) -> tuple[dict[str, Any] | None, dict[str, tool = pending["tool"] ok = "error" not in message and isinstance(result, dict) and result.get("isError", False) is False no_write = ok and verified_no_write_response(tool, result) + begin_busy = "error" not in message and verified_begin_busy_response(tool, result) memory_ids = verified_write_ids(tool, pending["arguments"], result) if ok and tool in WRITE_TOOLS and not no_write else [] if tool in WRITE_TOOLS and not memory_ids and not no_write: ok = False self.record_result(request_id, tool, ok, memory_ids if ok else [], - write_outcome=NO_WRITE_OUTCOME if no_write else None) + write_outcome=NO_WRITE_OUTCOME if no_write else (BEGIN_BUSY_OUTCOME if begin_busy else None), + write_outcome_version=BEGIN_BUSY_VERSION if begin_busy else None) if not ok and "error" not in message and isinstance(result, dict) and result.get("isError", False) is False: return tool_error(request_id, "write was not verified by an Engram receipt; do not retry blindly"), None return message, None diff --git a/scripts/tests/test_codex_learner_runner.py b/scripts/tests/test_codex_learner_runner.py index 736af30..e726933 100644 --- a/scripts/tests/test_codex_learner_runner.py +++ b/scripts/tests/test_codex_learner_runner.py @@ -614,6 +614,20 @@ def test_provider_known_conflict_cannot_support_stored_claim(self): self.assertEqual(result["status"], "failed") self.assertEqual(result["writes"], []) + def test_native_begin_busy_stays_failed_without_claiming_a_write(self): + # Standalone has ordinary failed/backoff handling, not a permanent + # reconciliation gate. A known no-write failure is still not success. + for tool in ("remember", "update"): + with self.subTest(tool=tool): + rows = [{"event": "tool_call", "id": 1, "tool": tool}, + {"event": "tool_result", "id": 1, "tool": tool, "ok": False, + "forwarded": True, "memory_ids": [], + "write_outcome": "not_stored_transaction_not_started", "write_outcome_version": 1}] + result = self.provider_run(events=[{"type": "turn.completed"}], audit_rows=rows) + self.assertEqual(result["status"], "failed") + self.assertEqual(result["tool_errors"], 1) + self.assertEqual(result["writes"], []) + def test_provider_mixed_verified_write_and_conflict_counts_only_write(self): rows = self.no_write_rows() + [ {"event": "tool_call", "id": 2, "tool": "remember"}, diff --git a/scripts/tests/test_codex_memory_proxy.py b/scripts/tests/test_codex_memory_proxy.py index 38bbea4..33d94b9 100644 --- a/scripts/tests/test_codex_memory_proxy.py +++ b/scripts/tests/test_codex_memory_proxy.py @@ -32,6 +32,8 @@ '\n - Use `remember(..., force: true)` to keep both' '\n - Use `forget(id: "UUID")` to remove the old one, then `remember` the new one') CONFLICT_TEXT = CONFLICT_PREFIX + f"\n [id:{ID_A}] (distance: 0.123, term overlap: 90%) PRIVATE_CONFLICT_SENTINEL" + CONFLICT_SUFFIX +BEGIN_BUSY_TEXT = "Memory was not stored: the database was busy before the write transaction started. Retry on a later turn." +BEGIN_BUSY_UPDATE_TEXT = "Memory was not updated: the database was busy before the write transaction started. Retry on a later turn." class CaptureAudit: @@ -379,6 +381,129 @@ def test_parser_contract_matches_native_source_and_packaged_copy(self): self.assertEqual(module.NEAR_DUPLICATE_SUFFIX, CONFLICT_SUFFIX) +class BeginBusyContractTests(unittest.TestCase): + def native(self, tool): + return {"jsonrpc": "2.0", "id": 1, "result": { + "isError": True, "content": [{"type": "text", "text": BEGIN_BUSY_UPDATE_TEXT if tool == "update" else BEGIN_BUSY_TEXT}], + "structuredContent": {"engram_write_receipt": { + "schema_version": 1, "tool": tool, + "write_outcome": "not_stored_transaction_not_started", + "reason": "database_busy", "memory_ids": []}}}} + + def exercise(self, module, tool, native): + audit = CaptureAudit() + policy = module.Policy(audit, "fixture", 3, 1) + arguments = {"content": "SYNTHETIC_BODY"} if tool == "remember" else {"id": ID_A, "importance": 3} + forwarded, denied = policy.client_message(call(1, tool, arguments)) + self.assertIsNone(denied) + self.assertIsNotNone(forwarded) + output, _ = policy.server_message(native) + return audit, policy, output + + def test_exact_begin_failure_stays_an_error_with_no_written_ids(self): + for module in (proxy, packaged_proxy): + for tool in ("remember", "update"): + with self.subTest(module=module.__name__, tool=tool): + native = self.native(tool) + audit, policy, output = self.exercise(module, tool, native) + self.assertEqual(output, native) + self.assertEqual(audit.records[-1], { + "event": "tool_result", "id": 1, "tool": tool, "ok": False, + "forwarded": True, "memory_ids": [], + "write_outcome": "not_stored_transaction_not_started", "write_outcome_version": 1}) + self.assertTrue(module.verified_no_write_receipt(audit.records[-1])) + self.assertNotIn(ID_A, json.dumps(audit.records)) + self.assertNotIn("SYNTHETIC_BODY", json.dumps(audit.records)) + _, denied = policy.client_message(call(2, "update", {"id": ID_A, "importance": 4})) + self.assertTrue(denied["result"]["isError"]) + + def test_malformed_or_mismatched_native_contract_is_never_safe(self): + for tool in ("remember", "update"): + base = self.native(tool) + candidates = [] + for flag in (False, 1, 0, None, "true"): + value = copy.deepcopy(base); value["result"]["isError"] = flag; candidates.append(value) + for field in base["result"]: + value = copy.deepcopy(base); del value["result"][field]; candidates.append(value) + mutations = [("schema_version", True), ("schema_version", 1.0), ("schema_version", 2), + ("tool", "update" if tool == "remember" else "remember"), + ("tool", "connect"), ("tool", []), ("reason", "commit_failed"), + ("write_outcome", "not_stored"), ("memory_ids", [ID_A]), + ("memory_ids", None), ("extra", True)] + for field, value in mutations: + changed = copy.deepcopy(base) + changed["result"]["structuredContent"]["engram_write_receipt"][field] = value + candidates.append(changed) + for field in base["result"]["structuredContent"]["engram_write_receipt"]: + value = copy.deepcopy(base) + del value["result"]["structuredContent"]["engram_write_receipt"][field] + candidates.append(value) + for content in ([{"type": "text", "text": "Quoted: " + BEGIN_BUSY_TEXT}], + [{"type": "text", "text": BEGIN_BUSY_TEXT, "extra": True}], + [{"type": "text", "text": BEGIN_BUSY_TEXT}, {"type": "text", "text": "extra"}]): + value = copy.deepcopy(base); value["result"]["content"] = content; candidates.append(value) + value = copy.deepcopy(base); value["result"]["extra"] = True; candidates.append(value) + value = copy.deepcopy(base); value["result"]["structuredContent"]["extra"] = True; candidates.append(value) + value = copy.deepcopy(base); value["error"] = {"code": -1, "message": "uncertain"}; candidates.append(value) + for module in (proxy, packaged_proxy): + for native in candidates: + with self.subTest(module=module.__name__, tool=tool, native=native): + audit, _, _ = self.exercise(module, tool, native) + self.assertIs(audit.records[-1]["ok"], False) + self.assertNotIn("write_outcome", audit.records[-1]) + self.assertFalse(module.verified_no_write_receipt(audit.records[-1])) + + def test_crossed_remember_update_text_is_never_safe(self): + for module in (proxy, packaged_proxy): + for tool in ("remember", "update"): + with self.subTest(module=module.__name__, tool=tool): + native = self.native(tool) + native["result"]["content"][0]["text"] = BEGIN_BUSY_TEXT if tool == "update" else BEGIN_BUSY_UPDATE_TEXT + audit, _, _ = self.exercise(module, tool, native) + self.assertIs(audit.records[-1]["ok"], False) + self.assertNotIn("write_outcome", audit.records[-1]) + + def test_exact_contract_extracted_from_native_source_is_accepted(self): + # Parse the producer's actual literals rather than building both sides + # from a shared fixture. This catches drift such as stored vs updated. + source = (REPO_ROOT / "Sources/EngramKit/MemoryTools+Core.swift").read_text() + pattern = (r'content: \[\.text\(("(?:\\.|[^"\\])*")\)\],\s*' + r'structuredContent: \.object\(\[\s*"engram_write_receipt": \.object\(\[' + r'(.*?)\]\),\s*\]\),\s*isError: true') + contracts = {} + for text, fields in re.findall(pattern, source, re.DOTALL): + receipt = {} + for name, kind, literal in re.findall(r'"(\w+)": \.(int|string|array)\((.*?)\),', fields): + self.assertEqual(kind, {"schema_version": "int", "memory_ids": "array"}.get(name, "string")) + receipt[name] = json.loads(literal) + contracts[receipt["tool"]] = {"isError": True, + "content": [{"type": "text", "text": json.loads(text)}], + "structuredContent": {"engram_write_receipt": receipt}} + self.assertEqual(set(contracts), {"remember", "update"}) + for tool, contract in contracts.items(): + self.assertEqual(contract, self.native(tool)["result"]) + for module in (proxy, packaged_proxy): + with self.subTest(module=module.__name__, tool=tool): + audit, _, _ = self.exercise(module, tool, {"jsonrpc": "2.0", "id": 1, "result": contract}) + self.assertTrue(module.verified_no_write_receipt(audit.records[-1])) + + def test_exact_audit_contract_rejects_unsupported_tools_and_type_coercions(self): + for module in (proxy, packaged_proxy): + for tool in ("remember", "update"): + audit, _, _ = self.exercise(module, tool, self.native(tool)) + valid = audit.records[-1] + changes = [{"tool": "connect"}, {"tool": []}, {"ok": True}, {"ok": 0}, + {"id": True}, {"forwarded": False}, {"forwarded": 1}, + {"memory_ids": [ID_A]}, {"memory_ids": None}, + {"write_outcome_version": True}, {"write_outcome_version": 1.0}, + {"write_outcome_version": 2}, {"extra": True}] + candidates = [dict(valid, **change) for change in changes] + candidates += [{k: v for k, v in valid.items() if k != field} for field in valid] + for candidate in candidates: + with self.subTest(module=module.__name__, candidate=candidate): + self.assertFalse(module.verified_no_write_receipt(candidate)) + + class TransportTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory()