FluxFeed 0.24.3

FluxFeed

Document ingestion pipeline for .NET — track files, extract and chunk them, feed them to a vector index.

Overview

FluxFeed turns a folder of documents into an always-current search index. It watches the files, extracts and chunks them with FileFlux, and indexes the chunks into FluxIndex — then keeps the index in step as files change, move, or disappear.

FluxFeed owns ingestion only. Embedding, retrieval and ranking belong to FluxIndex, and the dependency is one-way: FluxFeed → FluxIndex.

Each tracked file gets a vault entry — a small git-backed directory holding the extracted text plus any notes you add by hand. Re-indexing is therefore cheap and auditable: you can diff a document's extracted content, see its commit history, and edit it without touching the original.

Features

• File-source vault — per-file git-tracked directory (refined.md, append-text.md, qa.md)
• Change detection — content hash for source changes, git status for vault edits
• Folder watching — real-time FileSystemWatcher with debounce and glob include/exclude patterns
• Background queue — bounded concurrency, automatic retry, operator requeue, pause/resume, SQLite-persisted
• Multi-tenant — isolated vaults via IVaultFactory, with single-call vector purge per tenant
• Extraction diagnostics — a legitimate zero-chunk result (scanned PDF, blank page) says so
• Damage-aware records — records are swapped in atomically, and an unreadable one is reported rather than dropped from listings
• Image enrichment — plug in a vision model and extracted images become indexed content
• Hybrid-ready — chunks are written to the keyword index alongside the vector store when one is registered

Installation

dotnet add package FluxFeed

Requirements

• .NET 10, and a registered FluxIndex vector store + IEmbeddingService (see Quick Start). The package pulls in FluxIndex.Core and FileFlux (the source of the extraction diagnostics described below) at the versions it was built against.
• git 2.x on PATH. Vault history (DiffAsync, LogAsync, GetContentAtCommitAsync) runs the git CLI. If git is installed elsewhere, set FileVaultOptions.GitExecutablePath; if you deliberately want a history-less vault, set FileVaultOptions.AllowMissingGit = true. Without either, the first vault operation fails with a message that says exactly this instead of silently creating a vault with no history.
• A Generic Host (Microsoft.Extensions.Hosting) for background processing — the queue worker is an IHostedService. Console apps without a host set EnableBackgroundProcessing = false (see below).

Quick Start

using FluxFeed.Extensions;
using FluxFeed.Interfaces;
using FluxIndex.Core.Application.Interfaces;  // IEmbeddingService
using FluxIndex.Storage.SQLite;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

var builder = Host.CreateApplicationBuilder(args);

// 1. FluxIndex side — a vector store and an embedding service must be registered.
IEmbeddingService embedder = new MyEmbeddingService();   // bring your own, or a FluxIndex.Providers.* service
builder.Services.AddSingleton<IEmbeddingService>(embedder);
builder.Services.AddSQLiteVecVectorStore(o =>
{
    o.DatabasePath = "fluxindex.db";
    o.VectorDimension = embedder.GetEmbeddingDimension();  // the store's dimension is the embedder's
});

// 2. FluxFeed side — vault + FileFlux extraction/chunking + FluxIndex indexing.
builder.Services.AddFileVaultWithFluxIndex(o => o.VaultBasePath = "./data/.vault");

using var host = builder.Build();
await host.StartAsync();   // starts the background queue worker

using (var scope = host.Services.CreateScope())
{
    var vault = scope.ServiceProvider.GetRequiredService<IVault>();

    // Index a file and wait for the pipeline to finish.
    var entry = await vault.MemorizeAsync("./docs/handbook.pdf", waitForCompletion: true);
    Console.WriteLine($"{entry.Stage} · {entry.ChunkCount} chunks");

    // Watch a folder; new and changed files are queued automatically.
    await vault.AddWatchedFolderAsync("./docs", autoMemorize: true);

    // Search, optionally scoped to a path.
    var results = await vault.SearchAsync("vacation policy", VaultSearchOptions.ForFolder("./docs"));
}

