src/ranking.py

from __future__ import annotations

from src.schemas import ExternalAgentResult, RankedRecommendation, UserMemory


def rank_results(result: ExternalAgentResult, memory: UserMemory) -> tuple[list[RankedRecommendation], str]:
    accepted = memory.accepted()
    recommendations: list[RankedRecommendation] = []
    for source in result.sources:
        haystack = " ".join([source.title, source.snippet, source.extracted_text or ""]).lower()
        matched = [fact.text for fact in accepted if any(token in haystack for token in _important_tokens(fact.text))]
        score = 0.5 + min(0.4, 0.1 * len(matched))
        recommendations.append(
            RankedRecommendation(
                title=source.title,
                url=source.url,
                score=round(score, 2),
                matched_memory=matched,
                conflicts=[],
                why_it_fits="Matches accepted local preferences after external retrieval." if matched else "Relevant to the sanitized task.",
                risks="Fixture or search snippets may be incomplete; verify important details.",
                next_step="Open the source or turn this into a concrete local action.",
            )
        )
    if not recommendations:
        recommendations.append(
            RankedRecommendation(
                title="Local answer",
                url=None,
                score=0.5,
                matched_memory=[],
                conflicts=[],
                why_it_fits="No external sources were needed.",
                risks="Local-only answer may miss current web changes.",
                next_step="Ask a web lookup if current information is required.",
            )
        )
    reasoning = f"Private ranking used {len(accepted)} accepted memory facts after the external result returned."
    return recommendations, reasoning


def _important_tokens(text: str) -> list[str]:
    return [word.lower() for word in text.split() if len(word) >= 5][:8]