graph_backfill processes messages strictly sequentially despite store's per-entity locking making concurrent extraction safe

[bug-ops]

Finding

AgentAccess::graph_backfill extracts entities/edges from each unprocessed message in a batch strictly sequentially — one extract_and_store(...).await (an LLM call plus SQLite/Qdrant writes) per message, one at a time. For a backfill over a large unprocessed-message backlog, wall time is messages * (LLM extraction latency) with zero concurrency.

This looks like a missed opportunity rather than a required serialization:

• EntityResolver (crates/zeph-memory/src/graph/resolver/mod.rs) already provides per-entity-name locking (lock_name returns an OwnedMutexGuard) and upsert-keyed entity creation (upsert_entity, keyed on canonical_name + entity_type), which is exactly the mechanism needed to make concurrent extraction across different messages safe.
• The function immediately above graph_backfill in the same file, semantic_scan_plugin_add (lines 141-224), already establishes the pattern for this exact kind of batched-LLM-call workload: futures::stream::iter(...).buffer_unordered(4) with an aggregate timeout — but graph_backfill does not reuse it.

Location

crates/zeph-core/src/agent/agent_access_impl.rs:602-660 (the for (_id, content) in &messages loop inside graph_backfill)

Compare with the existing bounded-concurrency pattern at crates/zeph-core/src/agent/agent_access_impl.rs:173-193.

Before

for (_id, content) in &messages {
    if content.trim().is_empty() {
        continue;
    }
    let extraction_cfg = GraphExtractionConfig { /* ... */ };
    let pool = store.pool().clone();
    match extract_and_store(
        content.clone(),
        vec![],
        provider.clone(),
        pool,
        extraction_cfg,
        None,
        None,
    )
    .await
    {
        Ok(result) => { /* accumulate stats */ }
        Err(e) => tracing::warn!("backfill extraction error: {e:#}"),
    }
}

After

use futures::stream::StreamExt as _;

let extractions = messages.iter().filter(|(_, content)| !content.trim().is_empty()).map(|(_, content)| {
    let extraction_cfg = build_extraction_cfg(&graph_cfg, embed_timeout_secs);
    let pool = store.pool().clone();
    let provider = provider.clone();
    let content = content.clone();
    async move {
        extract_and_store(content, vec![], provider, pool, extraction_cfg, None, None).await
    }
});
let results: Vec<_> = futures::stream::iter(extractions)
    .buffer_unordered(4) // same bound already used by semantic_scan_plugin_add
    .collect()
    .await;
for result in results {
    match result {
        Ok(r) => { /* accumulate stats */ }
        Err(e) => tracing::warn!("backfill extraction error: {e:#}"),
    }
}

Why

/graph backfill is an explicit, potentially long-running maintenance command over the full unprocessed-message backlog (batches of up to 50, looped until exhausted). Sequential per-message LLM calls make this needlessly slow with no correctness benefit, since the underlying store is already designed for concurrent-safe entity resolution (per-name mutex + upsert). Bounding concurrency at 4 (matching the codebase's existing convention) would cut backfill wall time roughly 4x without introducing new races.