fix: resolve 4 code-critical issues from Fase 2 audit

V2.07 (A01): Replace subprocess.run(['arq', ...]) with programmatic
  arq.worker.run_worker(WorkerSettings). Fixes signal handling,
  graceful shutdown, and logging unification.

V2.10 (D11): Fix filename → file_path in backfill_decay_metadata.py
  resolve_timestamp() now reads payload.file_path (consistent with
  what bulk_wiki_ingest.py writes). Replaced VAULT_ROOT path
  reconstruction with direct Path(file_path).

V2.11 (D05): Auto-create Qdrant collection in bulk_wiki_ingest.py
  If collection doesn't exist, creates it with hybrid schema
  (dense 4096d Cosine + sparse BM25) instead of sys.exit(1).

V2.12 (D27): Add sparse vector generation to bulk_wiki_ingest.py
  Optional fastembed BM25 sparse vectors alongside dense embeddings.
  Falls back to dense-only if fastembed not installed. Vectors stored
  as Qdrant named vectors: {'dense': ..., 'sparse': ...}

docker/worker/main.py

@@ -103,11 +103,11 @@ if __name__ == "__main__":
     import sys
 
     if len(sys.argv) > 1 and sys.argv[1] == "--run-worker":
-        # Start ARQ worker (concurrency via max_jobs, not multi-process)
-        import subprocess
-        logger.info("Starting ARQ worker...")
-        subprocess.run(["arq", "main.WorkerSettings"])
+        from arq.worker import run_worker
+        logger.info("Starting ARQ worker (programmatic)...")
+        run_worker(WorkerSettings)
     else:
         print("Usage: python main.py --run-worker")
         print("")
-        print("To enqueue jobs, use the enqueue_host.py script")
+        print("To enqueue jobs via Redis, use the enqueue functions")
+        print("or the Hermes cron scripts in the scripts/ directory.")

scripts/backfill_decay_metadata.py

@@ -107,7 +107,7 @@ def resolve_timestamp(point: dict) -> str:
     """Return ISO 8601 string for created_at/last_accessed_at.
 
     Session points: use payload.timestamp (Unix epoch float).
-    Wiki points: locate file via payload.filename → file mtime.
+    Wiki points: locate file via payload.file_path → file mtime.
     Fallback: now().
     """
     pl = point.get("payload", {})
@@ -120,13 +120,11 @@ def resolve_timestamp(point: dict) -> str:
         except (ValueError, TypeError, OSError):
             pass
 
-    # Wiki point — resolve filename to filesystem path
-    filename = pl.get("filename", "")
+    # Wiki point — resolve file_path to filesystem
+    file_path = pl.get("file_path", "")
     source = pl.get("source", "")
-    if filename and source.startswith("wiki-"):
-        # Map source to subfolder: wiki-concepts → concepts, wiki-entities → entities, etc.
-        subfolder = source.replace("wiki-", "")
-        candidate = VAULT_ROOT / "wiki" / subfolder / f"{filename}.md"
+    if file_path and source.startswith("wiki-"):
+        candidate = Path(file_path)
         try:
             mtime = candidate.stat().st_mtime
             return datetime.fromtimestamp(mtime, tz=timezone.utc).isoformat()

scripts/bulk_wiki_ingest.py

@@ -16,6 +16,15 @@ from collections import Counter
 import aiohttp
 import asyncio
 
+# Sparse embedding (BM25) — optional, falls back to dense-only
+try:
+    from fastembed import SparseTextEmbedding
+    _sparse_model = SparseTextEmbedding(model_name="Qdrant/bm25")
+    _has_sparse = True
+except ImportError:
+    _sparse_model = None
+    _has_sparse = False
+
 # ─── Config ────────────────────────────────────────────────────────────────
 OPENROUTER_KEY = os.environ.get("OPENROUTER_API_KEY")
 QDRANT_URL = "http://localhost:6333"
@@ -70,7 +79,22 @@ def get_tags_from_frontmatter(meta: dict) -> list[str]:
     return tags if isinstance(tags, list) else []
 
 async def get_embedding(session: aiohttp.ClientSession, text: str) -> list[float] | None:
-    """Gera embedding via OpenRouter."""
+    """Gera embedding denso via OpenRouter."""
+
+
+def get_sparse_vector(text: str) -> dict | None:
+    """Gera sparse vector BM25 via fastembed (se disponível)."""
+    if not _has_sparse:
+        return None
+    try:
+        sparse_result = list(_sparse_model.embed([text]))[0]
+        return {
+            "indices": sparse_result.indices.tolist(),
+            "values": sparse_result.values.tolist(),
+        }
+    except Exception as e:
+        print(f"⚠️ Sparse embedding error: {e}")
+        return None
     payload = {
         "model": EMBEDDING_MODEL,
         "input": text[:MAX_TEXT_LEN],
@@ -123,11 +147,32 @@ async def main():
 
     connector = aiohttp.TCPConnector(limit=20)
     async with aiohttp.ClientSession(connector=connector) as session:
-        # Verificar coleção
+        # Verificar coleção — criar se não existir
         async with session.get(f"{QDRANT_URL}/collections/{COLLECTION}") as r:
             if r.status != 200:
-                print(f"❌ Coleção {COLLECTION} não existe!")
-                sys.exit(1)
+                print(f"⚠️  Coleção {COLLECTION} não existe. Criando...")
+                collection_config = {
+                    "vectors": {
+                        "dense": {
+                            "size": EMBEDDING_DIMS,
+                            "distance": "Cosine",
+                        }
+                    },
+                    "sparse_vectors": {
+                        "sparse": {},
+                    },
+                }
+                async with session.put(
+                    f"{QDRANT_URL}/collections/{COLLECTION}",
+                    headers={"Content-Type": "application/json"},
+                    json=collection_config,
+                    timeout=aiohttp.ClientTimeout(total=10),
+                ) as cr:
+                    if cr.status not in (200, 201):
+                        body = await cr.text()
+                        print(f"❌ Falha ao criar coleção ({cr.status}): {body[:200]}")
+                        sys.exit(1)
+                print(f"✅ Coleção {COLLECTION} criada (dense {EMBEDDING_DIMS}d + sparse BM25)")
 
         print("\n🚀 Iniciando ingestão em batches...\n")
 
@@ -167,14 +212,21 @@ async def main():
                 embed_tasks = [get_embedding(session, b["embed_text"]) for b in batch]
                 vectors = await asyncio.gather(*embed_tasks)
 
+                # Gerar sparse vectors (sync, fastembed é CPU-bound)
+                sparse_vecs = [get_sparse_vector(b["embed_text"]) for b in batch]
+
                 # Preparar pontos Qdrant
                 points = []
-                for b, vec in zip(batch, vectors):
+                for b, vec, sparse in zip(batch, vectors, sparse_vecs):
                     if vec is None:
                         stats["fail"] += 1
                         errors.append(f"Embedding failed: {b['path']}")
                         continue
 
+                    vector_payload = {"dense": vec}
+                    if sparse is not None:
+                        vector_payload["sparse"] = sparse
+
                     # Heurística de importance_score baseada no path/nome
                     importance_score = 0.5
                     path_str_lower = b["path"].lower()
@@ -189,7 +241,7 @@ async def main():
                     
                     point = {
                         "id": str(uuid.uuid4()),
-                        "vector": {"dense": vec},
+                        "vector": vector_payload,
                         "payload": {
                             "text": b["embed_text"],
                             "source": b["source"],