await host.StopAsync();

FluxFeed binds the vector store to the embedder's identity for you — there is no BindIdentity call to make, and a store already bound to a different embedder fails at startup with EmbeddingModelMismatchException rather than mixing vectors.

IVault is scoped — resolve it from a scope, or inject it into a scoped service, rather than from the root provider.

Without a host

The background worker is an IHostedService; a plain ServiceCollection never starts it. Either start the hosted services yourself, or process inline:

var services = new ServiceCollection();
services.AddSingleton<IEmbeddingService>(embedder);
services.AddSQLiteVecVectorStore(o => { o.DatabasePath = "fluxindex.db"; o.VectorDimension = embedder.GetEmbeddingDimension(); });
services.AddFileVaultWithFluxIndex(o =>
{
    o.VaultBasePath = "./data/.vault";
    o.EnableBackgroundProcessing = false;   // MemorizeAsync/RefreshAsync run inline and return the terminal entry
});

using var provider = services.BuildServiceProvider();
using var scope = provider.CreateScope();
var vault = scope.ServiceProvider.GetRequiredService<IVault>();
var entry = await vault.MemorizeAsync("./docs/handbook.pdf", waitForCompletion: true);

If background processing is left on without a host, MemorizeAsync(..., waitForCompletion: true) does not hang: after WorkerStartupTimeout (5 s) it throws an InvalidOperationException that names both fixes above.

Registration entry points

FileFlux services are registered only if you have not registered them yourself, so a prior AddFileFlux(ServiceLifetime.Singleton) keeps its lifetime.

How it works

An entry moves through four stages, or lands on Error:

Source → Extracted → Refined → Memorized

Refined is skipped when there is nothing to refine, and Stale marks an entry whose vectors went missing — the integrity check sets it, and re-memorizing restores search.

MemorizeAsync runs the whole pipeline (re-extracting the source). RefreshAsync re-indexes the refined content without re-extracting — use it after hand-editing append-text.md or qa.md. With background processing on, both only enqueue a job and return the entry as it was; pass waitForCompletion: true (MemorizeAsync(path, waitForCompletion: true), RefreshAsync(path, waitForCompletion: true)) to get the re-indexed entry — and its new commit — back.

Each entry lives under the vault base path, keyed by a hash of its absolute file path:

.vault/{filepath-hash}/
├── meta.json          (not git-tracked)
├── images/            (not git-tracked)
│   └── manifest.json
└── vault/             (git-tracked)
    ├── refined.md     extracted + refined content
    ├── append-text.md your additions  (create it yourself, then RefreshAsync)
    └── qa.md          your Q&A         (same)

Memorize writes refined.md only; append-text.md and qa.md are yours to create next to it. Their content is indexed together with refined.md on the next RefreshAsync, which also commits them.

