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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,14 @@ public struct EmbeddingRefreshService: Sendable {
}

/// Runs the refresh until nothing is stale, the task is cancelled, or the
/// environment turns hostile. Returns how many clips were re-embedded.
/// environment turns hostile. Returns how many vectors were actually committed.
/// Sources without ContentBoundEmbeddingStoring skip the pass rather than use
/// the retained, unconditional legacy write requirement.
@discardableResult
public func run(store: any EmbeddingRefreshSource) async -> Int {
guard let embedder = makeEmbedder() else { return 0 }
guard let embeddingStore = store as? any ContentBoundEmbeddingStoring,
let embedder = makeEmbedder()
else { return 0 }
var refreshed = 0
while !Task.isCancelled, isEnvironmentSuitable() {
guard let batch = try? await store.staleEmbeddingClipIDs(limit: Self.batchSize),
Expand All @@ -82,7 +86,8 @@ public struct EmbeddingRefreshService: Sendable {
if Task.isCancelled || !isEnvironmentSuitable() { return refreshed }
guard case .text(let text)? = try? await store.content(for: id),
let vector = try? embedder.vector(for: String(text.prefix(Self.inputLimit))),
(try? await store.saveEmbedding(clipID: id, vector: vector)) != nil
(try? await embeddingStore.saveEmbeddingIfCurrent(
clipID: id, vector: vector, expectedText: text)) == true
else { continue }
refreshed += 1
progressed = true
Expand Down
35 changes: 23 additions & 12 deletions Packages/GanchoKit/Sources/GanchoAppCore/EnrichmentService.swift
Original file line number Diff line number Diff line change
Expand Up @@ -20,17 +20,26 @@ import GanchoKit
/// `refreshRecents()`. The caller passes that as `onTitleWritten`, which runs on
/// the caller's actor immediately after the write, in the original position.
///
/// Stateless and `Sendable`: it constructs `ImageTextExtractor` /
/// `TieredClipAnnotator` / `ContextualSentenceEmbedder` per call exactly as the
/// inlined code did (the process-lifetime-instance optimization is deliberately
/// NOT taken here — this is a pure move). Each per-call helper is created and
/// consumed inside its own `if` with no `await` between construction and last
/// use, so no non-`Sendable` value crosses an actor boundary and the type needs no
/// `@MainActor` isolation. On CI the annotator degrades to the heuristic and the
/// embedder reports `hasAvailableAssets == false`, so those stages yield nothing
/// there; the plan/`writeTitle` gating remains unit-testable.
/// Sendable configuration constructs each non-Sendable helper only for its own
/// stage, without suspending during model computation. The injected embedder
/// factory gives race tests deterministic vectors without requiring model assets.
/// Background vector writes require ContentBoundEmbeddingStoring; legacy stores
/// without that optional capability still receive title/OCR work but skip vectors.
public struct EnrichmentService: Sendable {
public init() {}
private let makeEmbedder: @Sendable () -> (any TextEmbedding)?

public init() {
self.init(makeEmbedder: {
guard let embedder = ContextualSentenceEmbedder(), embedder.hasAvailableAssets else {
return nil
}
return embedder
})
}

init(makeEmbedder: @escaping @Sendable () -> (any TextEmbedding)?) {
self.makeEmbedder = makeEmbedder
}

/// Runs the enrichment IO shared by both shells, in the SAME order as the
/// inlined code: OCR → title → embedding.
Expand Down Expand Up @@ -67,10 +76,12 @@ public struct EnrichmentService: Sendable {
}
// Semantic vector (the embedder caches its model after the first call).
if plan.runs(.embedding), case .text(let text)? = content,
let embedder = ContextualSentenceEmbedder(), embedder.hasAvailableAssets,
let embeddingStore = store as? any ContentBoundEmbeddingStoring,
let embedder = makeEmbedder(),
let vector = try? embedder.vector(for: String(text.prefix(1_000)))
{
_ = try? await store.saveEmbedding(clipID: item.id, vector: vector)
_ = try? await embeddingStore.saveEmbeddingIfCurrent(
clipID: item.id, vector: vector, expectedText: text)
}
}

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
import Foundation
import GRDB

/// Optional capability for background work derived from an earlier text snapshot.
/// The legacy ClipEnriching surface stays source-compatible; asynchronous producers
/// require this capability rather than falling back to an unconditional vector write.
public protocol ContentBoundEmbeddingStoring: Sendable {
/// Stores a vector only while the exact source body is still eligible and current.
/// Title/keyword-only edits do not invalidate a body-derived vector. Returning to
/// the same body is safe: the vector depends on that body, not its edit history.
@discardableResult
func saveEmbeddingIfCurrent(
clipID: UUID, vector: [Float], expectedText: String
) async throws -> Bool
}

extension GRDBClipboardStore {
/// Kinds background embedding never indexes: every masked-preview kind and the
/// payloads without a free-text body. Shared by the guarded write and the stale
/// refresh queue, so the queue never offers a row the write must reject.
static let embeddingIneligibleKinds: [String] = ClipContentKind.allCases.filter {
$0.prefersMaskedPreview || $0 == .image || $0 == .fileReference
}.map(\.rawValue)
}

extension GRDBClipboardStore: ContentBoundEmbeddingStoring {
public func saveEmbeddingIfCurrent(
clipID: UUID, vector: [Float], expectedText: String
) async throws -> Bool {
try Task.checkCancellation()
let data = vector.withUnsafeBufferPointer { Data(buffer: $0) }
let excludedKinds = Self.embeddingIneligibleKinds
let placeholders = excludedKinds.map { _ in "?" }.joined(separator: ", ")
return try await writer.write { db in
try Task.checkCancellation()
var arguments: [any DatabaseValueConvertible] = [
vector.count, data, EmbeddingModelInfo.currentVersion, clipID.uuidString,
expectedText, Date.now, "public.file-url"
]
arguments.append(contentsOf: excludedKinds)
// Visibility uses the shared read predicate: an expired row that
// retention keeps (pinned, boarded, or a snippet) stays searchable,
// so its current body must stay indexable too.
try db.execute(
sql: """
INSERT OR REPLACE INTO clip_embedding (clipID, dimension, vector, modelVersion)
SELECT clip.id, ?, ?, ? FROM clip
WHERE clip.id = ? AND clip.contentText = ? AND clip.contentBlobHash IS NULL
AND clip.isSensitive = 0 AND clip.isArchived = 0
AND \(Self.unexpiredPredicate)
AND (clip.contentTypeIdentifier IS NULL OR clip.contentTypeIdentifier != ?)
AND clip.kind NOT IN (\(placeholders))
""",
arguments: StatementArguments(arguments))
return db.changesCount == 1
}
}
}
14 changes: 11 additions & 3 deletions Packages/GanchoKit/Sources/GanchoKit/SemanticSearch.swift
Original file line number Diff line number Diff line change
Expand Up @@ -38,21 +38,29 @@ extension GRDBClipboardStore {
/// spend battery on rows no query can return. The sensitive exclusion is
/// belt-and-suspenders — capture never embeds sensitive clips, but this
/// query must not feed one to the re-embed loop even if some future path
/// flips `isSensitive` after a vector exists.
/// flips `isSensitive` after a vector exists. Kinds the guarded write
/// always rejects are excluded too: a permanently rejected row would stay
/// at the head of every batch and stall the pass for the rows behind it.
public func staleEmbeddingClipIDs(limit: Int) async throws -> [UUID] {
// SQLite treats a negative LIMIT as "no limit" — the exact unbounded
// read this API exists to prevent.
guard limit > 0 else { return [] }
let excludedKinds = Self.embeddingIneligibleKinds
let placeholders = excludedKinds.map { _ in "?" }.joined(separator: ", ")
return try await writer.read { db in
try String.fetchAll(
var arguments: [any DatabaseValueConvertible] = [EmbeddingModelInfo.currentVersion]
arguments.append(contentsOf: excludedKinds)
arguments.append(limit)
return try String.fetchAll(
db,
sql: """
SELECT e.clipID FROM clip_embedding e
JOIN clip c ON c.id = e.clipID
WHERE e.modelVersion < ? AND c.isArchived = 0 AND c.isSensitive = 0
AND c.kind NOT IN (\(placeholders))
LIMIT ?
""",
arguments: [EmbeddingModelInfo.currentVersion, limit]
arguments: StatementArguments(arguments)
).compactMap(UUID.init(uuidString:))
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import Testing
/// Scriptable refresh source: `stale` drains as saves land, exactly like the
/// real store (a refreshed row stops being stale), so batch-loop convergence
/// is exercised honestly.
private actor FakeRefreshStore: EmbeddingRefreshSource {
private actor FakeRefreshStore: EmbeddingRefreshSource, ContentBoundEmbeddingStoring {
private var stale: [UUID]
private let texts: [UUID: String]
private(set) var saved: [UUID: [Float]] = [:]
Expand All @@ -28,6 +28,13 @@ private actor FakeRefreshStore: EmbeddingRefreshSource {
saved[clipID] = vector
stale.removeAll { $0 == clipID }
}
func saveEmbeddingIfCurrent(
clipID: UUID, vector: [Float], expectedText: String
) async throws -> Bool {
guard texts[clipID] == expectedText else { return false }
try await saveEmbedding(clipID: clipID, vector: vector)
return true
}
}

/// Deterministic embedder — no model assets, no Accelerate.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
import ClipboardCore
import Foundation
import GanchoAI
import GanchoKit
import Testing

@testable import GanchoAppCore

private struct RaceEmbedder: TextEmbedding {
let dimension = 2
func vector(for text: String) throws -> [Float] { [1, 0] }
}

/// Holds the exact await between computation and guarded persistence. No timing
/// assumptions or on-device model assets are involved in either caller test.
private actor EmbeddingRaceStore: ClipEnriching, EmbeddingRefreshSource,
ContentBoundEmbeddingStoring
{
let id = UUID()
private var text = "original"
private var pendingWrite: CheckedContinuation<Void, Never>?
private var writeWaiters: [CheckedContinuation<Void, Never>] = []
private(set) var saved = false
private(set) var unsafeWrites = 0
private(set) var observedText: String?

func content(for id: UUID) async throws -> ClipContent? { .text(text) }
func staleEmbeddingClipIDs(limit: Int) async throws -> [UUID] { saved ? [] : [id] }
func updateTitle(id: UUID, title: String) async throws {}
func updateTitleIfEmpty(id: UUID, title: String) async throws -> Bool { false }
func attachExtractedText(id: UUID, text: String) async throws {}
func updateClipText(id: UUID, text: String) async throws { self.text = text }
func saveEmbedding(clipID: UUID, vector: [Float]) async throws {
unsafeWrites += 1
await pauseWrite()
saved = true
}

func saveEmbeddingIfCurrent(
clipID: UUID, vector: [Float], expectedText: String
) async throws -> Bool {
observedText = expectedText
await pauseWrite()
guard expectedText == text else { return false }
saved = true
return true
}

private func pauseWrite() async {
await withCheckedContinuation { continuation in
pendingWrite = continuation
for waiter in writeWaiters { waiter.resume() }
writeWaiters.removeAll()
}
}

func waitUntilWriteIsPending() async {
if pendingWrite != nil { return }
await withCheckedContinuation { writeWaiters.append($0) }
}

func resumeWrite(withCurrentText text: String) {
self.text = text
let continuation = pendingWrite
pendingWrite = nil
continuation?.resume()
}
}

@Suite("Background embedding write races", .timeLimit(.minutes(1)))
struct EmbeddingWriteRaceTests {
@Test func captureEnrichmentRejectsAChangedBody() async {
let store = EmbeddingRaceStore()
let item = ClipItem(id: store.id, preview: "original")
let plan = EnrichmentPlan(
content: .text(hasTitle: true), isSensitive: false, isPro: true,
preferences: IntelligencePreferences())
let service = EnrichmentService(makeEmbedder: { RaceEmbedder() })
let task = Task {
await service.enrich(
item, content: .text("original"), plan: plan, writeTitle: false, store: store,
onTitleWritten: {})
}
await store.waitUntilWriteIsPending()
await store.resumeWrite(withCurrentText: "edited")
await task.value
#expect(await store.observedText == "original")
#expect(await store.unsafeWrites == 0)
#expect(await store.saved == false)
}

@Test func refreshRejectsChangedBodyWithoutReportingProgressOrSpinning() async {
let store = EmbeddingRaceStore()
let service = EmbeddingRefreshService(
makeEmbedder: { RaceEmbedder() }, isEnvironmentSuitable: { true })
let task = Task { await service.run(store: store) }
await store.waitUntilWriteIsPending()
await store.resumeWrite(withCurrentText: "edited")
#expect(await task.value == 0)
#expect(await store.observedText == "original")
#expect(await store.unsafeWrites == 0)
#expect(await store.saved == false)
}

@Test func refreshAcceptsUnchangedBodyAsPositiveControl() async {
let store = EmbeddingRaceStore()
let service = EmbeddingRefreshService(
makeEmbedder: { RaceEmbedder() }, isEnvironmentSuitable: { true })
let task = Task { await service.run(store: store) }
await store.waitUntilWriteIsPending()
await store.resumeWrite(withCurrentText: "original")
#expect(await task.value == 1)
#expect(await store.saved)
#expect(await store.unsafeWrites == 0)
}
}
Loading
Loading