Resilience & Reliability

async def ingest_message(msg: NormalizedMessage):
    # Stage 1: Preprocess (required)
    preprocessed = await preprocessor.process(msg)
    
    # Stage 2a: Extract facts (required — queue to DLQ on failure)
    try:
        facts = await extractor.extract(preprocessed)
    except LLMUnavailableError:
        await dead_letter_queue.enqueue(msg)
        return
    
    # Stage 2b: Entity extraction (optional — skip if Neo4j/LLM down)
    entities = []
    if await health.check("neo4j") and await health.check("gemini"):
        try:
            entities = await entity_extractor.extract(preprocessed, facts)
        except Exception as e:
            logger.warning(f"Entity extraction failed, continuing: {e}")
            await backfill_queue.enqueue("entities", msg.id, preprocessed)
    
    # Stage 3: Embed (optional — queue if Jina down)
    embeddings = None
    if await health.check("jina"):
        embeddings = await embedder.embed(facts)
    else:
        await backfill_queue.enqueue("embeddings", msg.id, facts)
    
    # Stage 4: Persist via outbox pattern
    await persister.persist(facts, entities, embeddings)

# Atomic write to MongoDB
intent = WriteIntent(
    id=deterministic_uuid(facts),
    facts=facts,
    entities=entities,
    embeddings=embeddings,
    status={
        "weaviate": "pending",
        "neo4j": "pending" if entities else "skipped",
        "state": "pending"
    },
    retry_count=0
)
await mongo.write_intents.insert_one(intent.dict())

# Weaviate — idempotent via deterministic UUID
if intent.status["weaviate"] == "pending":
    try:
        await weaviate.upsert(intent.facts, intent.embeddings)
        await mark(intent.id, "weaviate", "done")
    except Exception:
        await mark(intent.id, "weaviate", "failed")

# Neo4j — idempotent via MERGE semantics
if intent.status["neo4j"] == "pending":
    try:
        for entity in intent.entities:
            await neo4j.upsert_entity(entity)
        await mark(intent.id, "neo4j", "done")
    except Exception:
        await mark(intent.id, "neo4j", "failed")

# MongoDB sync state — final step
await update_sync_state(intent)
await mark(intent.id, "state", "done")

async def reconcile():
    stale = await mongo.write_intents.find({
        "$or": [
            {"status.weaviate": {"$in": ["pending", "failed"]}},
            {"status.neo4j": {"$in": ["pending", "failed"]}}
        ],
        "created_at": {"$lt": now() - timedelta(minutes=5)},
        "retry_count": {"$lt": 5}
    }).to_list()
    
    for intent in stale:
        await fan_out(WriteIntent(**intent))
        await mongo.write_intents.update_one(
            {"id": intent["id"]},
            {"$inc": {"retry_count": 1}}
        )

try:
    graph_results = await graph_agent.traverse(entities)
except TimeoutError:
    logger.warning("Graph traversal timed out, using semantic-only")
    graph_results = []

if not await health.check("weaviate") and not await health.check("neo4j"):
    cached = await wiki_cache.get(channel_id)
    if cached:
        return cached
    return {"error": "System temporarily unavailable"}

try:
    external = await tavily.search(query)
except Exception:
    logger.warning("External search failed, using internal-only")
    external = None

# Retry failed writes
POST /api/admin/reconcile

# Rebuild wiki for channel
POST /api/admin/wiki/rebuild
{ "channel_id": "C123456" }

# Refresh ACL for user
POST /api/admin/acl/refresh
{ "user_id": "U123456" }

# Trigger backfill
POST /api/admin/backfill/:type
# types: entities, embeddings

GET /api/health
{
  "status": "degraded",
  "dependencies": {
    "weaviate": "healthy",
    "neo4j": "failing",
    "mongodb": "healthy",
    "gemini": "healthy",
    "jina": "degraded",
    "tavily": "healthy"
  },
  "uptime": 99.5,
  "degraded_since": "2026-04-13T12:00:00Z"
}