Because vault/ is a git repository, DiffAsync and LogAsync report exactly what changed and when. DiffAsync compares the working tree against HEAD — since Memorize/RefreshAsync auto-commit after every successful update, the working tree is normally clean by the time a caller checks, so DiffAsync usually returns an empty string. Use DiffLastChangeAsync instead to see what the most recent commit actually changed (HEAD vs. its parent, or vs. the empty tree on the entry's first commit). GetContentAtCommitAsync(filePath, commitHash) reads the combined content (refined.md + append-text.md + qa.md) as it existed at a specific commit from that history — read-only, it does not touch the working tree. Returns null when the commit is unknown. To actually roll an entry back, write the returned content through the entry's normal update path (e.g. RefreshAsync after overwriting append-text.md/qa.md) — the library hands back the past content, the caller decides how to apply it.

Note what is not in that repository: the entry record, the raw extracted text and the images are work products sitting above it. They are outside the repository rather than ignored by it — which is why there is no ignore file at the entry level, where git would never read one.

Breaking in 0.8.0 — VaultEntry.GitignorePath and IVaultStorageService.CreateGitignoreAsync were removed for that reason. There is no replacement; the file they produced had no effect under any condition, so calls to them can simply be deleted.

File selection patterns

FileVaultOptions.DefaultIncludePatterns / DefaultExcludePatterns apply to discovery paths only:

Exclusion wins over inclusion, and an empty include list means "include everything".

Diagnostics

A document that yields zero chunks is not necessarily a failure, and a failure is not necessarily described by its most recent error. Both cases are reported explicitly rather than left to inference.

var entry = await vault.MemorizeAsync(path, waitForCompletion: true);

if (entry.ExtractionHints?.TryGetValue("extraction_failure_reason", out var reason) == true)
{
    // "no_text_layer" (image-only/scanned) | "blank_page" (empty document)
    // entry.ExtractionWarnings carries the human-readable explanation.
}

• Extraction hints are an opaque pass-through. Keys and values are the extractor's own vocabulary (FileFlux RawContent.Hints / Warnings); FluxFeed persists them to meta.json without interpreting them. Only scalar values are stored, so new hints flow through without drift. They always describe the latest extraction, and are cleared when one reports none.
• FirstError vs LastError. FirstError is the failure that started the current episode and usually carries the extractor's diagnosis; LastError is the most recent one. A retry failing for its own reason overwrites the latter but not the former. Both clear on a successful stage or reset, and neither is cleared by sync-status transitions — so use Stage/SyncStatus, not FirstError != null, to decide whether an entry is currently broken.

Queue fairness

One IVaultQueueService shared across vaults is the default registration, and until 0.23.0 nothing stopped one owner's backlog from occupying the whole worker pool: everyone else waited behind it in arrival order, however small their work was.

Jobs carry an optional group key — QueueGroupKey, defaulting to the vault's own VaultId, so a multi-tenant setup built on VaultFactory gets this without wiring anything. MaxInFlightPerGroup (default 1) is how many of one group's jobs may run at once.

The cap is work-conserving: it binds only while another group has work queued. A vault alone on the queue still uses the full MaxConcurrentProcessing, so nothing is paid for having fairness on — and the moment a second vault enqueues, the first is held to its share and the newcomer is dequeued ahead of the backlog that arrived before it.

o.MaxInFlightPerGroup = 2;      // each owner may hold two slots while others wait
o.MaxInFlightPerGroup = 0;      // off: arrival order and priority only
o.QueueGroupKey = "team-a";     // group several vaults together

What it deliberately does not do: no preemption (a job already running is never interrupted, so one long document still holds its slot for as long as it takes — fairness is about the other slots), and no per-group weights. Ungrouped jobs are never capped, and priority still orders whatever is eligible.

• Deterministic failures are not retried (since 0.23.0). Whether a failure can succeed on a later attempt is decided from the exception type behind it, not from the message: a missing file, an extension no reader handles, or a corrupt archive fails identically every time, and each attempt holds the queue head for as long as the first did. A failure the library does not recognise stays retryable, so narrowing this cannot silently drop a recoverable job. MemorizeResult.FailureKind exposes the judgment (Permanent / Transient / Unknown) if you want to act on it yourself.

• A failed memorize is reported as failed (fixed in 0.23.0). MemorizeAsync/RefreshAsync signal failure by returning MemorizeResult.Failed(...), not by throwing. Before 0.23.0 the queue worker discarded that result and marked the job completed, so a document that failed to index still raised completedCount and never appeared in failedCount. If you built a workaround that re-checks entries the queue claims are done, it is no longer needed. The same release makes the inline paths agree: with EnableBackgroundProcessing = false the call is terminal, so a failed MemorizeAsync/RefreshAsync now throws on every overload rather than only on waitForCompletion: true. The queued path is unchanged — that failure belongs to the worker.

• Auto-retry actually runs now (fixed in 0.23.0). CanRetry requires a job in Failed state, but the worker held a snapshot left in Processing by dequeue, so the condition was never true and EnableAutoRetry / MaxRetryCount / RetryDelayMs did nothing on that path. If your deployment appeared to never retry, this is why.

• An operator can put a failed job back in the queue (since 0.24.0). RetryAsync enforces the automatic retry budget, which is right for the worker deciding whether to keep going unattended and wrong for a person: the jobs someone reaches for a retry button over are precisely the ones that have used the budget up, so that call succeeded only when it was not needed. RequeueAsync is the operator's path — it clears RetryCount and re-queues, and it throws rather than returning a bool, because the ways it can decline call for different answers. VaultJobNotFoundException means the list is stale; VaultJobNotRetryableException carries a VaultRetryRefusal of NotFailed (someone already dealt with it) or PermanentFailure (running it again would fail identically). Those map onto 404 / 409 / 409 directly.

  The PermanentFailure case is new information, not a new restriction. A permanent failure never spends retry budget — the worker stops before the auto-retry branch — so before 0.24.0 the ordinary retry path saw a failed job with attempts to spare and re-queued a password-protected document quite happily, and the operator saw a button that appeared to work. The classification is now written to the job row so the question can still be answered later; a job that failed before the column existed reads as unclassified, and unclassified is allowed through.

• Refresh has a precondition. It needs refined content to exist, which ProcessingStage does not imply — a memorize with nothing to index skips the refine step, so a Memorized entry legitimately may have none. RefreshAsync rejects those, and DetectChangesAsync recommends Memorize instead, since re-extraction lets a failed or empty entry recover on its own.

Damaged records

An entry record (meta.json) is written to a scratch file and swapped into place. Concurrent writers therefore only decide which record wins — they never interleave — and an interrupted write never leaves half a record behind.

An unreadable record is distinguished from an absent one:

// absent → null; present but unreadable → VaultRecordUnreadableException
var entry = VaultEntry.LoadByHash(hash, vaultBasePath);

// entry directories missing from the listing, exposed so they can be repaired
IReadOnlyList<string> damaged = await vault.ListUnreadableAsync();

• ListAsync() skips unreadable records but logs a warning. To display those entries or offer to repair them, use the paths returned by ListUnreadableAsync().
• The two listings split on the same signal, so an entry appears in exactly one of them and their counts sum to the total. Any other IO error propagates rather than being swallowed — a listing that quietly gets shorter is the failure this reporting exists to eliminate.
• The swap sets the outgoing record aside under a scratch name. When the platform cannot clear that scratch file — likely enough while a reader holds the record open — it stays in the entry directory and the next write removes it, but only once it is old enough that no in-flight swap still needs it to roll back to. One may briefly be visible right after a concurrent write; they do not accumulate.
• Paths that rewrite the record anyway (memorize, refresh) report an unreadable record and then recreate it rather than failing, which would strand the entry permanently. A recreated record starts with no history.

Optional integrations

Each of these is enabled by registering a service. Register nothing and the pipeline behaves as if the feature did not exist.

Image enrichment — IVaultImageEnricher

Images extracted from documents are always stored. Register a describer and those descriptions get indexed too, which is what makes scanned or diagram-only documents searchable at all.

public sealed class VisionEnricher : IVaultImageEnricher
{
    public async Task<string?> DescribeAsync(VaultImageDescriptionRequest request, CancellationToken ct)
        => await _vision.CaptionAsync(request.Image.FilePath, request.DocumentText, ct);
        // returning null means "not this time" — the pipeline retries that image on the next run
}

services.AddSingleton<IVaultImageEnricher, VisionEnricher>();

Descriptions are persisted per image, so re-memorizing does not re-describe images that already succeeded, and one image's failure aborts neither the others nor the memorize. Each description is indexed as its own chunk tagged chunk_kind="image_description" with image_id / image_file metadata — no markers are injected into the document text.

Descriptions go through the same chunker as the body, so MaxChunkSize bounds them too and a long description becomes several chunks that each carry the same image_id / image_file. Returning a long description is therefore safe: it cannot push the document's embedding request past the model's context window.

An image that keeps failing (unsupported format, corrupt data — a null return or a thrown exception, either counts) is not retried forever. Once it has failed FileVaultOptions.MaxImageEnrichmentAttempts times (default 3), it is marked permanently failed and the pipeline stops offering it to the enricher — this survives a process restart, since the attempt count is persisted in the image manifest, not held in memory. A later success (e.g. after you fix the enricher) clears the failure record for that image.

Read back what the enricher wrote — or why an image is still pending — with GetImageManifestAsync(filePath). Each VaultImage carries its Description when one has been persisted, or LastEnrichmentFailure (reason, attempt count, whether it is now permanent) while none has. Returns an empty list when the entry doesn't exist or has no images.

Keyword index — IKeywordSearchService

When one is registered, every chunk written to the vector store is written to the keyword index as well, and RemoveAsync deletes from both. Without it the keyword index stays empty and hybrid search degenerates to vector-only. Check IVaultPipeline.SupportsKeywordIndex to confirm the wiring — it is on the interface, so holding the pipeline as IVaultPipeline is enough (SupportsGraphRAG reports the GraphRAG leg the same way).

Hybrid search — store-native, or IHybridSearchService

VaultSearchOptions.SearchStrategy = VaultSearchStrategy.Hybrid is honored when either the vector store fuses natively (INativeHybridSearch — FluxIndex.Storage.SQLite's sqlite-vec store does, over the FTS5 rows it writes itself at ingestion; preferred, no second index) or an IHybridSearchService is registered. A PathScope is pushed into the native path as a filter and applied to both legs before fusion, so a scoped request gets the fused ranking of the in-scope chunks (FluxIndex.Core 0.32.0+). Otherwise the query runs as vector search and says so via VaultSearchResult.ExecutedStrategy — compare it against RequestedStrategy rather than assuming the request was honored.

Keyword-only search — VaultSearchStrategy.Keyword

VaultSearchOptions.SearchStrategy = VaultSearchStrategy.Keyword runs pure BM25 through the same IKeywordSearchService the keyword index above writes to — no query embedding, no vector search. It degrades to vector the same way Hybrid does when no IKeywordSearchService is registered (reported via ExecutedStrategy). Use this over Hybrid with a zero vector weight when you actually want keyword-only: a weighted hybrid request still generates a query embedding and runs a vector search it then discards.

RAG security — FluxGuard.Remote.RAG.IRAGSecurityPipeline

When one is registered, MemorizeAsync/RefreshAsync validate every chunk through it before indexing — a chunk the pipeline suggests blocking (RAG poisoning / indirect prompt injection) is dropped from the batch entirely, one it suggests sanitizing has its content replaced. Off by default; nothing changes without it.

using FluxGuard.Remote.RAG;

var pipeline = new VaultPipeline(
    git, hasher, storage, logger,
    vectorStore: vectorStore,
    embeddingService: embeddingService,
    ragSecurityPipeline: new IndirectInjectionDetector());

Contextual enrichment — FluxIndex.Core.Application.Interfaces.IContextualEnrichmentService

Opt-in. Before a document's text chunks are embedded and keyword-indexed, each one gets a short LLM-written context — where it sits in its document — prepended (Anthropic's "contextual retrieval"). The same enriched text is what gets stored, embedded and keyword-indexed, so retrieval and display agree; the context alone is also kept in chunk metadata (context_summary) and the step is recorded as enrichment=contextual. Image-description chunks are not enriched. Refresh re-runs it, since it happens at the chunk stage.

The port is FluxIndex.Core's own IContextualEnrichmentService (GenerateContextBatchAsync(chunks, fullDocumentText) → one context per chunk, in order). FluxFeed does not depend on any particular LLM library for it; the FluxImprover-backed implementation ships in FluxIndex.Integrations.FluxImprover, or implement the two methods yourself.

Two things are required — registering the port alone does nothing, so a container that already has an enrichment service for other reasons never pays one LLM call per chunk by accident:

services.AddScoped<IContextualEnrichmentService, MyContextualEnrichment>();   // FluxIndex.Core port
services.AddFileVaultWithFluxIndex(options =>
{
    options.ContextualEnrichment.Enabled = true;          // default false
    options.ContextualEnrichment.ContinueOnError = true;  // default: warn + index plain chunks tagged enrichment=failed
});

Cost is whatever the port spends — typically one generation call per chunk with the whole document in the prompt, so budget it per document size. With ContinueOnError = false an enrichment failure fails the memorize instead of degrading; a port that returns the wrong number of contexts is always a failure (never a silent misalignment).

Multi-tenant

AddFileVaultFactoryWithFluxIndex swaps the single IVault for an IVaultFactory. Each tenant gets its own .vault/ directory, processing queue and queue worker (started by the factory, stopped when the tenant is disposed - no host involvement), while stateless services and the vector store are shared.

services.AddFileVaultFactoryWithFluxIndex(o => o.VaultBasePath = "./data");

var vault = factory.GetOrCreate("tenant-a");
await vault.MemorizeAsync(path);

// Deleting a tenant: one filtered delete per backend removes all of its chunks, no per-entry loop.
await factory.DisposeAsync("tenant-a", purgeVectors: true);

Chunks are tagged with a vault_id metadata field, which is what makes the bulk purge (IVault.PurgeAsync) possible.

Changed in 0.10.0 — the purge now clears the keyword index too. Before this it removed the vectors, logged a warning, and returned success while the tenant's text stayed searchable through keyword and hybrid search. Requires FluxIndex.Core 0.25.0+, which added the tag-scoped bulk delete this needs. The returned count is the number of chunks, not a sum across backends.

Configuration

FileVaultOptions (bindable from the FileVault configuration section):

The background worker (VaultBackgroundService) holds a lease from IVaultQueueService.RegisterWorker() while it consumes the queue; that lease is how WaitForJobAsync tells "a worker is busy" from "nobody will ever process this job". A custom IVaultQueueService implementation should return a real lease from RegisterWorker() (the interface default is a no-op lease, which disables the check).

Same-file work is serialized for you

A vault's git repository lives per entry (VaultEntry.VaultPath = <EntryPath>/vault), not per FileVault. Two jobs for two files therefore commit into two different repositories and are safe to run together — that is the parallelism MaxConcurrentProcessing buys. Two jobs for one file are not: they would write one working tree and race one index.lock.

The queue handles this itself, so a consumer does not need a per-file lock of its own:

• DequeueAsync skips any entry that already has a job in flight, and hands that job out as soon as the running one finishes.
• Enqueuing work that is already queued for the same file and type merges into the waiting job rather than adding a second row — the caller awaits the job that already exists. A more urgent request raises that job's priority instead of being demoted into it. A job already processing is not merged into: it read the file as it was, so a later request needs its own run.

Priority

MemorizeAsync / RefreshAsync / SyncAsync take an optional VaultJobPriority. The queue has always ordered by priority; until 0.22.0 nothing on IVault could set it, so a bulk crawl and a user waiting on one file competed purely on arrival order.

await vault.SyncAsync(VaultJobPriority.Low, ct);                              // background crawl, yields
await vault.MemorizeAsync(path, VaultJobPriority.High, waitForCompletion: true, ct);   // user is waiting

Observing the queue

GetStatisticsAsync() answers two different questions, and conflating them is a reported source of false alarms:

LastSucceededAt was called LastProcessedAt before 0.22.0 while only ever reflecting successes — a worker that was running steadily and failing every job left it frozen and read as stopped. Note that ProcessingCount is a point-in-time count and is legitimately 0 between jobs; it is the pair (ProcessingCount + a fresh LastAttemptedAt) that says "between jobs" rather than "stopped".

GetJobsAsync orders and pages in SQL:

// the latest 50 failures, not the oldest 50
var recent = await queue.GetJobsAsync(VaultJobStatus.Failed, limit: 50, newestFirst: true, ct: ct);
// second page
var next = await queue.GetJobsAsync(VaultJobStatus.Failed, limit: 50, offset: 50, newestFirst: true, ct: ct);

newestFirst sorts by queued_at alone — priority decides what runs next, not what is most recent. The default order is unchanged (priority, then oldest first).

License

MIT — see LICENSE.