Brain Query Interface — Temporal Weighting Spec

Q: "Is Friday a half-day for the kids?"
A: "Yes, all Fridays are half-days per the 2024-2025 school calendar"
→ Wrong. The 2025-2026 calendar changed this. The correct answer was in 
   yesterday's principal email.

def temporal_score(document, query_pattern):
    """
    Calculate temporal relevance score for a document.
    Returns: float [0.0, 1.0]
    """

    # Document metadata
    doc_date = document.ingestion_date  # When Icarus processed it
    source_date = document.source_date   # When the original was created (email sent, receipt dated)
    doc_type = document.type            # email | invoice | receipt | manual | newsletter | calendar_event

    # Query pattern parameters
    pattern = query_pattern.type        # logistical | entity | historical
    decay_model = query_pattern.decay   # exponential | linear | none
    half_life_days = query_pattern.half_life

    # Use the more recent of ingestion or source date
    effective_date = max(doc_date, source_date)
    age_days = (now - effective_date).days

    # Decay function
    if decay_model == "exponential":
        recency_multiplier = 2 ** (-age_days / half_life_days)
    elif decay_model == "linear":
        recency_multiplier = max(0, 1 - (age_days / (half_life_days * 2)))
    else:  # none
        recency_multiplier = 1.0

    # Source type weight
    source_weights = {
        "calendar_event": 1.0,
        "email": 0.9,
        "invoice": 0.95,
        "receipt": 0.95,
        "newsletter": 0.7,
        "manual": 0.8,
        "static_pdf": 0.5,
        "photo": 0.6
    }
    type_weight = source_weights.get(doc_type, 0.5)

    # Combined score
    score = recency_multiplier * type_weight

    # Logistical queries: hard recency cutoff
    if pattern == "logistical" and age_days > 30:
        score *= 0.1  # Drastically reduce, don't zero (emergency fallback)

    return score

query_patterns:
  logistical:
    decay: exponential
    half_life_days: 7
    recency_cutoff: 30
    source_priority: [calendar_event, email, newsletter, static_pdf]

  entity:
    decay: linear
    half_life_days: 90
    recency_cutoff: null  # No hard cutoff
    source_priority: [invoice, receipt, manual, email, static_pdf]

  historical:
    decay: none
    half_life_days: null
    recency_cutoff: null
    source_priority: [invoice, receipt, email, photo, static_pdf]

collection.add(
    documents=[chunk.text],
    embeddings=[chunk.embedding],
    metadatas=[{
        "doc_id": doc.id,
        "source_date": doc.source_date.isoformat(),
        "ingestion_date": doc.ingestion_date.isoformat(),
        "doc_type": doc.type,
        "temporal_score": calculate_score(doc, pattern)
    }],
    ids=[chunk.id]
)

# Pre-filter by recency, then semantic search
results = collection.query(
    query_embeddings=[query_embedding],
    where={
        "$or": [
            {"source_date": {"$gte": fourteen_days_ago}},
            {"doc_type": {"$in": ["calendar_event", "manual"]}}  # Timeless docs
        ]
    },
    n_results=10
)
# Re-rank by temporal_score metadata

def answer_confidence(results, pattern):
    """
    Determine if we have enough confidence to answer.
    """
    if not results:
        return 0.0, "No relevant documents found"

    top_score = results[0].temporal_score

    if pattern == "logistical":
        if top_score < 0.3:
            return 0.2, "Stale information — verify with school/organizer"
        if top_score < 0.6:
            return 0.5, "Possibly outdated — check for newer updates"
        return 0.9, "Recent information found"

    elif pattern == "entity":
        if top_score < 0.4:
            return 0.4, "Entity found in old document — verify if still current"
        return 0.85, "Entity found with source document"

    else:  # historical
        return 0.8, "Historical record found"