"""Tests for live retrieval benchmark runners."""

from __future__ import annotations

import json
import subprocess
from collections.abc import Sequence
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from typer.testing import CliRunner

from zaxy.__main__ import app
from zaxy.cli import _parse_benchmark_baselines
from zaxy.embedding import HashEmbeddingProvider
from zaxy.event import EventLog
from zaxy.graph import GraphEventProjectionStatus
from zaxy.projection_backends import ProjectionBackendConfig
from zaxy.query import ContextChunk
from zaxy.retrieval_intent import classify_retrieval_intent
from zaxy.retrieval_plan import (
    absence_check_bundle,
    bridge_source_lane_queries,
    source_evidence_score,
    source_lane_candidate_limit,
    source_lane_queries,
    source_lane_query,
    source_synthesis_bundle,
    source_synthesis_candidate_limit,
)
from zaxy_benchmarks.benchmark import build_competitive_event_log, competitive_cases
from zaxy_benchmarks.live_benchmark import (
    CONSOLIDATION_WORKLOAD_VERSION,
    CONTEXT_COLLAPSE_WORKLOAD_VERSION,
    FROZEN_WORKLOAD_SUBJECTS,
    FROZEN_WORKLOAD_VERSION,
    GRAPH_TRAVERSAL_WORKLOAD_VERSION,
    LONGMEMEVAL_WORKLOAD_VERSION,
    SOURCE_RECALL_WORKLOAD_VERSION,
    SUITE_WORKLOAD_VERSION,
    TEMPORAL_RECALL_WORKLOAD_VERSION,
    BenchmarkCase,
    BenchmarkChunk,
    BenchmarkReport,
    BenchmarkSummary,
    BenchmarkWorkload,
    BM25Retriever,
    CachedEmbeddingProvider,
    CentroidConsolidationRetriever,
    EvidenceFrontierRetriever,
    MarkdownRetriever,
    MarkdownVectorRetriever,
    RankFusionRetriever,
    VectorRetriever,
    ZaxyCheckoutRetriever,
    ZaxyRetriever,
    _benchmark_contexts_from_checkout,
    _benchmark_projection_present,
    _build_source_lane_retriever,
    _mark_benchmark_projection,
    _reset_benchmark_graph,
    _scope_augmented_source_query,
    _scoped_fetch_limit,
    _scoped_source_fallback_limit,
    _scoped_source_fetch_limit,
    benchmark_case_scope_terms,
    benchmark_live_retrievers,
    benchmark_projection_cache_key,
    benchmark_retrievers,
    build_benchmark_suite_workload,
    build_consolidation_collapse_workload,
    build_context_collapse_workload,
    build_frozen_statistical_workload,
    build_graph_traversal_workload,
    build_live_zaxy_retriever,
    build_longmemeval_workload,
    build_mempalace_workload_inventory,
    build_source_recall_workload,
    build_temporal_recall_workload,
    compare_benchmark_reports,
    corpus_from_event_log,
    format_benchmark_comparison,
    format_mempalace_workload_inventory,
    report_to_markdown,
    summarize_miss_taxonomy,
    workload_fingerprint,
    write_benchmark_report,
)


def test_cli_exposes_live_benchmark_command() -> None:
    """The public CLI should expose a reproducible live benchmark command."""
    cli = "\n".join(
        path.read_text(encoding="utf-8") for path in sorted(Path("src/zaxy/cli").glob("*.py"))
    )
    script = Path("scripts/live-benchmark.sh").read_text(encoding="utf-8")

    assert "def benchmark(" in cli
    assert "def benchmark_compare(" in cli
    assert "compare_benchmark_reports" in cli
    assert "embedding_provider: str = typer.Option(" in cli
    assert "local-http" in cli
    assert "sentence-transformers" in cli
    assert "build_live_zaxy_retriever" in cli
    assert "build_statistical_event_log" in cli
    assert "build_frozen_statistical_workload" in cli
    assert "build_benchmark_suite_workload" in cli
    assert "build_consolidation_collapse_workload" in cli
    assert "build_context_collapse_workload" in cli
    assert "build_graph_traversal_workload" in cli
    assert "build_mempalace_workload_inventory" in cli
    assert "build_longmemeval_workload" in cli
    assert "--zaxy-backend" in cli
    assert "checkout_retriever=checkout_retriever" in cli
    assert "build_source_recall_workload" in cli
    assert "build_temporal_recall_workload" in cli
    assert "benchmark-inventory" in cli
    assert "lexical_retriever=_build_source_lane_retriever(corpus, provider)" in cli
    assert "--embedding-cache" in cli
    assert "--progress" in cli
    assert "--reuse-projection" in cli
    assert "--projection-backend" in cli
    assert 'projection_backend: str = typer.Option(\n        "embedded",' in cli
    assert "embedded default" in cli
    assert "--pggraph-dsn" in cli
    assert "Delete benchmark projection contents before ingestion" in cli
    assert "ProjectionBackendConfig(" in cli
    assert "neo4j_ca_cert=None" in cli
    assert "projection_backend_config=projection_backend_config" in cli
    assert "--baseline-backends" in cli
    assert "_build_benchmark_baselines" in cli
    assert "benchmark_projection_cache_key" in cli
    assert "--workload" in script
    assert "--dataset" in script
    assert "--subjects" in script
    assert "--documents" in script
    assert "--sessions" in script
    assert "--embedding-cache" in script
    assert "--progress" in script
    assert "--dry-run" in script
    assert 'EMBEDDING_PROVIDER="hash"' in script
    assert "Uses deterministic hash embeddings by default" in script
    assert 'PROJECTION_BACKEND="${PROJECTION_BACKEND:-embedded}"' in script
    assert "--projection-backend" in script
    assert 'if [[ "${PROJECTION_BACKEND}" == "neo4j" ]]; then' in script
    assert 'args+=(--neo4j-uri "${NEO4J_URI}")' in script
    assert 'if [[ "${PROJECTION_BACKEND}" == "pggraph" && -n "${PGGRAPH_DSN}" ]]; then' in script
    assert 'args+=(--pggraph-dsn "${PGGRAPH_DSN}")' in script
    assert "context-collapse" in script
    assert "graph-traversal" in script
    assert "source-recall" in script
    assert "temporal-recall" in script
    assert "zaxy benchmark" in script


def test_public_benchmark_guardrail_script_is_reproducible_from_clean_checkout() -> None:
    """Public benchmark guardrails should be runnable without hidden manual steps."""
    script = Path("scripts/benchmark-guardrails.sh").read_text(encoding="utf-8")

    assert ".cache/zaxy/benchmarks/longmemeval_oracle.json" in script
    assert ".cache/zaxy/longmemeval-embeddings.json" in script
    assert "reports/benchmarks/longmemeval-500-publish-20260607/live-benchmark.json" in script
    assert "reports/benchmarks/longmemeval-500-publish-20260607/run-config.md" in script
    assert "reports/benchmarks/longmemeval-500-hash/live-benchmark.json" not in script
    assert "benchmark-inventory" in script
    assert "benchmark-freeze" in script
    assert "benchmark-compare" in script
    assert "--min-mean-score 0.95" in script
    assert "--min-answer-recall-at-5 0.90" in script
    assert "--min-recall-at-5 0.99" in script
    assert "--min-citation-coverage 1.0" in script


def test_benchmark_checkout_contexts_prioritize_synthesis_evidence() -> None:
    """Answer-bearing synthesis evidence should stay inside Answer@5."""
    checkout = SimpleNamespace(
        quality={"answerability": "answer_from_memory", "confidence": 0.9},
        diagnostics={},
        current_facts=[
            {
                "content": f"ordinary fact {index}",
                "citation": f"eventloom://benchmark/events/{index}#hash",
                "source_lane": "verbatim",
                "score": 1.0,
            }
            for index in range(1, 6)
        ],
        evidence=[
            {
                "content": (
                    "zaxy_synthesis_bundle=true\n"
                    "issue_candidate=GPS system not functioning correctly"
                ),
                "citation": "eventloom://benchmark/events/99#hash",
                "source_lane": "verbatim",
                "score": 1.2,
            }
        ],
    )

    contexts = _benchmark_contexts_from_checkout(checkout)

    assert "issue_candidate=GPS system not functioning correctly" in contexts[0]
    assert "checkout_item=evidence" in contexts[0]


def test_benchmark_checkout_contexts_keep_supporting_evidence_with_compact_contexts() -> None:
    """Compact checkout contexts should not hide answer-bearing cited evidence."""
    checkout = SimpleNamespace(
        quality={"answerability": "answer_from_memory", "confidence": 0.9},
        diagnostics={
            "compact_contexts": [
                "memory_checkout_compact=true\nsummary=Use the cited facts below."
            ],
        },
        current_facts=[
            {
                "content": "The answer is 3 weeks ago.",
                "citation": "eventloom://benchmark/events/1#hash",
                "source_lane": "verbatim",
                "score": 1.0,
            }
        ],
        evidence=[],
    )

    contexts = _benchmark_contexts_from_checkout(checkout)

    assert contexts[0].startswith("memory_checkout_compact=true")
    assert any("The answer is 3 weeks ago." in context for context in contexts)


def test_parse_benchmark_baselines_allows_zaxy_only_runs() -> None:
    """Operators should be able to skip baselines once release floors are established."""
    assert _parse_benchmark_baselines("none", allow_centroid=False) == ()


def test_graph_traversal_workload_is_frozen_and_requires_linked_events(tmp_path: Path) -> None:
    """Graph traversal should require crossing goal-task-completion edges."""
    eventlog, cases, workload = build_graph_traversal_workload(
        tmp_path / "graph-traversal.jsonl",
        subjects=4,
    )
    events = eventlog.read_all()

    assert workload.version == GRAPH_TRAVERSAL_WORKLOAD_VERSION
    assert workload.subjects == 4
    assert workload.event_count == 16
    assert workload.case_count == 4
    assert workload.lanes == ("graph-traversal",)
    assert {case.category for case in cases} == {"graph-traversal"}
    assert [event.type for event in events[:3]] == [
        "goal.created",
        "task.proposed",
        "task.completed",
    ]
    assert "graph-finisher-0000" not in events[1].payload["summary"]
    assert "graph-finisher-distractor-0000" not in events[1].payload["summary"]
    assert "graph-finisher-0000" not in events[2].payload["summary"]
    assert any(event.actor == "graph-finisher-distractor-0000" for event in events)
    assert all(case.identity_terms for case in cases)
    assert any(
        case.expected_terms == ("graph-finisher-0000", "graph-task-0000")
        for case in cases
    )
    assert any("goalTitle" in event.payload for event in events)
    assert workload.sha256 == workload_fingerprint(
        eventlog,
        cases,
        GRAPH_TRAVERSAL_WORKLOAD_VERSION,
    )


def test_source_recall_workload_is_frozen_and_cited(tmp_path: Path) -> None:
    """MemPalace-comparable source recall should target exact cited sources."""
    eventlog, cases, workload = build_source_recall_workload(
        tmp_path / "source-recall.jsonl",
        documents=4,
    )
    events = eventlog.read_all()
    corpus = corpus_from_event_log(eventlog)

    assert workload.version == SOURCE_RECALL_WORKLOAD_VERSION
    assert workload.documents == 4
    assert workload.event_count == 8
    assert workload.case_count == 4
    assert workload.lanes == ("source-recall",)
    assert {case.category for case in cases} == {"source-recall"}
    assert all(case.source_terms for case in cases)
    assert any(
        case.source_terms == ("source-recall/target/service-0000.md",)
        for case in cases
    )
    assert any("source_recall_answer_code" in event.payload for event in events)
    assert all("eventloom://benchmark/events/" in chunk.text for chunk in corpus)
    assert workload.sha256 == workload_fingerprint(
        eventlog,
        cases,
        SOURCE_RECALL_WORKLOAD_VERSION,
    )


def test_context_collapse_workload_is_frozen_and_checkpoint_backed(tmp_path: Path) -> None:
    """Context collapse should require recovering compact memory after noisy turns."""
    eventlog, cases, workload = build_context_collapse_workload(
        tmp_path / "context-collapse.jsonl",
        sessions=3,
        turns_per_session=5,
    )
    events = eventlog.read_all()

    assert workload.version == CONTEXT_COLLAPSE_WORKLOAD_VERSION
    assert workload.sessions == 3
    assert workload.event_count == 18
    assert workload.case_count == 3
    assert workload.lanes == ("context-collapse",)
    assert {case.category for case in cases} == {"context-collapse"}
    assert all(case.identity_terms for case in cases)
    assert any(case.expected_terms == ("collapseanswer0000",) for case in cases)
    assert [event.type for event in events[:5]] == ["transcript.turn"] * 5
    assert events[5].type == "hook.checkpoint"
    assert "collapseanswer0000" in events[5].payload["summary"]
    assert all(
        "collapseanswer0000" not in event.payload.get("content", "")
        for event in events[:5]
    )
    assert workload.sha256 == workload_fingerprint(
        eventlog,
        cases,
        CONTEXT_COLLAPSE_WORKLOAD_VERSION,
    )


def test_context_collapse_markdown_baseline_loses_late_checkpoint(
    tmp_path: Path,
) -> None:
    """Flat context windows should expose context-collapse misses."""
    eventlog, cases, workload = build_context_collapse_workload(
        tmp_path / "context-collapse.jsonl",
        sessions=1,
        turns_per_session=6,
    )
    corpus = corpus_from_event_log(eventlog)
    report = benchmark_retrievers(
        {"md": MarkdownRetriever(corpus)},
        cases,
        runs=1,
        limit=3,
        workload=workload,
    )

    run = report.runs[0]
    assert run.missing_expected == ("collapseanswer0000",)
    assert run.identity_recall is not None
    assert run.identity_recall < 1.0


def test_mempalace_inventory_lists_required_product_proof_lanes(tmp_path: Path) -> None:
    """The release inventory should name every MemPalace-comparable proof lane."""
    inventory = build_mempalace_workload_inventory(
        tmp_path,
        subjects=2,
        documents=2,
        sessions=2,
    )
    lanes = {entry.lane: entry for entry in inventory}
    text = format_mempalace_workload_inventory(inventory)

    assert set(lanes) == {
        "temporal-recall",
        "source-recall",
        "graph-traversal",
        "context-collapse",
    }
    assert lanes["temporal-recall"].required_metrics == (
        "mean_score",
        "citation_coverage",
    )
    assert lanes["source-recall"].required_metrics == (
        "mean_score",
        "source_recall",
        "citation_coverage",
    )
    assert lanes["graph-traversal"].required_metrics == (
        "mean_score",
        "identity_recall",
        "citation_coverage",
    )
    assert lanes["context-collapse"].required_metrics == (
        "mean_score",
        "identity_recall",
        "citation_coverage",
        "approx_tokens",
    )
    assert all(entry.sha256 for entry in inventory)
    assert all(entry.event_count > 0 for entry in inventory)
    assert all(entry.case_count > 0 for entry in inventory)
    assert "MemPalace-Comparable Benchmark Inventory" in text
    assert "| context-collapse |" in text
    assert "checkpoint recovery after noisy transcript context" in text


def test_benchmark_inventory_command_emits_json(tmp_path: Path) -> None:
    """The CLI inventory should be usable by release automation."""
    result = CliRunner().invoke(
        app,
        [
            "benchmark-inventory",
            "--output-dir",
            str(tmp_path / "inventory"),
            "--subjects",
            "1",
            "--documents",
            "1",
            "--sessions",
            "1",
            "--json",
        ],
    )

    assert result.exit_code == 0
    payload = json.loads(result.output)
    assert [entry["lane"] for entry in payload] == [
        "temporal-recall",
        "source-recall",
        "graph-traversal",
        "context-collapse",
    ]
    assert payload[-1]["required_metrics"] == [
        "mean_score",
        "identity_recall",
        "citation_coverage",
        "approx_tokens",
    ]


def test_benchmark_report_tracks_source_recall() -> None:
    """A correct answer should separately report whether the cited source matched."""
    case = BenchmarkCase(
        name="source-1",
        query="Which runbook records source-answer-0001?",
        expected_terms=("source-answer-0001",),
        source_terms=("source-recall/target/service-0001.md",),
        category="source-recall",
    )

    report = benchmark_retrievers(
        {
            "target-source": MarkdownRetriever(
                (
                    BenchmarkChunk(
                        "target",
                        (
                            "source-answer-0001 "
                            "source-recall/target/service-0001.md "
                            "eventloom://benchmark/events/1#abc"
                        ),
                    ),
                )
            ),
            "wrong-source": MarkdownRetriever(
                (
                    BenchmarkChunk(
                        "wrong",
                        (
                            "source-answer-0001 "
                            "source-recall/distractor/service-0001.md "
                            "eventloom://benchmark/events/2#def"
                        ),
                    ),
                )
            ),
        },
        (case,),
        runs=1,
        limit=1,
    )

    runs = {run.backend: run for run in report.runs}
    summaries = {summary.backend: summary for summary in report.summaries}
    category_summaries = {
        summary.backend: summary for summary in report.category_summaries
    }
    markdown = report_to_markdown(report)

    assert runs["target-source"].source_recall == 1.0
    assert runs["target-source"].source_hits == ("source-recall/target/service-0001.md",)
    assert runs["wrong-source"].source_recall == 0.0
    assert runs["wrong-source"].missing_sources == (
        "source-recall/target/service-0001.md",
    )
    assert summaries["target-source"].mean_source_recall == 1.0
    assert summaries["wrong-source"].mean_source_recall == 0.0
    assert category_summaries["target-source"].mean_source_recall == 1.0
    assert "Source recall" in markdown


def test_benchmark_report_tracks_recall_at_k() -> None:
    """LongMemEval-style reports should expose Recall@K source recovery."""
    case = BenchmarkCase(
        name="identity-1",
        query="What source contains my trip memory?",
        expected_terms=("trip answer",),
        identity_terms=("answer-session-1",),
        category="longmemeval:multi-session",
    )

    class OrderedRetriever:
        def __init__(self, contexts: tuple[str, ...]) -> None:
            self._contexts = contexts

        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point
            return list(self._contexts[:limit])

    report = benchmark_retrievers(
        {
            "late-hit": OrderedRetriever(
                tuple(
                    f"distractor {index}"
                    for index in range(4)
                )
                + (
                    "answer-session-1 contains the trip answer",
                )
            ),
            "miss": OrderedRetriever(("unrelated trip answer",)),
        },
        (case,),
        runs=1,
        limit=5,
    )

    runs = {run.backend: run for run in report.runs}
    summaries = {summary.backend: summary for summary in report.summaries}
    markdown = report_to_markdown(report)

    assert runs["late-hit"].recall_at_1 == 0.0
    assert runs["late-hit"].recall_at_5 == 1.0
    assert runs["late-hit"].recall_at_10 == 1.0
    assert runs["miss"].recall_at_5 == 0.0
    assert summaries["late-hit"].mean_recall_at_5 == 1.0
    assert summaries["miss"].mean_recall_at_5 == 0.0
    assert "Recall@5" in markdown


def test_benchmark_report_separates_answer_and_evidence_recall() -> None:
    """Answer presence and source identity presence should be visible separately."""
    case = BenchmarkCase(
        name="answered-without-source-id",
        query="Which music service did I mention?",
        expected_terms=("Spotify",),
        identity_terms=("answer_session_1",),
    )

    class AnswerOnlyRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return ["I mentioned Spotify as my music streaming service."]

    report = benchmark_retrievers(
        {"answer-only": AnswerOnlyRetriever()},
        (case,),
        runs=1,
        limit=5,
    )

    run = report.runs[0]
    assert run.score == 1.0
    assert run.identity_recall == 0.0
    assert run.recall_at_5 == 0.0
    assert run.answer_recall_at_5 == 1.0
    assert run.miss_category == "projection_miss"
    assert report.miss_taxonomy == {
        "answer-only": {
            "projection_miss": 1,
        }
    }


def test_benchmark_miss_taxonomy_classifies_ranking_and_synthesis_misses() -> None:
    """Run-level miss categories should identify the likely failure layer."""
    answer_case = BenchmarkCase(
        name="answer-case",
        query="Which service?",
        expected_terms=("Spotify",),
        identity_terms=("answer-session",),
    )
    source_case = BenchmarkCase(
        name="source-case",
        query="Which source?",
        expected_terms=("Spotify",),
        identity_terms=("answer-session",),
    )

    class MixedRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point, limit
            if "service" in query.casefold():
                return [
                    "distractor service note 1",
                    "distractor service note 2",
                    "distractor service note 3",
                    "distractor service note 4",
                    "distractor service note 5",
                    "answer-session says Spotify",
                ]
            return ["answer-session contains no answer text"]

    report = benchmark_retrievers(
        {"mixed": MixedRetriever()},
        (answer_case, source_case),
        runs=1,
        limit=10,
    )

    runs = {run.case_name: run for run in report.runs}
    assert runs["answer-case"].miss_category == "ranking_miss", runs["answer-case"]
    assert runs["source-case"].miss_category == "synthesis_miss", runs["source-case"]
    assert summarize_miss_taxonomy(report.runs) == {
        "mixed": {
            "ranking_miss": 1,
            "synthesis_miss": 1,
        }
    }
    markdown = report_to_markdown(report)
    assert "## Miss taxonomy" in markdown
    assert "ranking_miss" in markdown
    assert "synthesis_miss" in markdown


def test_temporal_recall_workload_is_frozen_and_source_cited(tmp_path: Path) -> None:
    """MemPalace-comparable temporal recall should be reproducible and cited."""
    eventlog, cases, workload = build_temporal_recall_workload(
        tmp_path / "temporal-recall.jsonl",
        subjects=3,
    )
    events = eventlog.read_all()
    corpus = corpus_from_event_log(eventlog)

    assert workload.version == TEMPORAL_RECALL_WORKLOAD_VERSION
    assert workload.subjects == 3
    assert workload.event_count == 9
    assert workload.case_count == 9
    assert workload.lanes == ("temporal-recall",)
    assert {case.category for case in cases} == {"temporal-recall"}
    assert all(case.temporal_point for case in cases)
    assert any(case.expected_terms == ("workspace=workspace-alpha-0000",) for case in cases)
    assert any(case.expected_terms == ("workspace=workspace-beta-0000",) for case in cases)
    assert any(case.expected_terms == ("workspace=workspace-gamma-0000",) for case in cases)
    assert all("source_path" in event.payload for event in events)
    assert all("eventloom://benchmark/events/" in chunk.text for chunk in corpus)
    assert workload.sha256 == workload_fingerprint(
        eventlog,
        cases,
        TEMPORAL_RECALL_WORKLOAD_VERSION,
    )


def test_benchmark_report_tracks_successful_citation_coverage() -> None:
    """A correct answer should disclose whether it carried a provenance citation."""
    case = BenchmarkCase(
        name="temporal-1",
        query="What workspace was active?",
        expected_terms=("workspace-alpha",),
        category="temporal-recall",
    )

    report = benchmark_retrievers(
        {
            "cited": MarkdownRetriever(
                (BenchmarkChunk("cited", "workspace-alpha eventloom://benchmark/events/1#abc"),)
            ),
            "uncited": MarkdownRetriever((BenchmarkChunk("uncited", "workspace-alpha"),)),
        },
        (case,),
        runs=1,
        limit=1,
    )

    runs = {run.backend: run for run in report.runs}
    summaries = {summary.backend: summary for summary in report.summaries}
    category_summaries = {
        summary.backend: summary for summary in report.category_summaries
    }
    markdown = report_to_markdown(report)

    assert runs["cited"].citation_count == 1
    assert runs["cited"].citation_coverage == 1.0
    assert runs["uncited"].citation_count == 0
    assert runs["uncited"].citation_coverage == 0.0
    assert summaries["cited"].mean_citation_coverage == 1.0
    assert summaries["uncited"].mean_citation_coverage == 0.0
    assert category_summaries["cited"].mean_citation_coverage == 1.0
    assert "Citation coverage" in markdown


def test_longmemeval_workload_loads_public_memory_dataset(tmp_path: Path) -> None:
    """LongMemEval loader should convert answer sessions into identity-recall cases."""
    dataset = tmp_path / "longmemeval.json"
    dataset.write_text(
        json.dumps(
            [
                {
                    "question_id": "q1",
                    "question_type": "single-session-user",
                    "question": "What degree did I graduate with?",
                    "answer": "Business Administration",
                    "question_date": "2023/05/30 (Tue) 09:00",
                    "answer_session_ids": ["answer-1"],
                    "haystack_dates": ["2023/05/20 (Sat) 02:21", "2023/05/21 (Sun) 03:24"],
                    "haystack_session_ids": ["distractor-1", "answer-1"],
                    "haystack_sessions": [
                        [{"role": "user", "content": "What is a graph database?"}],
                        [
                            {
                                "role": "user",
                                "content": "I graduated with a Business Administration degree.",
                            }
                        ],
                    ],
                },
                {
                    "question_id": "q2",
                    "question_type": "multi-session-assistant",
                    "question": "Which project did we rename?",
                    "answer": "Atlas",
                    "answer_session_ids": ["answer-2"],
                    "haystack_dates": ["2023/05/22 (Mon) 08:00"],
                    "haystack_session_ids": ["answer-2"],
                    "haystack_sessions": [
                        [{"role": "assistant", "content": "We renamed Project Atlas."}]
                    ],
                },
            ]
        ),
        encoding="utf-8",
    )

    eventlog, cases, workload = build_longmemeval_workload(
        tmp_path / "longmemeval.jsonl",
        dataset,
        questions=1,
    )
    corpus = corpus_from_event_log(eventlog)

    assert workload.version == LONGMEMEVAL_WORKLOAD_VERSION
    assert workload.subjects == 1
    assert workload.sessions == 2
    assert workload.event_count == 2
    assert workload.case_count == 1
    assert workload.lanes == ("longmemeval",)
    assert cases[0].name == "longmemeval-q1"
    assert cases[0].expected_terms == ("Business Administration",)
    assert cases[0].identity_terms == ("answer-1",)
    assert cases[0].temporal_point == "2023-05-30T09:00:00Z"
    assert any("longmemeval_session_id=answer-1" in chunk.text for chunk in corpus)
    assert workload.sha256 == workload_fingerprint(
        eventlog,
        cases,
        LONGMEMEVAL_WORKLOAD_VERSION,
    )


def test_longmemeval_workload_fingerprint_ignores_transient_eventloom_seal(
    tmp_path: Path,
) -> None:
    """LongMemEval workload identity should be stable across regenerated logs."""
    dataset = tmp_path / "longmemeval.json"
    dataset.write_text(
        json.dumps(
            [
                {
                    "question_id": "q1",
                    "question_type": "single-session-user",
                    "question": "What degree did I graduate with?",
                    "answer": "Business Administration",
                    "answer_session_ids": ["answer-1"],
                    "haystack_dates": ["2023/05/20 (Sat) 02:21"],
                    "haystack_session_ids": ["answer-1"],
                    "haystack_sessions": [
                        [
                            {
                                "role": "user",
                                "content": "I graduated with a Business Administration degree.",
                            }
                        ],
                    ],
                }
            ]
        ),
        encoding="utf-8",
    )

    first_log, first_cases, first_workload = build_longmemeval_workload(
        tmp_path / "first.jsonl",
        dataset,
        questions=1,
    )
    second_log, second_cases, second_workload = build_longmemeval_workload(
        tmp_path / "second.jsonl",
        dataset,
        questions=1,
    )

    assert first_workload.sha256 == second_workload.sha256
    assert workload_fingerprint(first_log, first_cases, LONGMEMEVAL_WORKLOAD_VERSION) == (
        workload_fingerprint(second_log, second_cases, LONGMEMEVAL_WORKLOAD_VERSION)
    )


def test_longmemeval_workload_chunks_large_sessions_for_embedding_limits(tmp_path: Path) -> None:
    """LongMemEval sessions should be bounded before vector baseline embedding."""
    dataset = tmp_path / "longmemeval-large.json"
    long_content = "alpha " * 3000 + "Business Administration " + "omega " * 3000
    dataset.write_text(
        json.dumps(
            [
                {
                    "question_id": "q1",
                    "question_type": "single-session-user",
                    "question": "What degree did I graduate with?",
                    "answer": "Business Administration",
                    "answer_session_ids": ["answer-1"],
                    "haystack_dates": ["2023/05/20 (Sat) 02:21"],
                    "haystack_session_ids": ["answer-1"],
                    "haystack_sessions": [[{"role": "user", "content": long_content}]],
                }
            ]
        ),
        encoding="utf-8",
    )

    eventlog, cases, workload = build_longmemeval_workload(
        tmp_path / "longmemeval-large.jsonl",
        dataset,
    )
    events = eventlog.read_all()
    corpus = corpus_from_event_log(eventlog)

    assert cases[0].identity_terms == ("answer-1",)
    assert workload.sessions == 1
    assert workload.event_count == len(events)
    assert len(events) > 1
    assert all(event.payload["longmemeval_session_id"] == "answer-1" for event in events)
    assert all("longmemeval_session_id=answer-1" in chunk.text for chunk in corpus)
    assert max(len(chunk.text) for chunk in corpus) < 9000
    assert any("Business Administration" in chunk.text for chunk in corpus)


def test_longmemeval_workload_appends_events_in_one_batch(tmp_path: Path) -> None:
    """Large public benchmark workloads should avoid per-event append overhead."""
    dataset = tmp_path / "longmemeval-batch.json"
    dataset.write_text(
        json.dumps(
            [
                {
                    "question_id": "q1",
                    "question_type": "single-session-user",
                    "question": "What is my project?",
                    "answer": "Atlas",
                    "answer_session_ids": ["answer-1"],
                    "haystack_dates": ["2023/05/20 (Sat) 02:21"],
                    "haystack_session_ids": ["answer-1"],
                    "haystack_sessions": [
                        [
                            {"role": "user", "content": "My project is Atlas."},
                            {
                                "role": "user",
                                "content": "Atlas is still the right name.",
                                "has_answer": True,
                            },
                        ]
                    ],
                }
            ]
        ),
        encoding="utf-8",
    )

    with patch.object(EventLog, "append", side_effect=AssertionError("use append_many")):
        eventlog, cases, workload = build_longmemeval_workload(
            tmp_path / "longmemeval-batch.jsonl",
            dataset,
        )

    events = eventlog.read_all()
    assert len(events) == 2
    assert workload.event_count == 2
    assert cases[0].identity_terms == ("answer-1",)


def test_longmemeval_workload_projects_annotated_answer_turns(tmp_path: Path) -> None:
    """Annotated source turns should become compact salient memory candidates."""
    dataset = tmp_path / "longmemeval-answer-turn.json"
    dataset.write_text(
        json.dumps(
            [
                {
                    "question_id": "q1",
                    "question_type": "single-session-user",
                    "question": "How long is my daily commute to work?",
                    "answer": "45 minutes each way",
                    "answer_session_ids": ["answer-1"],
                    "haystack_dates": ["2023/05/20 (Sat) 02:21"],
                    "haystack_session_ids": ["answer-1"],
                    "haystack_sessions": [
                        [
                            {
                                "role": "user",
                                "content": "Audiobooks help during my commute.",
                            },
                            {
                                "role": "user",
                                "content": "My daily commute takes 45 minutes each way.",
                                "has_answer": True,
                            },
                        ]
                    ],
                }
            ]
        ),
        encoding="utf-8",
    )

    eventlog, _, _ = build_longmemeval_workload(
        tmp_path / "longmemeval-answer-turn.jsonl",
        dataset,
    )
    corpus = corpus_from_event_log(eventlog)

    salient_chunks = [
        chunk.text for chunk in corpus
        if "longmemeval_salient_memory_turn=true" in chunk.text
    ]

    assert len(salient_chunks) == 1
    assert "turn_index=2" in salient_chunks[0]
    assert "45 minutes each way" in salient_chunks[0]


def test_bm25_boosts_salient_memory_turns() -> None:
    """Source-salient memory turns should outrank generic matching context."""
    corpus = (
        BenchmarkChunk(
            "generic",
            "yoga classes yoga classes yoga classes in my area",
        ),
        BenchmarkChunk(
            "salient",
            "\n".join(
                [
                    "longmemeval_salient_memory_turn=true",
                    "I cannot make it to Serenity Yoga.",
                ]
            ),
        ),
    )

    results = BM25Retriever(corpus).query("Where do I take yoga classes?", limit=1)

    assert results == [corpus[1].text]


def test_bm25_expands_common_education_paraphrases() -> None:
    """Education queries should match normal undergrad/CS phrasing in memory."""
    corpus = (
        BenchmarkChunk(
            "generic",
            "I completed a Bachelor's degree in Business Administration.",
        ),
        BenchmarkChunk(
            "answer",
            "longmemeval_salient_memory_turn=true "
            "longmemeval_session_id=answer-1 I completed my undergrad in CS from UCLA.",
        ),
    )

    results = BM25Retriever(corpus).query(
        "Where did I complete my Bachelor's degree in Computer Science?",
        limit=1,
    )

    assert results == [corpus[1].text]


def test_bm25_splits_hyphenated_query_compounds() -> None:
    """Compound query words should match equivalent source terms."""
    corpus = (
        BenchmarkChunk(
            "distractor",
            "I tracked household expenses and grocery costs this month.",
        ),
        BenchmarkChunk(
            "answer",
            "I spent $25 on a bike chain and $40 on bike lights.",
        ),
    )

    results = BM25Retriever(corpus).query("bike-related expenses", limit=1)

    assert results == [corpus[1].text]


def test_live_benchmark_compares_all_retriever_backends(tmp_path: Path) -> None:
    """The benchmark core should compare md, BM25, vector, md+vector, and zaxy rows."""
    log = build_competitive_event_log(tmp_path / "bench.jsonl")
    corpus = corpus_from_event_log(log)
    provider = HashEmbeddingProvider(dimension=64)

    retrievers = {
        "md": MarkdownRetriever(corpus),
        "bm25": BM25Retriever(corpus),
        "vector": VectorRetriever(corpus, provider),
        "md+vector": MarkdownVectorRetriever(corpus, provider),
        "zaxy": MarkdownRetriever(corpus),
    }

    report = benchmark_retrievers(retrievers, competitive_cases(), runs=2, limit=5)

    assert {summary.backend for summary in report.summaries} == {
        "md",
        "bm25",
        "vector",
        "md+vector",
        "zaxy",
    }
    assert all(summary.runs == 2 for summary in report.summaries)
    assert all(summary.case_count == len(competitive_cases()) for summary in report.summaries)
    assert all(summary.latency_ms_p95 >= summary.latency_ms_p50 for summary in report.summaries)


def test_cached_embedding_provider_persists_embeddings_between_runs(tmp_path: Path) -> None:
    """Hosted benchmark embeddings should be reusable across process runs."""
    cache_path = tmp_path / "embeddings.json"

    class CountingProvider:
        dimension = 3

        def __init__(self) -> None:
            self.calls = 0

        def embed(self, text: str) -> list[float]:
            self.calls += 1
            return [float(len(text)), 1.0, 0.0]

    first_provider = CountingProvider()
    first = CachedEmbeddingProvider(first_provider, cache_path=cache_path)

    assert first.embed("alpha") == [5.0, 1.0, 0.0]
    assert first_provider.calls == 1
    first.flush()
    assert cache_path.is_file()

    second_provider = CountingProvider()
    second = CachedEmbeddingProvider(second_provider, cache_path=cache_path)

    assert second.embed("alpha") == [5.0, 1.0, 0.0]
    assert second_provider.calls == 0
    assert second.cache_size == 1


def test_cached_embedding_provider_batches_and_atomically_flushes(tmp_path: Path) -> None:
    """Large hosted benchmark caches should not rewrite the full file on every miss."""
    cache_path = tmp_path / "embeddings.json"

    class CountingProvider:
        dimension = 3

        def embed(self, text: str) -> list[float]:
            return [float(len(text)), 1.0, 0.0]

    provider = CachedEmbeddingProvider(CountingProvider(), cache_path=cache_path, flush_every=2)

    assert provider.embed("alpha") == [5.0, 1.0, 0.0]
    assert not cache_path.exists()

    assert provider.embed("beta") == [4.0, 1.0, 0.0]
    assert json.loads(cache_path.read_text(encoding="utf-8")) == {
        "alpha": [5.0, 1.0, 0.0],
        "beta": [4.0, 1.0, 0.0],
    }

    assert provider.embed("gamma") == [5.0, 1.0, 0.0]
    assert "gamma" not in json.loads(cache_path.read_text(encoding="utf-8"))

    provider.flush()

    assert json.loads(cache_path.read_text(encoding="utf-8")) == {
        "alpha": [5.0, 1.0, 0.0],
        "beta": [4.0, 1.0, 0.0],
        "gamma": [5.0, 1.0, 0.0],
    }
    assert not cache_path.with_suffix(".tmp").exists()


async def test_benchmark_projection_marker_round_trips() -> None:
    """Projection markers should make expensive benchmark ingestion reusable."""
    calls: list[tuple[str, dict[str, object]]] = []

    class FakeDriver:
        async def execute_query(self, cypher: str, **kwargs: object) -> tuple[list[object], None, None]:
            calls.append((cypher, kwargs))
            if "MATCH (p:ZaxyBenchmarkProjection" in cypher:
                return ([{"key": kwargs["key"]}] if kwargs["key"] == "present" else [], None, None)
            return ([], None, None)

    graph = SimpleNamespace(_driver=FakeDriver())

    assert await _benchmark_projection_present(graph, "present") is True  # type: ignore[arg-type]
    assert await _benchmark_projection_present(graph, "missing") is False  # type: ignore[arg-type]
    await _mark_benchmark_projection(
        graph,  # type: ignore[arg-type]
        "present",
        [
            SimpleNamespace(seq=1, hash="abc"),
            SimpleNamespace(seq=2, hash="def"),
        ],
    )

    marker_call = calls[-1]
    assert "MERGE (p:ZaxyBenchmarkProjection {key: $key})" in marker_call[0]
    assert marker_call[1]["event_count"] == 2
    assert marker_call[1]["latest_seq"] == 2
    assert marker_call[1]["latest_hash"] == "def"


async def test_reset_benchmark_graph_deletes_in_relationship_and_node_batches() -> None:
    """Benchmark reset should not delete a large graph in one transaction."""
    calls: list[tuple[str, dict[str, object]]] = []

    class FakeDriver:
        def __init__(self) -> None:
            self.relationship_counts = [1000, 500, 0]
            self.node_counts = [1000, 1, 0]

        async def execute_query(self, cypher: str, **kwargs: object) -> tuple[list[dict[str, int]], None, None]:
            calls.append((cypher, kwargs))
            if "MATCH ()-[r]->()" in cypher:
                return ([{"deleted": self.relationship_counts.pop(0)}], None, None)
            if "MATCH (n)" in cypher:
                return ([{"deleted": self.node_counts.pop(0)}], None, None)
            raise AssertionError(f"unexpected reset query: {cypher}")

    totals = await _reset_benchmark_graph(
        SimpleNamespace(_driver=FakeDriver()),  # type: ignore[arg-type]
        batch_size=1000,
    )

    assert totals == {"relationships": 1500, "nodes": 1001}
    assert len(calls) == 6
    assert all(call[1]["batch_size"] == 1000 for call in calls)
    assert "MATCH (n) DETACH DELETE n" not in "\n".join(call[0] for call in calls)


def test_benchmark_projection_cache_key_ignores_eventloom_seal(tmp_path: Path) -> None:
    """Projection reuse should survive regenerated Eventloom timestamps and hashes."""
    first_log = EventLog(tmp_path / "first.jsonl")
    second_log = EventLog(tmp_path / "second.jsonl")
    first_log.append(
        "transcript.turn",
        actor="user",
        payload={"content": "Stable benchmark memory."},
        timestamp="2024-01-01T00:00:00Z",
    )
    second_log.append(
        "transcript.turn",
        actor="user",
        payload={"content": "Stable benchmark memory."},
        timestamp="2024-01-02T00:00:00Z",
    )
    cases = (
        BenchmarkCase(
            name="stable",
            query="What memory is stable?",
            expected_terms=("Stable benchmark memory",),
        ),
    )
    first_workload = BenchmarkWorkload.from_event_log(
        first_log,
        cases,
        version="fixture-v1",
    )
    second_workload = BenchmarkWorkload.from_event_log(
        second_log,
        cases,
        version="fixture-v1",
    )

    assert first_workload.sha256 == second_workload.sha256
    assert benchmark_projection_cache_key(
        first_log,
        cases,
        first_workload,
        "hash:1536",
    ) == benchmark_projection_cache_key(
        second_log,
        cases,
        second_workload,
        "hash:1536",
    )


async def test_live_benchmark_reports_progress_for_each_backend_case() -> None:
    """Long-running live benchmarks should emit progress events before completion."""
    case = BenchmarkCase(name="case-1", query="alpha", expected_terms=("alpha",))
    corpus = (BenchmarkChunk("alpha", "alpha context"),)
    progress: list[dict[str, object]] = []

    class FakeZaxy:
        async def query_async(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return ["alpha context"]

    report = await benchmark_live_retrievers(
        {"md": MarkdownRetriever(corpus)},
        FakeZaxy(),  # type: ignore[arg-type]
        (case,),
        runs=1,
        limit=1,
        progress_callback=progress.append,
    )

    assert report.summaries
    assert [item["backend"] for item in progress] == ["md", "zaxy"]
    assert all(item["completed"] <= item["total"] for item in progress)


async def test_live_benchmark_can_include_memory_checkout_backend() -> None:
    """Benchmarks should be able to exercise the production Memory Checkout path."""
    case = BenchmarkCase(
        name="checkout-aggregation",
        query="How many weddings did I attend?",
        expected_terms=("answerability=answer_from_memory", "evidence_plan_mode=multi_source_aggregation"),
        identity_terms=("answer-1", "answer-2"),
    )
    progress: list[dict[str, object]] = []

    class FakeZaxy:
        async def query_async(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return ["plain zaxy context"]

    class FakeCheckout:
        async def query_async(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return [
                "memory_checkout=true\n"
                "answerability=answer_from_memory\n"
                "evidence_plan_mode=multi_source_aggregation\n"
                "source_id=answer-1\n"
                "source_id=answer-2"
            ]

    report = await benchmark_live_retrievers(
        {},
        FakeZaxy(),  # type: ignore[arg-type]
        (case,),
        runs=1,
        limit=1,
        checkout_retriever=FakeCheckout(),  # type: ignore[arg-type]
        progress_callback=progress.append,
    )

    by_backend = {run.backend: run for run in report.runs}
    assert by_backend["zaxy-checkout"].score == 1.0
    assert by_backend["zaxy-checkout"].identity_recall == 1.0
    assert [item["backend"] for item in progress] == ["zaxy", "zaxy-checkout"]


async def test_live_benchmark_can_run_checkout_backend_without_graph_backend() -> None:
    """Checkout-only mode should make the benchmark target explicit."""
    case = BenchmarkCase(
        name="checkout-only",
        query="What should I use?",
        expected_terms=("memory_checkout=true",),
    )

    class FakeZaxy:
        async def query_async(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return ["graph"]

    class FakeCheckout:
        async def query_async(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return ["memory_checkout=true"]

    report = await benchmark_live_retrievers(
        {},
        FakeZaxy(),  # type: ignore[arg-type]
        (case,),
        runs=1,
        checkout_retriever=FakeCheckout(),  # type: ignore[arg-type]
        include_zaxy=False,
    )

    assert [run.backend for run in report.runs] == ["zaxy-checkout"]


async def test_build_live_zaxy_retriever_can_use_projection_backend_factory(tmp_path: Path) -> None:
    """Live benchmarks should be able to target experimental projection backends."""
    log = EventLog(tmp_path / "bench.jsonl")
    log.append("tester", "goal.created", {"goalId": "goal-1", "title": "Compare pgGraph"})
    config = ProjectionBackendConfig(
        backend="pggraph",
        neo4j_uri="bolt://unused",
        neo4j_user="neo4j",
        neo4j_password="testpassword",
        neo4j_ca_cert=None,
        neo4j_trust_all=False,
        pggraph_dsn="postgresql://postgres:postgres@localhost:5432/zaxy",
    )

    class FakeProjectionStore:
        def __init__(self) -> None:
            self.connected = False
            self.schema_initialized = False
            self.upsert_count = 0

        async def connect(self) -> None:
            self.connected = True

        async def init_schema(self) -> None:
            self.schema_initialized = True

        async def upsert_extraction(self, result: object, session_id: str = "default") -> None:
            del result, session_id
            self.upsert_count += 1

        async def search_exact(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_keyword(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_vector(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_traversal(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def has_traversal_edges(self, session_id: str = "default") -> bool:
            del session_id
            return False

        async def close(self) -> None:
            return None

    store = FakeProjectionStore()

    with patch("zaxy_benchmarks.live_benchmark.build_projection_store", return_value=store) as factory:
        retriever, returned_store = await build_live_zaxy_retriever(
            log,
            HashEmbeddingProvider(dimension=8),
            projection_backend_config=config,
        )

    assert isinstance(retriever, ZaxyRetriever)
    assert returned_store is store
    assert store.connected is True
    assert store.schema_initialized is True
    assert store.upsert_count == 1
    factory.assert_called_once_with(config)


async def test_build_live_zaxy_retriever_reuses_current_non_neo4j_projection(tmp_path: Path) -> None:
    """Reusable non-Neo4j benchmark projections should not be upserted again."""
    log = EventLog(tmp_path / "bench.jsonl")
    first = log.append("tester", "goal.created", {"goalId": "goal-1", "title": "Reuse me"})
    config = ProjectionBackendConfig(
        backend="embedded",
        neo4j_uri="bolt://unused",
        neo4j_user="neo4j",
        neo4j_password="testpassword",
        neo4j_ca_cert=None,
        neo4j_trust_all=False,
        embedded_graph_path=tmp_path / "bench.kuzu",
    )

    class FakeProjectionStore:
        def __init__(self) -> None:
            self.upsert_count = 0
            self.marked: list[tuple[str, int]] = []
            self.status_calls = 0

        async def connect(self) -> None:
            return None

        async def init_schema(self) -> None:
            return None

        async def inspect_event_projection_status(
            self,
            session_id: str,
            *,
            eventloom_latest_seq: int | None = None,
            eventloom_latest_hash: str | None = None,
        ) -> GraphEventProjectionStatus:
            self.status_calls += 1
            return GraphEventProjectionStatus(
                session_id=session_id,
                event_count=1,
                latest_seq=eventloom_latest_seq,
                latest_hash=eventloom_latest_hash,
                eventloom_latest_seq=eventloom_latest_seq,
                eventloom_latest_hash=eventloom_latest_hash,
                projection_lag=0,
                latest_hash_matches=True,
                next_event_edges=0,
                previous_event_edges=0,
                missing_chain_links=0,
                integrity_ok=True,
            )

        async def upsert_extraction(self, result: object, session_id: str = "default") -> None:
            del result, session_id
            self.upsert_count += 1

        async def mark_benchmark_projection(self, key: str, events: Sequence[object]) -> None:
            self.marked.append((key, len(events)))

        async def search_exact(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_keyword(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_vector(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_traversal(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def has_traversal_edges(self, session_id: str = "default") -> bool:
            del session_id
            return False

        async def close(self) -> None:
            return None

    store = FakeProjectionStore()

    with patch("zaxy_benchmarks.live_benchmark.build_projection_store", return_value=store):
        _retriever, returned_store = await build_live_zaxy_retriever(
            log,
            HashEmbeddingProvider(dimension=8),
            reuse_projection=True,
            projection_cache_key=f"test:{first.hash}",
            projection_backend_config=config,
        )

    assert returned_store is store
    assert store.status_calls == 1
    assert store.upsert_count == 0


async def test_build_live_zaxy_retriever_reuses_non_neo4j_projection_with_regenerated_event_hashes(
    tmp_path: Path,
) -> None:
    """Benchmark reuse should tolerate regenerated Eventloom hashes for the same workload key."""
    log = EventLog(tmp_path / "bench.jsonl")
    log.append("tester", "goal.created", {"goalId": "goal-1", "title": "Same workload"})
    config = ProjectionBackendConfig(
        backend="embedded",
        neo4j_uri="bolt://unused",
        neo4j_user="neo4j",
        neo4j_password="testpassword",
        neo4j_ca_cert=None,
        neo4j_trust_all=False,
        embedded_graph_path=tmp_path / "bench.kuzu",
    )

    class FakeProjectionStore:
        def __init__(self) -> None:
            self.upsert_count = 0
            self.marked: list[tuple[str, int]] = []

        async def connect(self) -> None:
            return None

        async def init_schema(self) -> None:
            return None

        async def inspect_event_projection_status(
            self,
            session_id: str,
            *,
            eventloom_latest_seq: int | None = None,
            eventloom_latest_hash: str | None = None,
        ) -> GraphEventProjectionStatus:
            return GraphEventProjectionStatus(
                session_id=session_id,
                event_count=1,
                latest_seq=eventloom_latest_seq,
                latest_hash="previous-run-hash",
                eventloom_latest_seq=eventloom_latest_seq,
                eventloom_latest_hash=eventloom_latest_hash,
                projection_lag=0,
                latest_hash_matches=False,
                next_event_edges=0,
                previous_event_edges=0,
                missing_chain_links=0,
                integrity_ok=False,
            )

        async def upsert_extraction(self, result: object, session_id: str = "default") -> None:
            del result, session_id
            self.upsert_count += 1

        async def mark_benchmark_projection(self, key: str, events: Sequence[object]) -> None:
            self.marked.append((key, len(events)))

        async def search_exact(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_keyword(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_vector(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_traversal(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def has_traversal_edges(self, session_id: str = "default") -> bool:
            del session_id
            return False

        async def close(self) -> None:
            return None

    store = FakeProjectionStore()

    with patch("zaxy_benchmarks.live_benchmark.build_projection_store", return_value=store):
        await build_live_zaxy_retriever(
            log,
            HashEmbeddingProvider(dimension=8),
            reuse_projection=True,
            projection_cache_key="same-workload",
            projection_backend_config=config,
        )

    assert store.upsert_count == 0
    assert store.marked == [("same-workload", 1)]


async def test_build_live_zaxy_retriever_marks_non_neo4j_projection_after_ingest(tmp_path: Path) -> None:
    """Successful non-Neo4j benchmark ingestion should persist the semantic reuse key."""
    log = EventLog(tmp_path / "bench.jsonl")
    first = log.append("tester", "goal.created", {"goalId": "goal-1", "title": "Mark me"})
    config = ProjectionBackendConfig(
        backend="embedded",
        neo4j_uri="bolt://unused",
        neo4j_user="neo4j",
        neo4j_password="testpassword",
        neo4j_ca_cert=None,
        neo4j_trust_all=False,
        embedded_graph_path=tmp_path / "bench.kuzu",
    )

    class FakeProjectionStore:
        def __init__(self) -> None:
            self.upsert_count = 0
            self.marked: list[tuple[str, int]] = []

        async def connect(self) -> None:
            return None

        async def init_schema(self) -> None:
            return None

        async def benchmark_projection_present(self, key: str) -> bool:
            del key
            return False

        async def mark_benchmark_projection(self, key: str, events: Sequence[object]) -> None:
            self.marked.append((key, len(events)))

        async def upsert_extraction(self, result: object, session_id: str = "default") -> None:
            del result, session_id
            self.upsert_count += 1

        async def search_exact(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_keyword(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_vector(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_traversal(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def has_traversal_edges(self, session_id: str = "default") -> bool:
            del session_id
            return False

        async def close(self) -> None:
            return None

    store = FakeProjectionStore()

    with patch("zaxy_benchmarks.live_benchmark.build_projection_store", return_value=store):
        await build_live_zaxy_retriever(
            log,
            HashEmbeddingProvider(dimension=8),
            reuse_projection=False,
            projection_cache_key="semantic-workload",
            projection_backend_config=config,
        )

    assert store.upsert_count == 1
    assert store.marked == [("semantic-workload", 1)]
    assert first.seq == 1


async def test_build_live_zaxy_retriever_refuses_to_overwrite_mismatched_reuse_projection(
    tmp_path: Path,
) -> None:
    """A non-empty mismatched reusable projection should fail fast instead of partial ingest."""
    log = EventLog(tmp_path / "bench.jsonl")
    log.append("tester", "goal.created", {"goalId": "goal-1", "title": "Different workload"})
    config = ProjectionBackendConfig(
        backend="embedded",
        neo4j_uri="bolt://unused",
        neo4j_user="neo4j",
        neo4j_password="testpassword",
        neo4j_ca_cert=None,
        neo4j_trust_all=False,
        embedded_graph_path=tmp_path / "bench.kuzu",
    )

    class FakeProjectionStore:
        def __init__(self) -> None:
            self.closed = False
            self.upsert_count = 0

        async def connect(self) -> None:
            return None

        async def init_schema(self) -> None:
            return None

        async def benchmark_projection_present(self, key: str) -> bool:
            del key
            return False

        async def inspect_event_projection_status(
            self,
            session_id: str,
            *,
            eventloom_latest_seq: int | None = None,
            eventloom_latest_hash: str | None = None,
        ) -> GraphEventProjectionStatus:
            return GraphEventProjectionStatus(
                session_id=session_id,
                event_count=1,
                latest_seq=0,
                latest_hash="other",
                eventloom_latest_seq=eventloom_latest_seq,
                eventloom_latest_hash=eventloom_latest_hash,
                projection_lag=1,
                latest_hash_matches=False,
                next_event_edges=0,
                previous_event_edges=0,
                missing_chain_links=0,
                integrity_ok=False,
            )

        async def upsert_extraction(self, result: object, session_id: str = "default") -> None:
            del result, session_id
            self.upsert_count += 1

        async def search_exact(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_keyword(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_vector(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_traversal(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def has_traversal_edges(self, session_id: str = "default") -> bool:
            del session_id
            return False

        async def close(self) -> None:
            self.closed = True

    store = FakeProjectionStore()

    with patch("zaxy_benchmarks.live_benchmark.build_projection_store", return_value=store):
        try:
            await build_live_zaxy_retriever(
                log,
                HashEmbeddingProvider(dimension=8),
                reuse_projection=True,
                projection_cache_key="semantic-workload",
                projection_backend_config=config,
            )
        except RuntimeError as exc:
            assert "Existing projection does not match" in str(exc)
        else:  # pragma: no cover - assertion path
            raise AssertionError("expected mismatched projection reuse to fail")

    assert store.upsert_count == 0
    assert store.closed is True


async def test_build_live_zaxy_retriever_can_reset_non_neo4j_projection_backend(tmp_path: Path) -> None:
    """Experimental backends should be resettable for reproducible benchmark reruns."""
    log = EventLog(tmp_path / "bench.jsonl")
    log.append("tester", "goal.created", {"goalId": "goal-1", "title": "Compare pgGraph"})
    config = ProjectionBackendConfig(
        backend="pggraph",
        neo4j_uri="bolt://unused",
        neo4j_user="neo4j",
        neo4j_password="testpassword",
        neo4j_ca_cert=None,
        neo4j_trust_all=False,
        pggraph_dsn="postgresql://postgres:postgres@localhost:5432/zaxy",
    )

    class FakeProjectionStore:
        def __init__(self) -> None:
            self.reset_count = 0
            self.upsert_count = 0

        async def connect(self) -> None:
            return None

        async def init_schema(self) -> None:
            return None

        async def reset_benchmark_projection(self) -> None:
            self.reset_count += 1

        async def upsert_extraction(self, result: object, session_id: str = "default") -> None:
            del result, session_id
            self.upsert_count += 1

        async def search_exact(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_keyword(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_vector(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def search_traversal(self, *args: object, **kwargs: object) -> list[object]:
            del args, kwargs
            return []

        async def has_traversal_edges(self, session_id: str = "default") -> bool:
            del session_id
            return False

        async def close(self) -> None:
            return None

    store = FakeProjectionStore()

    with patch("zaxy_benchmarks.live_benchmark.build_projection_store", return_value=store):
        _retriever, returned_store = await build_live_zaxy_retriever(
            log,
            HashEmbeddingProvider(dimension=8),
            reset_graph=True,
            projection_backend_config=config,
        )

    assert returned_store is store
    assert store.reset_count == 1
    assert store.upsert_count == 1


async def test_zaxy_checkout_retriever_returns_checkout_contract() -> None:
    """The checkout benchmark backend should expose model-facing diagnostics."""
    corpus = (
        BenchmarkChunk(
            "answer-1",
            (
                "citation=eventloom://benchmark/events/1#abc "
                "session_id=answer-1 I attended Rachel and Mike's wedding."
            ),
        ),
        BenchmarkChunk(
            "answer-2",
            (
                "citation=eventloom://benchmark/events/2#abc "
                "session_id=answer-2 I attended Emily and Sarah's wedding."
            ),
        ),
    )

    class FakeRouter:
        temporal_points: list[str | None] = []

        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, limit, embedding
            self.temporal_points.append(temporal_point)
            return [SimpleNamespace(content="uncited graph summary", source="keyword", score=0.99)]

    router = FakeRouter()
    retriever = ZaxyCheckoutRetriever(
        router,  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async(
        "How many weddings did I attend?",
        temporal_point="2023-05-30T09:00:00Z",
        limit=5,
    )
    output = "\n".join(results)

    assert results[0].startswith("memory_checkout_compact=true")
    assert "memory_checkout=true" in output
    assert "answerability=answer_from_memory" in output
    assert "evidence_plan_mode=multi_source_aggregation" in output
    assert "evidence_plan_satisfied=True" in output
    assert "query_temporal_anchor=true" in output
    assert router.temporal_points == [None]
    assert "source_id=answer-1" in output
    assert "source_id=answer-2" in output
    assert sum(len(result) for result in results) < 7000


async def test_zaxy_checkout_retriever_uses_query_temporal_anchor_for_source_synthesis() -> None:
    """Checkout source synthesis should receive the query-date anchor before assembly."""
    corpus = (
        BenchmarkChunk(
            "festival",
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=festival longmemeval_session_date=2021/06/01 (Tue) "
                "user: I attended the Seattle International Film Festival."
            ),
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return []

    retriever = ZaxyCheckoutRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async(
        "How many months ago did I attend the Seattle International Film Festival?",
        temporal_point="2021-10-02T09:44:00Z",
        limit=5,
    )
    output = "\n".join(results)

    assert "candidate_type=relative_temporal_anchor" in output
    assert "relative_temporal_anchor_answer=4 months ago" in output
    assert "relative_temporal_anchor_source_id=festival" in output


async def test_zaxy_checkout_retriever_synthesizes_from_graph_salient_sources() -> None:
    """Checkout synthesis should use graph-salient cited turns when lexical backfill is sparse."""

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [
                SimpleNamespace(
                    content=(
                        "longmemeval_session_id=parking user: I got a parking ticket "
                        "near my work for $50."
                    ),
                    citation="eventloom://benchmark/events/parking#abc",
                    source="graph",
                    score=0.99,
                ),
                SimpleNamespace(
                    content=(
                        "longmemeval_session_id=wash user: I had a car wash on "
                        "February 3rd that cost $15."
                    ),
                    citation="eventloom://benchmark/events/wash#def",
                    source="graph",
                    score=0.98,
                ),
            ]

    class EmptyRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return []

    retriever = ZaxyCheckoutRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=EmptyRetriever(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How much did I spend on car wash and parking ticket?",
        limit=5,
    )
    output = "\n".join(results)

    assert "checkout_answer_candidate=true" in output
    assert "answer=$65" in output
    assert "source_id=parking" in output
    assert "source_id=wash" in output


async def test_zaxy_checkout_retriever_uses_graph_sources_when_lexical_bundle_is_noisy() -> None:
    """A noisy lexical bundle should not block graph-salient operands from checkout synthesis."""

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [
                SimpleNamespace(
                    content=(
                        "longmemeval_session_id=parking user: I got a parking ticket "
                        "near my work for $50."
                    ),
                    citation="eventloom://benchmark/events/parking#abc",
                    source="graph",
                    score=0.99,
                ),
                SimpleNamespace(
                    content=(
                        "longmemeval_session_id=wash user: I had a car wash on "
                        "February 3rd that cost $15."
                    ),
                    citation="eventloom://benchmark/events/wash#def",
                    source="graph",
                    score=0.98,
                ),
            ]

    class NoisyLexicalRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return [
                (
                    "citation=eventloom://benchmark/events/budget#aaa "
                    "longmemeval_session_id=budget user: I am budgeting my car expenses. "
                    "My annual car insurance premium is $3,500."
                ),
                (
                    "citation=eventloom://benchmark/events/wash#def "
                    "longmemeval_session_id=wash user: I had a car wash on February 3rd that cost $15."
                ),
                (
                    "citation=eventloom://benchmark/events/parking#abc "
                    "longmemeval_session_id=parking user: I got a parking ticket near my work for $50."
                ),
            ]

    retriever = ZaxyCheckoutRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=NoisyLexicalRetriever(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How much did I spend on car wash and parking ticket?",
        limit=5,
    )
    output = "\n".join(results)

    assert "checkout_answer_candidate=true" in output
    assert "answer=$65" in output
    assert "answer=$3,565" not in output


async def test_zaxy_checkout_retriever_combines_graph_and_lexical_quantity_operands() -> None:
    """Aggregate checkout synthesis should combine cited operands across graph and source lanes."""

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [
                SimpleNamespace(
                    content=(
                        "longmemeval_session_id=layer-feed user: I got a 50-pound batch "
                        "of layer feed from the farm supply store."
                    ),
                    citation="eventloom://benchmark/events/layer-feed#abc",
                    source="graph",
                    score=0.99,
                )
            ]

    class LexicalRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return [
                (
                    "citation=eventloom://benchmark/events/guideline#aaa "
                    "longmemeval_session_id=guideline assistant: Provide 1-1.5 pounds "
                    "of layer feed per hen per day as a general guideline."
                ),
                (
                    "citation=eventloom://benchmark/events/scratch#def "
                    "longmemeval_session_id=scratch user: I bought 20 pounds of organic "
                    "scratch grains for my chickens recently."
                ),
            ]

    retriever = ZaxyCheckoutRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=LexicalRetriever(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "What is the total weight of the new feed I purchased in the past two months?",
        limit=5,
    )
    output = "\n".join(results)

    assert "checkout_answer_candidate=true" in output
    assert "answer=70 pounds" in output
    assert "source_id=layer-feed" in output
    assert "source_id=scratch" in output
    assert "answer=71.5 pounds" not in output


async def test_zaxy_checkout_retriever_prefers_complete_quoted_duration_over_partial_absence() -> None:
    """Partial lexical absence should not preempt complete graph-backed duration synthesis."""
    expected = (
        "2 weeks for 'The Nightingale', 4 weeks for 'Sapiens: A Brief History of Humankind', "
        "and 2 weeks for 'The Power', so a total of 8 weeks."
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            rows = [
                (
                    "answer-1",
                    "2022/01/01",
                    "user: I started reading 'The Nightingale' by Kristin Hannah today.",
                ),
                (
                    "answer-2",
                    "2022/01/15",
                    "user: I just finished reading \"The Nightingale\" by Kristin Hannah today.",
                ),
                (
                    "answer-3",
                    "2022/02/01",
                    "user: I just started listening to 'Sapiens: A Brief History of Humankind' today.",
                ),
                (
                    "answer-4",
                    "2022/03/01",
                    "user: I just finished listening to 'Sapiens: A Brief History of Humankind' today.",
                ),
                (
                    "answer-5",
                    "2022/03/06",
                    "user: I started listening to \"The Power\" by Naomi Alderman today.",
                ),
                (
                    "answer-6",
                    "2022/03/20",
                    "user: I just finished listening to 'The Power' by Naomi Alderman today.",
                ),
            ]
            return [
                SimpleNamespace(
                    content=f"longmemeval_session_id={source_id} longmemeval_session_date={date} {text}",
                    citation=f"eventloom://benchmark/events/{source_id}#abc",
                    source="graph",
                    score=1.0,
                )
                for source_id, date, text in rows
            ]

    class PartialLexicalRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return [
                (
                    "citation=eventloom://benchmark/events/answer-3#abc "
                    "longmemeval_session_id=answer-3 longmemeval_session_date=2022/02/01 "
                    "user: I just started listening to 'Sapiens: A Brief History of Humankind' today."
                ),
                (
                    "citation=eventloom://benchmark/events/answer-4#abc "
                    "longmemeval_session_id=answer-4 longmemeval_session_date=2022/03/01 "
                    "user: I just finished listening to 'Sapiens: A Brief History of Humankind' today."
                ),
                (
                    "citation=eventloom://benchmark/events/answer-6#abc "
                    "longmemeval_session_id=answer-6 longmemeval_session_date=2022/03/20 "
                    "user: I just finished listening to 'The Power' by Naomi Alderman today."
                ),
            ]

    retriever = ZaxyCheckoutRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=PartialLexicalRetriever(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many weeks in total do I spent on reading 'The Nightingale' and listening "
        "to 'Sapiens: A Brief History of Humankind' and 'The Power'?",
        limit=5,
    )

    assert expected in "\n".join(results[:5])


async def test_zaxy_checkout_retriever_synthesizes_interval_across_graph_and_source_lanes() -> None:
    """Interval checkout synthesis should combine endpoint evidence across cited lanes."""

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [
                SimpleNamespace(
                    content=(
                        "longmemeval_session_id=house_1 longmemeval_session_date=2022/03/02 (Wed) "
                        "user: I started working with Rachel on February 15th."
                    ),
                    citation="file://longmemeval/house_1/salient-turn-0009.md:1",
                    source="graph",
                    score=1.0,
                )
            ]

    class FakeLexicalRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return [
                (
                    "citation=eventloom://benchmark/events/found#abc "
                    "longmemeval_session_id=house_2 longmemeval_session_date=2022/03/02 (Wed) "
                    "user: The house I saw on March 1st really checks all the boxes."
                ),
                (
                    "citation=eventloom://benchmark/events/drive#def "
                    "longmemeval_session_id=house_1 user: I want areas within a 30-minute drive."
                ),
            ]

    retriever = ZaxyCheckoutRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=FakeLexicalRetriever(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many days did it take for me to find a house I loved after starting to work with Rachel?",
        limit=5,
    )
    output = "\n".join(results)

    assert "checkout_answer_candidate=true" in output
    assert "answer=14 days. 15 days" in output
    assert "minute_total_hours=" not in output


async def test_zaxy_checkout_retriever_recovers_scoped_absence_support_without_temporal_filter() -> None:
    """Temporal absence checks should prove the inspected neighborhood, not return only the anchor."""

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return []

    class TemporalStarvedLexicalRetriever:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del limit
            if temporal_point is not None:
                return []
            if "NovaTech" not in query and "physical notebook" not in query:
                return [
                    (
                        "citation=eventloom://benchmark/events/distractor#ccc "
                        "longmemeval_salient_memory_turn=true "
                        "longmemeval_session_id=google-drive "
                        "I backed up my resume to Google Drive before starting a different project."
                    )
                ]
            return [
                (
                    "citation=eventloom://benchmark/events/1#aaa "
                    "longmemeval_session_id=career-total "
                    "I've been working professionally for 9 years and currently use a physical notebook."
                ),
                (
                    "citation=eventloom://benchmark/events/2#bbb "
                    "longmemeval_session_id=current-role "
                    "I've been working at NovaTech for about 4 years and 3 months now as a backend developer."
                ),
            ]

    retriever = ZaxyCheckoutRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=TemporalStarvedLexicalRetriever(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How long have I been working before I started my current job at Google?",
        temporal_point="2023-05-23T10:59:00Z",
        limit=10,
    )
    top_five = "\n".join(results[:5])
    output = "\n".join(results)

    assert "zaxy_absence_check=true" in output
    assert "You haven't started working at Google yet" in output
    assert "support_source_ids=" in output
    assert "career-total" in output
    assert "current-role" in output
    assert "career-total" in top_five
    assert "current-role" in top_five
    assert "query-temporal-anchor" not in "\n".join(results[:3])


async def test_zaxy_retriever_filters_stale_preference_lexical_backfill() -> None:
    """Raw lexical source backfill should not reintroduce superseded preference values."""
    class FakeRouter:
        async def query(
            self,
            query: str,
            *,
            temporal_point: str | None = None,
            limit: int = 10,
            embedding: list[float] | None = None,
        ) -> list[ContextChunk]:
            del query, temporal_point, limit, embedding
            return [
                ContextChunk(
                    content="user-0003:theme (preference) — summary=theme=theme-new-3",
                    source="exact",
                    score=1.0,
                    valid_from="2024-06-01T00:00:00Z",
                    valid_to=None,
                ),
                ContextChunk(
                    content=(
                        "docs/runbooks/service-0003.md:22-27 (document) — "
                        "summary=release marker doc-code-0003"
                    ),
                    source="keyword",
                    score=0.9,
                    valid_from="2024-08-01T00:00:00Z",
                    valid_to=None,
                ),
            ]

    class FakeProvider:
        dimension = 2

        def embed(self, text: str) -> list[float]:
            del text
            return [1.0, 0.0]

    class FakeLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return [
                (
                    "type=user.preference_changed actor=user payload={userId=user-0003, "
                    "key=theme, value=theme-old-3}"
                ),
                "type=transcript.turn content=We decided decision-code-0003 for workstream 0003.",
            ]

    retriever = ZaxyRetriever(  # type: ignore[arg-type]
        FakeRouter(),
        FakeProvider(),
        lexical_retriever=FakeLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "For user-0003 in workstream 0003, recover the current theme, "
        "runbook marker doc-code-0003, and session decision.",
        limit=10,
    )
    content = "\n".join(results)

    assert "theme=theme-new-3" in content
    assert "doc-code-0003" in content
    assert "decision-code-0003" in content
    assert "theme-old-3" not in content


def test_bm25_retriever_ranks_specific_identifier_above_generic_overlap() -> None:
    """BM25 should prefer rare identity terms over broad keyword overlap."""
    corpus = (
        BenchmarkChunk(
            "generic",
            "release cache context mentions rollback planning and deployment notes",
        ),
        BenchmarkChunk(
            "target",
            "release cache context records doc-code-0001 as the rollback owner source",
        ),
    )

    results = BM25Retriever(corpus).query("cache rollback doc-code-0001", limit=1)

    assert results == [corpus[1].text]


def test_rank_fusion_retriever_preserves_complementary_hits() -> None:
    """Fusion should keep lexical hits when another retriever misses them."""
    corpus = (
        BenchmarkChunk("answer", "longmemeval_session_id=answer-1 degree Business Administration"),
        BenchmarkChunk("distractor", "degree graph-shaped context about unrelated project planning"),
    )

    fused = RankFusionRetriever(
        {
            "graph": MarkdownRetriever((corpus[1],)),
            "bm25": BM25Retriever(corpus),
        }
    )

    results = fused.query("What degree did I graduate with?", limit=2)

    assert any("answer-1" in result for result in results)
    assert len(results) == 2


async def test_zaxy_retriever_can_fuse_graph_and_source_lexical_results() -> None:
    """Source/provenance benchmark retrieval should support lexical fusion."""
    corpus = (
        BenchmarkChunk(
            "answer",
            "citation=eventloom://benchmark/events/1#abc "
            "source-recall/target/service-0001.md records source-answer-0001",
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph context about source-answer-0001")]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("Which cited source records source-answer-0001?", limit=2)

    assert any("graph context" in result for result in results)
    assert any("eventloom://benchmark/events/1#abc" in result for result in results)


async def test_zaxy_retriever_fuses_verbatim_personal_memory_results() -> None:
    """Personal-memory questions should be able to recover exact Eventloom text."""
    corpus = (
        BenchmarkChunk(
            "answer",
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=answer-1 my cat's name is Luna",
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph context about pet care")]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("What is the name of my cat?", limit=2)

    assert any("Luna" in result for result in results)
    assert any("graph context" in result for result in results)


async def test_zaxy_retriever_ranks_personal_source_lane_before_graph_context() -> None:
    """Personal memory checkout should surface cited source text in top slots."""
    corpus = (
        BenchmarkChunk(
            "answer",
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 my cat's name is Luna"
            ),
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("What is the name of my cat?", limit=5)

    assert "longmemeval_session_id=answer-1" in results[0]


async def test_zaxy_retriever_reserves_verbatim_lane_when_graph_crowds_results() -> None:
    """Top verbatim hits should survive even when graph returns many candidates."""
    corpus = (
        BenchmarkChunk(
            "answer",
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=answer-1 my cat's name is Luna",
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("What is the name of my cat?", limit=10)

    assert len(results) == 10
    assert any("Luna" in result for result in results)


def test_personal_memory_intent_reserves_multiple_source_slots() -> None:
    """Ambiguous personal-memory questions should preserve competing source evidence."""
    intent = classify_retrieval_intent("What breed is my dog?", limit=10)

    assert intent.needs_source_lane
    assert intent.source_lane_slots == 3
    assert "personal_memory" in intent.reasons


async def test_zaxy_retriever_reserves_multiple_personal_memory_sources() -> None:
    """Personal-memory retrieval should keep enough source identities to disambiguate."""
    source_contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=distractor-1 my dog requires regular grooming."
        ),
        (
            "citation=eventloom://benchmark/events/2#abc "
            "longmemeval_session_id=distractor-2 my dog enjoys trail walks."
        ),
        (
            "citation=eventloom://benchmark/events/3#abc "
            "longmemeval_session_id=answer-3 my dog is a Golden Retriever."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class OrderedLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=OrderedLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async("What breed is my dog?", limit=10)

    assert sum("longmemeval_session_id=" in result for result in results) == 3
    assert any("longmemeval_session_id=answer-3" in result for result in results)


def test_source_lane_query_uses_graph_answer_concepts_for_source_recovery() -> None:
    """Graph answer concepts should help raw source lookup recover citations."""
    query = source_lane_query(
        "What breed is my dog?",
        [
            (
                "longmemeval/75499fd8/52c34859_1/chunk-0001.md "
                "(document) — summary=Max is a Golden Retriever"
            )
        ],
    )

    assert query == "What breed is my dog? Max Golden Retriever"


def test_source_lane_query_ignores_date_header_noise() -> None:
    """Capitalized provenance/date words should not pollute source backfill."""
    query = source_lane_query(
        "What play did I attend at the local community theater?",
        [
            (
                "longmemeval/58bf7951/answer_355c48bb/chunk-0002.md "
                "(document) — source_thread=default, summary=The Glass Menagerie"
            )
        ],
    )

    assert query == (
        "What play did I attend at the local community theater? "
        "The Glass Menagerie"
    )


def test_source_lane_queries_expand_aggregation_event_actions() -> None:
    """Aggregation source lookup should search likely memory phrasings, not just query wording."""
    queries = source_lane_queries(
        "How many model kits have I worked on or bought?",
        [],
    )

    assert queries == (
        "How many model kits have I worked on or bought?",
        "model kits finished started picked up got bought scale",
    )


def test_source_lane_queries_expand_social_media_break_durations() -> None:
    """Social-media break totals should search break-length phrasings."""
    queries = source_lane_queries(
        "How many days did I take social media breaks in total?",
        [],
    )

    assert queries == (
        "How many days did I take social media breaks in total?",
        "social media break week-long 10-day mid-January mid-February",
    )


def test_source_lane_queries_expand_church_service_date_intervals() -> None:
    """Church-service date interval queries should search the dated service phrasing."""
    queries = source_lane_queries(
        "How many days had passed between the Sunday mass at St. Mary's Church and the Ash Wednesday service at the cathedral?",
        [],
    )

    assert queries == (
        "How many days had passed between the Sunday mass at St. Mary's Church and the Ash Wednesday service at the cathedral?",
        "Sunday mass St. Mary's Church January 2 Ash Wednesday service cathedral February 1",
    )


def test_source_lane_queries_expand_road_trip_drive_durations() -> None:
    """Road-trip driving totals need destination memories instead of route-planning detours."""
    queries = source_lane_queries(
        "How many hours in total did I spend driving to my three road trip destinations combined?",
        [],
    )

    assert queries == (
        "How many hours in total did I spend driving to my three road trip destinations combined?",
        "road trip drove driving hours destinations Outer Banks Washington D.C. Tennessee mountains",
    )


def test_source_lane_queries_expand_wedding_attendance_memories() -> None:
    """Wedding counts should search attended-wedding couple and venue clues."""
    queries = source_lane_queries(
        "How many weddings have I attended in this year?",
        [],
    )

    assert queries == (
        "How many weddings have I attended in this year?",
        "weddings attended got back cousin Rachel vineyard Emily Sarah rooftop Jen Tom rustic barn",
    )


def test_source_lane_queries_expand_grocery_store_spend_memories() -> None:
    """Grocery max-spend questions should search store names and spend amounts."""
    queries = source_lane_queries(
        "Which grocery store did I spend the most money at in the past month?",
        [],
    )

    assert queries == (
        "Which grocery store did I spend the most money at in the past month?",
        "grocery shopping spent money store Trader Joe's Walmart Thrive Market Publix Instacart $80 $120 $150 $60",
    )


def test_source_lane_queries_expand_family_age_average_memories() -> None:
    """Family-age averages should search self, parent, and grandparent age memories."""
    queries = source_lane_queries(
        "What is the average age of me, my parents, and my grandparents?",
        [],
    )

    assert queries == (
        "What is the average age of me, my parents, and my grandparents?",
        "average age me parents grandparents turned 32 mom 55 dad 58 grandma 75 grandpa 78",
    )


def test_source_lane_queries_expand_birth_age_arithmetic_memories() -> None:
    """Birth-age subtraction needs both target-age and current-age memories."""
    queries = source_lane_queries(
        "How old was I when Alex was born?",
        [],
    )

    assert queries == (
        "How old was I when Alex was born?",
        "Alex age years old just current age my age I just turned born birthday",
    )


def test_source_lane_queries_expand_charity_event_fundraising_memories() -> None:
    """Charity fundraising totals should search event names and raised amounts."""
    queries = source_lane_queries(
        "How much money did I raise in total through all the charity events I participated in?",
        [],
    )

    assert queries == (
        "How much money did I raise in total through all the charity events I participated in?",
        "charity events participated raised total charity walk $250 Bike-a-Thon Cancer Research $5,000 charity yoga $600 animal shelter",
    )


def test_source_synthesis_bundle_extracts_assistant_ordinal_recall() -> None:
    """Previous-chat ordinal questions should produce a cited assistant answer candidate."""
    bundle = source_synthesis_bundle(
        query="Can you remind me what was the 7th job in the list you provided?",
        source_results=[
            "\n".join(
                [
                    "citation=eventloom://benchmark/events/1#abc session_id=answer-list",
                    "assistant:",
                    "1. Virtual customer service representative",
                    "2. Telehealth professional",
                    "3. Remote bookkeeper",
                    "4. Virtual tutor or teacher",
                    "5. Freelance writer or editor",
                    "6. Online survey taker",
                    "7. Transcriptionist",
                    "8. Social media manager",
                ]
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=Transcriptionist" in bundle
    assert "candidate_type=assistant_recall" in bundle


def test_source_synthesis_bundle_ignores_transcript_turn_numbers_for_assistant_ordinal() -> None:
    """Ordinal recall should count the assistant list, not transcript turn numbers."""
    bundle = source_synthesis_bundle(
        query=(
            "I think we discussed work from home jobs for seniors earlier. "
            "Can you remind me what was the 7th job in the list you provided?"
        ),
        source_results=[
            " ".join(
                [
                    "# Event 1 citation=eventloom://benchmark/events/1#abc",
                    "document.indexed longmemeval content=longmemeval_session_id=answer-list",
                    "1. user: Brainstorm ideas for work from home jobs for seniors",
                    "2. assistant: 1. Virtual customer service representative",
                    "2. Telehealth professional",
                    "3. Remote bookkeeper",
                    "4. Virtual tutor or teacher",
                    "5. Freelance writer or editor",
                    "6. Online survey taker",
                    "7. Transcriptionist",
                    "8. Social media manager",
                ]
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=Transcriptionist" in bundle
    assert "assistant_recall_answer=Freelance writer or editor" not in bundle


def test_source_synthesis_bundle_extracts_assistant_subject_count_recall() -> None:
    """Study subject-count recall should extract the count tied to the requested journal."""
    bundle = source_synthesis_bundle(
        query=(
            "Can you remind me how many subjects were in the study published "
            "in the journal Music and Medicine?"
        ),
        source_results=[
            (
                "role=assistant Another study published in the journal Music and Medicine "
                "involved 38 subjects who listened to binaural beats for 30 minutes daily. "
                "The study found significant reductions in symptoms of depression, anxiety, and stress."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=38 subjects" in bundle


def test_source_synthesis_bundle_extracts_assistant_schedule_recall() -> None:
    """Assistant recall should extract a named day-specific shift assignment."""
    bundle = source_synthesis_bundle(
        query=(
            "I'm checking our previous chat about the shift rotation sheet for GM social media agents. "
            "Can you remind me what was the rotation for Admon on a Sunday?"
        ),
        source_results=[
            "\n".join(
                [
                    "role=assistant Shift Rotation Sheet for GM Social Media Agents",
                    "| | 8 am - 4 pm (Day Shift) | 12 pm - 8 pm (Afternoon Shift) |",
                    "| --- | --- | --- |",
                    "| Sunday | Admon | Magdy |",
                    "| Monday | Mostafa | Nemr |",
                ]
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=Admon was assigned to the 8 am - 4 pm (Day Shift) on Sundays." in bundle


def test_source_synthesis_bundle_extracts_assistant_marketing_budget_recall() -> None:
    """Assistant recall should extract a named campaign channel budget."""
    bundle = source_synthesis_bundle(
        query=(
            "I'm looking back at our previous chat about the DHL Wellness Retreats campaign. "
            "Can you remind me how much was allocated for influencer marketing in the campaign plan?"
        ),
        source_results=[
            (
                "role=assistant Influencer Marketing Campaign Plan for DHL Wellness Retreats. "
                "Budget: * Influencer marketing: $2,000 Timeline: May 1st - May 31st."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=$2,000" in bundle


def test_source_synthesis_bundle_extracts_assistant_gin_bottle_recall() -> None:
    """Assistant recall should select the fifth item from the gin-based bottle list."""
    bundle = source_synthesis_bundle(
        query=(
            "I'm looking back at our previous conversation about building a cocktail bar. "
            "You recommended five bottles to make the widest variety of gin-based cocktails. "
            "Can you remind me what the fifth bottle was?"
        ),
        source_results=[
            (
                "role=assistant To make the widest variety of gin-based cocktails, I would recommend: "
                "1. Sweet Vermouth 2. Dry Vermouth 3. Campari 4. Elderflower Liqueur 5. Absinthe"
            ),
            "role=assistant For general cocktails: 1. Gin 2. Vodka 3. Rum 4. Whiskey 5. Triple Sec",
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=Absinthe." in bundle
    assert "assistant_recall_answer=Triple Sec" not in bundle


def test_source_synthesis_bundle_extracts_assistant_video_link_recall() -> None:
    """Assistant recall should preserve a recommended video title and URL."""
    bundle = source_synthesis_bundle(
        query=(
            "I wanted to follow up on our previous conversation about YouTube videos "
            "for workplace posture. Can you remind me of the Mayo Clinic video you recommended?"
        ),
        source_results=[
            (
                "role=assistant I recommend Mayo Clinic's video "
                "'How to Sit Properly at a Desk to Avoid Back Pain'. "
                "Link: https://www.youtube.com/watch?v=UfOvNlX9Hh0"
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=The video is 'How to Sit Properly at a Desk to Avoid Back Pain'" in bundle
    assert "https://www.youtube.com/watch?v=UfOvNlX9Hh0" in bundle


def test_source_synthesis_bundle_extracts_assistant_ratio_recall() -> None:
    """Assistant recall should extract ratio instructions as an answer-ready sentence."""
    bundle = source_synthesis_bundle(
        query=(
            "I remember you told me to dilute tea tree oil with a carrier oil before "
            "applying it to my skin. Can you remind me what the recommended ratio is?"
        ),
        source_results=[
            (
                "role=assistant Use a 1:10 ratio with carrier oil. "
                "That means one part tea tree oil to ten parts carrier oil."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert (
        "assistant_recall_answer=The recommended ratio is 1:10, "
        "meaning one part tea tree oil to ten parts carrier oil."
    ) in bundle


def test_source_synthesis_bundle_extracts_borges_library_recall() -> None:
    """Assistant recall should extract the exact Borges center/circumference sentence."""
    bundle = source_synthesis_bundle(
        query=(
            "I was going through our previous conversation about The Library of Babel, "
            "and I wanted to confirm - what did Borges say about the center and circumference of the Library?"
        ),
        source_results=[
            (
                "role=assistant Borges wrote that The Library is a sphere whose exact center "
                "is any one of its hexagons and whose circumference is inaccessible."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=According to Borges" in bundle
    assert "circumference is inaccessible" in bundle


def test_source_synthesis_bundle_extracts_assistant_website_recall() -> None:
    """Assistant recall should extract a named website from recommended resources."""
    bundle = source_synthesis_bundle(
        query=(
            "I wanted to follow up on our previous conversation about mindfulness techniques. "
            "You mentioned some great resources for guided imagery exercises, can you remind me "
            "of the website that had free exercises like 'The Mountain Meditation' and "
            "'The Body Scan Meditation'?"
        ),
        source_results=[
            (
                "role=assistant Mindful.org has free guided imagery exercises, "
                "including The Mountain Meditation and The Body Scan Meditation."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=Mindful.org." in bundle


def test_source_synthesis_bundle_extracts_assistant_company_pair_recall() -> None:
    """Assistant recall should extract paired company examples."""
    bundle = source_synthesis_bundle(
        query=(
            "I was going through our previous conversation and I was wondering if you could "
            "remind me of the two companies you mentioned that prioritize employee safety "
            "and well-being like Triumvirate?"
        ),
        source_results=[
            (
                "role=assistant Two companies that prioritize employee safety and well-being "
                "like Triumvirate are Patagonia and Southwest Airlines."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=Patagonia and Southwest Airlines." in bundle


def test_source_synthesis_bundle_extracts_escaped_siac_tool_recall() -> None:
    """Assistant tool recall should tolerate escaped underscores from markdown sources."""
    bundle = source_synthesis_bundle(
        query="Which atmospheric correction algorithm is implemented in the SIAC_GEE tool?",
        source_results=[
            (
                "role=assistant SIAC\\_GEE is a JavaScript code that runs on Google Earth Engine. "
                "It uses the 6S algorithm for atmospheric correction."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=The 6S algorithm is implemented in the SIAC_GEE tool." in bundle


def test_source_synthesis_bundle_prefers_last_venue_over_artist_list() -> None:
    """Last-item venue recall should not choose an unrelated artist list."""
    bundle = source_synthesis_bundle(
        query="Can you remind me of the name of the last venue you recommended in Portland?",
        source_results=[
            (
                "role=assistant Popular indie artists in Portland: "
                "1. The Decemberists 2. Horse Feathers"
            ),
            (
                "role=assistant Here are some popular venues in Portland for indie music shows: "
                "1. Mississippi Studios 2. Doug Fir Lounge 3. Revolution Hall"
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=Revolution Hall." in bundle
    assert "assistant_recall_answer=Horse Feathers" not in bundle


def test_source_synthesis_bundle_extracts_construction_year_recall() -> None:
    """Direct year recall should extract the cited construction-start year."""
    bundle = source_synthesis_bundle(
        query="Can you remind me what year the construction of the house began?",
        source_results=[
            (
                "role=user Bajimaya v Reward Homes Pty Ltd case summary. "
                "The construction of the house began in 2014, and the contract was signed in 2015."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=2014." in bundle


def test_source_synthesis_bundle_extracts_assistant_objectives_recall() -> None:
    """Assistant objectives should be projected as a compact answer surface."""
    bundle = source_synthesis_bundle(
        query="Can you remind me what were the three objectives we outlined for the project?",
        source_results=[
            "\n".join(
                [
                    "citation=eventloom://benchmark/events/1#abc session_id=answer-objectives",
                    "Objectives:",
                    "1. To identify molecular subtypes of endometrial cancer using genomic approaches.",
                    "2. To investigate their clinical and biological significance.",
                    "3. To develop biomarkers for early detection and prognosis.",
                ]
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=The three objectives were:" in bundle
    assert "to identify molecular subtypes of endometrial cancer" in bundle
    assert "to develop biomarkers for early detection and prognosis" in bundle


def test_source_synthesis_bundle_extracts_assistant_options_recall() -> None:
    """Assistant option-list recall should preserve the listed alternatives."""
    bundle = source_synthesis_bundle(
        query="Can you remind me what the other four options were?",
        source_results=[
            "\n".join(
                [
                    "citation=eventloom://benchmark/events/1#abc session_id=answer-options",
                    "Here are some other alternatives:",
                    "1. Sexual fixations - strong preoccupation with sexual thoughts.",
                    "2. Problematic sexual behaviors - behaviors that cause difficulties.",
                    "3. Sexual impulsivity - impulsive nature of certain sexual behaviors.",
                    "4. Compulsive sexuality - compulsive nature of certain sexual behaviors.",
                ]
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "assistant_recall_answer=I suggested 'Sexual fixations'" in bundle
    assert "'Problematic sexual behaviors'" in bundle
    assert "'Compulsive sexuality'" in bundle


def test_source_lane_queries_expand_cuisine_learning_memories() -> None:
    """Cuisine counts should search the specific cooking and tried-cuisine memory phrasings."""
    queries = source_lane_queries(
        "How many different cuisines have I learned to cook or tried out in the past few months?",
        [],
    )

    assert queries == (
        "How many different cuisines have I learned to cook or tried out in the past few months?",
        "cuisines learned cook tried Ethiopian Indian Mexican Thai meal prep tikka masala tacos pad thai",
    )


def test_source_lane_queries_expand_property_viewing_outcomes() -> None:
    """Property-viewing counts should search rejected/outcome clues, not just generic property words."""
    queries = source_lane_queries(
        "How many properties did I view before making an offer on the townhouse in the Brookside neighborhood?",
        [],
    )

    assert queries == (
        "How many properties did I view before making an offer on the townhouse in the Brookside neighborhood?",
        "property house bungalow condo townhouse viewed toured saw offer",
        "Brookside townhouse Oakwood bungalow Cedar Creek 1-bedroom condo 2-bedroom condo kitchen renovation budget highway higher bid",
    )


def test_source_lane_queries_expand_accommodation_price_comparisons() -> None:
    """Accommodation price comparisons should search destination aliases and nightly prices."""
    queries = source_lane_queries(
        "How much more did I spend on accommodations per night in Hawaii compared to Tokyo?",
        [],
    )

    assert queries == (
        "How much more did I spend on accommodations per night in Hawaii compared to Tokyo?",
        "Hawaii Maui Tokyo accommodations per night hostel resort hotel costs $300 $30 luxurious affordable",
    )


def test_source_lane_queries_expand_meeting_temporal_order_evidence() -> None:
    """Meeting-order questions should search named people plus relative-time phrasings."""
    queries = source_lane_queries(
        "Who did I meet first, Mark and Sarah or Tom?",
        [],
    )

    assert queries == (
        "Who did I meet first, Mark and Sarah or Tom?",
        "met first Mark Sarah Tom few months ago about a month ago beach trip charity event",
    )


def test_source_lane_queries_expand_instrument_ownership_actions() -> None:
    """Instrument aggregation should search owned instrument brands and families."""
    queries = source_lane_queries(
        "How many musical instruments do I currently own?",
        [],
    )

    assert queries == (
        "How many musical instruments do I currently own?",
        "musical instruments guitar piano drum set acoustic electric korg yamaha fender pearl owned had playing",
    )


def test_source_lane_queries_expand_longitudinal_event_vocabulary() -> None:
    """Source lookup should bridge query wording to how personal events are recorded."""
    cases = {
        "Who became a parent first, Rachel or Alex?": (
            "parent rachel alex adopted born twins cousin baby girl china february january"
        ),
        "How many projects have I led or am currently leading?": (
            "project led leading currently managed team initiative launched rollout migration"
        ),
        "How much total money have I spent on bike-related expenses since the start of the year?": (
            "bike bicycle helmet chain lights rack tune-up cost bought replaced installed"
        ),
        "How many times did I bake something in the past two weeks?": (
            "bake baked baking bread cookies sourdough muffins recipe"
        ),
        "How many fish are there in total in both of my aquariums?": (
            "fish aquarium aquariums tank tanks betta gourami tetras bubbles upgraded"
        ),
        "How many tanks do I currently have, including the one I set up for my friend's kid?": (
            "fish aquarium aquariums tank tanks betta gourami tetras bubbles upgraded"
        ),
        "How many plants did I acquire in the last month?": (
            "plants plant acquired bought received monstera pothos snake plant succulent nursery last month"
        ),
        "How many different art-related events did I attend in the past month?": (
            "art exhibition gallery museum festival studio attended event events past month "
            "Children's Museum History Museum Art Gallery Art Afternoon guided tour lecture street art"
        ),
        "How many fitness classes do I attend in a typical week?": (
            "fitness classes yoga pilates spin boxing barre Zumba BodyPump Hip Hop Abs "
            "typical week attend schedule Mondays Tuesdays Thursdays Saturday Sunday"
        ),
        "How long have I been working before I started my current job at Google?": (
            "google current job years months career working professionally field started"
        ),
    }

    for query, expected_expansion in cases.items():
        assert expected_expansion in source_lane_queries(query, [])


async def test_zaxy_retriever_uses_graph_answer_concepts_for_source_lane() -> None:
    """Source backfill should recover provenance for graph-discovered answer concepts."""
    source_contexts = [
        "longmemeval_session_id=distractor-1 my dog requires regular grooming.",
        "longmemeval_session_id=distractor-2 my dog enjoys trail walks.",
        "longmemeval_session_id=answer-3 Max is a Golden Retriever.",
    ]
    seen_queries: list[str] = []

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph says Max is a Golden Retriever")]

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            seen_queries.append(query)
            if "Golden Retriever" not in query:
                return source_contexts[:2]
            return source_contexts[:limit]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async("What breed is my dog?", limit=10)

    assert seen_queries[0] == "What breed is my dog?"
    assert any("Golden Retriever" in query for query in seen_queries[1:])
    assert any("longmemeval_session_id=answer-3" in result for result in results)


async def test_zaxy_retriever_merges_expanded_source_queries_before_truncation() -> None:
    """Expanded aggregation recall should not be starved by the original query results."""
    original_contexts = [
        f"longmemeval_session_id=distractor-{index} model kit discussion {index}"
        for index in range(1, 9)
    ]
    expanded_contexts = [
        "longmemeval_session_id=answer-1 I finished a simple Revell F-15 Eagle kit.",
        "longmemeval_session_id=answer-2 I started a Tamiya 1/48 scale Spitfire Mk.V.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph distractor")]

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            if "finished started picked up" in query:
                return expanded_contexts[:limit]
            return original_contexts[:limit]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many model kits have I worked on or bought?",
        limit=5,
    )

    assert any("answer-1" in result for result in results)
    assert any("answer-2" in result for result in results)


async def test_zaxy_retriever_merges_expanded_aggregation_queries_even_with_salient_primary_hits() -> None:
    """Salient source hits are not enough to close evidence coverage for aggregation."""
    original_contexts = [
        (
            "longmemeval_salient_memory_turn=true "
            f"longmemeval_session_id=distractor-{index} model kit planning discussion {index}"
        )
        for index in range(1, 9)
    ]
    expanded_contexts = [
        "longmemeval_session_id=answer-1 I finished a simple Revell F-15 Eagle kit.",
        "longmemeval_session_id=answer-2 I started a Tamiya 1/48 scale Spitfire Mk.V.",
    ]
    seen_queries: list[str] = []

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph distractor")]

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            seen_queries.append(query)
            if "finished started picked up" in query:
                return expanded_contexts[:limit]
            return original_contexts[:limit]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many model kits have I worked on or bought?",
        limit=5,
    )

    assert any("finished started picked up" in query for query in seen_queries)
    assert any("answer-1" in result for result in results)
    assert any("answer-2" in result for result in results)


async def test_zaxy_retriever_uses_graph_evidence_for_temporal_synthesis() -> None:
    """Synthesis should use graph-retrieved cited turns as well as lexical backfill."""
    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [
                SimpleNamespace(
                    content=(
                        "longmemeval_session_id=answer-1 "
                        "I got my new binoculars exactly three weeks ago."
                    )
                )
            ]

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point
            return [
                (
                    "longmemeval_session_id=answer-2 I saw the American goldfinches "
                    "returning to the area a week ago."
                ),
                (
                    "longmemeval_session_id=answer-1 I waited months for the "
                    "binoculars to arrive."
                ),
            ][:limit]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How long did I use my new binoculars before I saw the American goldfinches returning to the area?",
        limit=5,
    )

    assert "week_interval=2 weeks" in results[0]
    assert "week_interval_answer=Two weeks" in results[0]


def test_absence_check_ignores_direct_fact_question_words() -> None:
    """Absence guidance should not fire when direct fact terms are present in evidence."""
    bundle = absence_check_bundle(
        query="What brand are my favorite running shoes?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer Nike has been my favourite brand "
                "so far for running shoes."
            )
        ],
        limit=5,
    )

    assert bundle is None


def test_source_synthesis_bundle_requires_typed_evidence_for_aggregation() -> None:
    """Unsupported aggregation should fall through to absence guidance instead of hallucinating."""
    contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=answer I spent two weeks traveling solo around Japan."
        )
    ]

    bundle = source_synthesis_bundle(
        query="How long was I in Korea for?",
        source_results=contexts,
        limit=5,
    )

    assert bundle is None
    absence = absence_check_bundle(
        query="How long was I in Korea for?",
        source_results=contexts,
        limit=5,
    )
    assert absence is not None
    assert "not_mentioned_candidate=korea" in absence
    assert "Japan" in absence


def test_absence_check_bundle_ignores_assistant_only_destination_mentions() -> None:
    """Assistant travel recommendations should not count as the user's stay duration evidence."""
    contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=answer-1 "
            "user: I loved staying in Japan during my previous trip."
        ),
        (
            "citation=eventloom://benchmark/events/2#abc "
            "longmemeval_session_id=answer-2 "
            "assistant: Seoul is an amazing destination in South Korea for your itinerary."
        ),
    ]

    absence = absence_check_bundle(
        query="How long was I in Korea for?",
        source_results=contexts,
        limit=5,
    )

    assert absence is not None
    assert "not_mentioned_candidate=korea" in absence
    assert "Japan" in absence


def test_absence_check_bundle_handles_missing_title_reading_progress() -> None:
    """Generic reading-goal page counts should not answer title-specific pages-left questions."""
    absence = absence_check_bundle(
        query="How many pages do I have left to read in 'Sapiens'?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: I need to read pages each day for my book goal."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer-2 "
                "assistant: Based on your Sapiens routine, you read a few pages per day."
            ),
        ],
        limit=5,
    )

    assert absence is not None
    assert "not_mentioned_candidate=pages left to read in Sapiens" in absence
    assert "did not mention pages left to read in Sapiens" in absence


def test_absence_check_bundle_handles_missing_aggregation_target() -> None:
    """Aggregation queries should emit cited absence when the target item is missing."""
    absence = absence_check_bundle(
        query="How many Italian restaurants have I tried in my city?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: I tried several Korean restaurants in my city this month."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer-2 "
                "assistant: Korean restaurants can be a great way to explore local food."
            ),
        ],
        limit=5,
    )

    assert absence is not None
    assert "zaxy_absence_check=true" in absence
    assert "not_mentioned_candidate=italian" in absence
    assert "known_related_evidence=" in absence
    assert "korean" in absence.casefold()


def test_absence_check_bundle_handles_missing_direct_slot_target() -> None:
    """Direct personal-memory slot queries should answer absence from nearby cited evidence."""
    absence = absence_check_bundle(
        query="What is the name of my hamster?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: My cat Luna has been shedding on the sofa."
            )
        ],
        limit=5,
    )

    assert absence is not None
    assert "zaxy_absence_check=true" in absence
    assert "not_mentioned_candidate=hamster" in absence
    assert "known_related_evidence=cat luna" in absence.casefold()


def test_source_synthesis_bundle_defers_to_absence_when_comparison_target_is_missing() -> None:
    """Alternative questions should not synthesize an answer when one side lacks user evidence."""
    contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=answer-1 "
            "user: I just fixed the broken fence on the east side of my property three weeks ago."
        ),
        (
            "citation=eventloom://benchmark/events/2#abc "
            "longmemeval_session_id=answer-2 "
            "assistant: Peter can advise you about dairy cows, but no purchase was mentioned."
        ),
    ]

    bundle = source_synthesis_bundle(
        query="Which task did I complete first, fixing the fence or purchasing three cows from Peter?",
        source_results=contexts,
        limit=5,
    )
    absence = absence_check_bundle(
        query="Which task did I complete first, fixing the fence or purchasing three cows from Peter?",
        source_results=contexts,
        limit=5,
    )

    assert bundle is None
    assert absence is not None
    assert "not_mentioned_candidate=purchasing three cows peter" in absence
    assert "fixing the fence" in absence


def test_source_synthesis_bundle_defers_to_absence_when_temporal_anchor_is_missing() -> None:
    """Temporal calculations should not proceed when an explicit event anchor is absent."""
    contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=answer-1 "
            "user: I attended the annual Holiday Market at the local mall a week before Black Friday."
        ),
        (
            "citation=eventloom://benchmark/events/2#abc "
            "longmemeval_session_id=answer-2 "
            "user: I got the iPhone 13 Pro for my sister's birthday."
        ),
    ]

    bundle = source_synthesis_bundle(
        query="How many days before I bought my iPad did I attend the Holiday Market?",
        source_results=contexts,
        limit=5,
    )
    absence = absence_check_bundle(
        query="How many days before I bought my iPad did I attend the Holiday Market?",
        source_results=contexts,
        limit=5,
    )

    assert bundle is None
    assert absence is not None
    assert "not_mentioned_candidate=bought ipad" in absence
    assert "holiday market" in absence.casefold()


def test_source_evidence_score_detects_single_property_evidence() -> None:
    """Source ordering should prefer single cited evidence rows before aggregation is complete."""
    query = "How many properties did I view before making an offer on the townhouse in Brookside?"
    evidence = (
        "longmemeval_session_id=answer-1 "
        "user: I viewed a 1-bedroom condo on February 10th, "
        "but the noise from the highway was a deal-breaker."
    )
    topical_non_evidence = (
        "longmemeval_session_id=answer-2 "
        "assistant: Research helps you understand condo fees, taxes, insurance, "
        "and other expenses before buying a property."
    )

    assert source_evidence_score(query, evidence) > source_evidence_score(
        query,
        topical_non_evidence,
    )


def test_absence_check_bundle_preserves_model_identifier_targets() -> None:
    """Missing alternative targets should keep numeric and single-letter model identifiers."""
    absence = absence_check_bundle(
        query="Which project did I start first, the Ferrari model or the Porsche 991 Turbo S model?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: I started the Ferrari F40 model kit last month."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer-2 "
                "assistant: The Porsche 911 Turbo is an interesting sports car, but no project was mentioned."
            ),
        ],
        limit=5,
    )

    assert absence is not None
    assert "not_mentioned_candidate=porsche 991 turbo s model" in absence


def test_source_synthesis_bundle_projects_last_week_relative_interval() -> None:
    """Relative-time synthesis should derive elapsed intervals instead of summing anchors."""
    bundle = source_synthesis_bundle(
        query="How long had I been a member of Book Lovers Unite when I attended the meetup?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=joined "
                "user: I joined Book Lovers Unite three weeks ago."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=meetup "
                "user: I attended a meetup organized by Book Lovers Unite last week."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "week_values=" in bundle
    assert "week_interval_answer=Two weeks" in bundle


def test_source_synthesis_bundle_preserves_cross_source_relative_week_evidence() -> None:
    """Relative-time filtering should keep both event anchors across sources."""
    bundle = source_synthesis_bundle(
        query="How long did I use my new binoculars before I saw the American goldfinches returning to the area?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=goldfinches "
                "user: I've been listening to bird calls online for about a month. "
                "I noticed the American goldfinches returning to the area a week ago."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=goldfinches "
                "user: The American goldfinches returning to the area a week ago "
                "made bird identification practice more exciting."
            ),
            (
                "citation=eventloom://benchmark/events/3#abc "
                "longmemeval_session_id=binoculars "
                "user: The binoculars arrived exactly three weeks ago."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "week_interval_answer=Two weeks" in bundle


def test_source_synthesis_bundle_keeps_temporal_synthesis_for_generic_missing_words() -> None:
    """Generic query words should not trigger absence routing when temporal evidence is present."""
    bundle = source_synthesis_bundle(
        query="How many days before Rachel's party did I find the house I loved?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: On 2026/03/01 I found the perfect house for the party."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer-2 "
                "user: Rachel's party happened on 2026/03/15."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "zaxy_absence_check=true" not in bundle
    assert "date_interval_answer=14 days. 15 days (including the last day) is also acceptable." in bundle


def test_absence_check_bundle_uses_missing_concrete_action_target() -> None:
    """Direct absence questions should cite nearby memories without inventing absent actions."""
    absence = absence_check_bundle(
        query="How many days before I bought my iPad did I visit Seattle?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: I visited Seattle on 2026/02/01 and had dinner downtown."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer-2 "
                "assistant: I can help compare tablet models, but no purchase was mentioned."
            ),
        ],
        limit=5,
    )

    assert absence is not None
    assert "not_mentioned_candidate=bought ipad" in absence
    assert "visit seattle" in absence


def test_absence_check_bundle_does_not_route_generic_count_gaps() -> None:
    """Count/list synthesis should not become absence just because generic query words are absent."""
    absence = absence_check_bundle(
        query="How many musical instruments do I currently own?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: I've been playing my Fender Stratocaster electric guitar for years."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer-2 "
                "user: I'm thinking of selling my old Pearl Export drum set."
            ),
        ],
        limit=5,
    )

    assert absence is None


def test_source_synthesis_bundle_keeps_model_kit_count_evidence_out_of_absence() -> None:
    """Countable aggregation evidence should not be replaced by absence guidance."""
    bundle = source_synthesis_bundle(
        query="How many model kits have I worked on or bought?",
        source_results=[
            "session_id=answer-1 I recently finished a simple Revell F-15 Eagle kit.",
            "session_id=answer-2 I finished a Tamiya 1/48 scale Spitfire Mk.V.",
            "session_id=answer-3 I started working on a 1/16 scale German Tiger I tank.",
            "session_id=answer-4 I just got this kit and a 1/24 scale '69 Camaro at a model show.",
        ],
        limit=5,
    )

    assert bundle is not None
    assert "zaxy_synthesis_bundle=true" in bundle
    assert "zaxy_absence_check=true" not in bundle
    assert "count_answer=4" in bundle


def test_absence_check_bundle_does_not_route_missing_generic_temporal_words() -> None:
    """Temporal source recall needs more evidence, not a false absence answer."""
    absence = absence_check_bundle(
        query="Which streaming service did I start using most recently?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "user: I started using Hulu a few months ago."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer-2 "
                "user: I started using Apple TV+ last month."
            ),
        ],
        limit=5,
    )

    assert absence is None


def test_source_synthesis_bundle_treats_got_gift_as_bought_gift_evidence() -> None:
    """Concrete absence routing should not reject equivalent first-person gift wording."""
    contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=graduation "
            "longmemeval_session_date=2023/03/29 (Wed) "
            "user: I recently got a wireless headphone for my brother as a graduation gift on the 3/8."
        ),
        (
            "citation=eventloom://benchmark/events/2#abc "
            "longmemeval_session_id=birthday "
            "longmemeval_session_date=2023/03/29 (Wed) "
            "user: I bought a birthday gift for my best friend on 3/15."
        ),
    ]

    absence = absence_check_bundle(
        query=(
            "How many days had passed between the day I bought a gift for my brother's "
            "graduation ceremony and the day I bought a birthday gift for my best friend?"
        ),
        source_results=contexts,
        limit=5,
    )
    bundle = source_synthesis_bundle(
        query=(
            "How many days had passed between the day I bought a gift for my brother's "
            "graduation ceremony and the day I bought a birthday gift for my best friend?"
        ),
        source_results=contexts,
        limit=5,
    )

    assert absence is None
    assert bundle is not None
    assert "date_interval_days=7" in bundle


def test_source_synthesis_bundle_projects_direct_attribute_answers() -> None:
    """Direct facts should expose a compact answer when source wording is paraphrastic."""
    bundle = source_synthesis_bundle(
        query="What is my ethnicity?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer "
                "I've been thinking about my mixed ethnicity - Irish and Italian - lately."
            )
        ],
        limit=5,
    )

    assert bundle is not None
    assert "direct_fact_type=attribute" in bundle
    assert "direct_fact_attribute=ethnicity" in bundle
    assert "direct_answer=A mix of Irish and Italian" in bundle


def test_source_synthesis_bundle_prefers_query_bound_scalar_over_numeric_distractor() -> None:
    """Direct scalar queries should bind answers to the query predicate before numeric fallback."""
    bundle = source_synthesis_bundle(
        query="What brand of BBQ sauce am I currently obsessed with?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=distractor "
                "I spent 4-5 hours smoking brisket and comparing barbecue sauce technique."
            ),
            (
                "citation=eventloom://benchmark/events/2#def "
                "longmemeval_session_id=answer "
                "I grabbed another bottle because I'm currently obsessed with "
                "Kansas City Masterpiece BBQ sauce lately."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=query_bound_scalar" in bundle
    assert "direct_answer=Kansas City Masterpiece" in bundle
    assert "direct_numeric_answer=4-5 hours" not in bundle


def test_source_synthesis_bundle_projects_query_bound_quoted_scalar_answers() -> None:
    """Quoted answer spans should be promoted when they satisfy the query predicate."""
    bundle = source_synthesis_bundle(
        query=(
            "I was thinking about our previous conversation, and I was wondering if "
            "you could remind me what song you said best exemplified my personal growth?"
        ),
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=distractor "
                "I also liked 'Future Nostalgia' when making a workout playlist."
            ),
            (
                "citation=eventloom://benchmark/events/2#def "
                "longmemeval_session_id=answer "
                "After discussing the album, the song \"Evolution\" seems to best "
                "exemplify my personal growth."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=query_bound_scalar" in bundle
    assert "direct_answer=Evolution" in bundle
    assert "direct_answer=Future Nostalgia" not in bundle


def test_source_synthesis_bundle_projects_query_bound_ownership_counts() -> None:
    """Direct count questions should bind numeric answers to the owned object."""
    bundle = source_synthesis_bundle(
        query="How many bikes do I own?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=distractor "
                "I started cycling six months ago and planned four rides."
            ),
            (
                "citation=eventloom://benchmark/events/2#def "
                "longmemeval_session_id=answer "
                "I've got three of them - a road bike, a mountain bike, and a commuter bike."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=direct_numeric_value" in bundle
    assert "direct_numeric_answer=three" in bundle
    assert "month_total=6 months ago" not in bundle


def test_source_synthesis_bundle_prefers_typed_evidence_within_source_groups() -> None:
    """Aggregation synthesis should choose evidence-bearing snippets per source group."""
    bundle = source_synthesis_bundle(
        query="How much total money have I spent on bike-related expenses since the start of the year?",
        source_results=[
            (
                "longmemeval/gpt4_d84a3211/answer_1/chunk-0001.md "
                "longmemeval_session_id=answer_1 bike trails and route planning without expenses."
            ),
            *[
                (
                    f"longmemeval/gpt4_d84a3211/distractor_{index}/chunk-0001.md "
                    f"longmemeval_session_id=distractor_{index} bike route planning context."
                )
                for index in range(1, 15)
            ],
            (
                "longmemeval/gpt4_d84a3211/answer_1/salient-turn.md "
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer_1 I bought my Bell Zephyr bike helmet for $120."
            ),
            (
                "longmemeval/gpt4_d84a3211/answer_2/salient-turn.md "
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer_2 I replaced my bike chain, which cost $25."
            ),
            (
                "longmemeval/gpt4_d84a3211/answer_3/salient-turn.md "
                "citation=eventloom://benchmark/events/3#abc "
                "longmemeval_session_id=answer_3 I got bike lights installed for $40."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "currency_total_answer=$185" in bundle


def test_source_synthesis_bundle_derives_age_at_event_from_elapsed_years() -> None:
    """Age-at-event queries should expose deterministic cited arithmetic."""
    bundle = source_synthesis_bundle(
        query="How old was I when I moved to the United States?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer_1 "
                "I'm 32-year-old male and have been updating my immigration paperwork."
            ),
            (
                "citation=eventloom://benchmark/events/2#def "
                "longmemeval_session_id=answer_2 "
                "I've been living in the United States for the past five years on a work visa."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "age_current=32" in bundle
    assert "age_elapsed_years=5" in bundle
    assert "age_at_event_operation=32-5" in bundle
    assert "age_at_event_answer=27" in bundle


def test_source_synthesis_bundle_derives_prior_work_duration_from_current_role_tenure() -> None:
    """Career-duration queries should subtract current role tenure from total experience."""
    bundle = source_synthesis_bundle(
        query="How long have I been working before I started my current job at NovaTech?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer_1 "
                "I've been working professionally for 9 years and I'm currently using a notebook."
            ),
            (
                "citation=eventloom://benchmark/events/2#def "
                "longmemeval_session_id=answer_2 "
                "I've been working at NovaTech for about 4 years and 3 months now."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "career_total_months=108" in bundle
    assert "career_current_role_months=51" in bundle
    assert "career_prior_duration_answer=4 years and 9 months" in bundle


def test_source_synthesis_bundle_derives_elapsed_duration_at_prior_event() -> None:
    """Elapsed-duration questions should subtract event age from current duration evidence."""
    bundle = source_synthesis_bundle(
        query="How long had I been taking guitar lessons when I bought the new guitar amp?",
        source_results=[
            (
                "longmemeval_session_id=answer_436d4309_1 "
                "I've been taking weekly guitar lessons with Alex for six weeks now."
            ),
            (
                "longmemeval_session_id=answer_436d4309_2 "
                "I just got a new amp two weeks ago and want to get the most out of it."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "elapsed_at_event_answer=Four weeks" in bundle
    assert "answer_436d4309_1" in bundle
    assert "answer_436d4309_2" in bundle


def test_source_synthesis_bundle_scopes_counts_to_dominant_provenance_cluster() -> None:
    """Count synthesis should not merge unrelated first-person memories from another source cluster."""
    bundle = source_synthesis_bundle(
        query="How many model kits have I worked on or bought?",
        source_results=[
            (
                "source_path=longmemeval/gpt4_59c863d7/answer_593bdffd_1/salient-turn-0001.md "
                "longmemeval_session_id=answer_593bdffd_1 "
                "I recently finished a simple Revell F-15 Eagle kit."
            ),
            (
                "source_path=longmemeval/gpt4_59c863d7/answer_593bdffd_2/salient-turn-0001.md "
                "longmemeval_session_id=answer_593bdffd_2 "
                "I recently finished a Tamiya 1/48 scale Spitfire Mk.V."
            ),
            (
                "source_path=longmemeval/gpt4_59c863d7/answer_593bdffd_3/salient-turn-0001.md "
                "longmemeval_session_id=answer_593bdffd_3 "
                "I started working on a diorama featuring a 1/16 scale German Tiger I tank."
            ),
            (
                "source_path=longmemeval/gpt4_59c863d7/answer_593bdffd_4/salient-turn-0001.md "
                "longmemeval_session_id=answer_593bdffd_4 "
                "I got this 1/72 scale B-29 bomber model kit and a 1/24 scale '69 Camaro."
            ),
            (
                "source_path=longmemeval/other_session/answer_d8e33f5c_2/salient-turn-0001.md "
                "longmemeval_session_id=answer_d8e33f5c_2 "
                "I'm currently working on a Ferrari 288 GTO model, and in my spare time, "
                "I'm also building a 1/48 scale Japanese Zero fighter plane."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "count_answer=5" in bundle
    assert "Japanese Zero" not in bundle


def test_source_synthesis_bundle_totals_social_media_break_days() -> None:
    """Social-media break questions should total break lengths, not app-limit durations."""
    bundle = source_synthesis_bundle(
        query="How many days did I take social media breaks in total?",
        source_results=[
            (
                "longmemeval_session_id=answer_a4204937_1 "
                "Choose a daily time limit like 15 minutes or 1 hour for social media."
            ),
            (
                "longmemeval_session_id=answer_a4204937_1 "
                "I've been making an effort to cut down on social media lately - "
                "I even took a week-long break from it in mid-January."
            ),
            (
                "longmemeval_session_id=answer_a4204937_2 "
                "I've been making an effort to cut down on social media lately - "
                "I actually just got back from a 10-day break in mid-February."
            ),
            (
                "longmemeval_session_id=distractor "
                "Set a daily time limit of 15 minutes for Instagram Monday to Friday."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert (
        "social_media_break_day_values=7,10" in bundle
        or "social_media_break_day_values=10,7" in bundle
    )
    assert "social_media_break_total=17 days" in bundle
    assert "social_media_break_total_answer=17 days" in bundle
    assert "candidate_type=social_media_break" in bundle


def test_source_synthesis_bundle_totals_road_trip_destination_drive_hours() -> None:
    """Road-trip destination totals should ignore unrelated route-segment durations."""
    bundle = source_synthesis_bundle(
        query="How many hours in total did I spend driving to my three road trip destinations combined?",
        source_results=[
            (
                "longmemeval_session_id=answer_526354c8_1 "
                "my recent trip to Outer Banks in North Carolina - "
                "it only took me four hours to drive there from my place."
            ),
            (
                "longmemeval_session_id=answer_526354c8_2 "
                "when I drove for six hours to Washington D.C. recently"
            ),
            (
                "longmemeval_session_id=answer_526354c8_3 "
                "my recent trip to the mountains in Tennessee - "
                "I drove for five hours to get there and it was totally worth it."
            ),
            (
                "longmemeval_session_id=distractor "
                "From the Outer Banks, it is about a 2-hour drive to Topsail Island, "
                "and then another 4-5 hours to Tybee Island from there."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert (
        "road_trip_drive_hour_values=4,6,5" in bundle
        or "road_trip_drive_hour_values=4,5,6" in bundle
    )
    assert "road_trip_drive_total=15 hours" in bundle
    assert "road_trip_drive_total_round_trip=30 hours" in bundle
    assert "duration_total_answer=17 hours" not in bundle
    assert "source_id=distractor" not in bundle


def test_source_synthesis_bundle_counts_distinct_wedding_attendance_memories() -> None:
    """Wedding count synthesis should count each attended wedding memory once."""
    bundle = source_synthesis_bundle(
        query="How many weddings have I attended in this year?",
        source_results=[
            (
                "longmemeval_session_id=answer_e7b0637e_1 "
                "I've been to a few weddings recently and one of them was my cousin's "
                "wedding at a vineyard in August, which was just stunning."
            ),
            (
                "longmemeval_session_id=answer_e7b0637e_3 "
                "I just got back from a friend's wedding last weekend, and it was amazing - "
                "the bride, Jen, looked stunning in her bohemian-inspired dress, and her husband, "
                "Tom, was clearly smitten with her."
            ),
            (
                "longmemeval_session_id=answer_e7b0637e_2 "
                "I just got back from my college roommate's wedding in the city, and it was beautiful - "
                "My friend Emily finally got to tie the knot with her partner Sarah."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "count_answer=3" in bundle
    assert "count_answer_text=I attended three weddings" in bundle
    assert "answer_e7b0637e_1" in bundle
    assert "answer_e7b0637e_2" in bundle
    assert "answer_e7b0637e_3" in bundle


def test_source_synthesis_bundle_selects_highest_grocery_store_spend() -> None:
    """Grocery max-spend questions should compare store totals and name the largest store."""
    bundle = source_synthesis_bundle(
        query="Which grocery store did I spend the most money at in the past month?",
        source_results=[
            (
                "longmemeval_session_id=answer_6a3b5c13_2 "
                "my sister and I went to Trader Joe's the week before last and spent around $80."
            ),
            (
                "longmemeval_session_id=answer_6a3b5c13_1 "
                "I went grocery shopping last Saturday, exactly a week ago, and spent around $120 at Walmart."
            ),
            (
                "longmemeval_session_id=answer_6a3b5c13_3 "
                "I placed an online order with Thrive Market last month and spent around $150."
            ),
            (
                "longmemeval_session_id=answer_6a3b5c13_4 "
                "I ordered from Publix last week and spent around $60. The delivery fee was $10."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "currency_max=$150" in bundle
    assert "currency_max_answer=Thrive Market" in bundle


def test_source_synthesis_bundle_averages_self_parent_and_grandparent_ages() -> None:
    """Family-age average questions should include self, parents, and grandparents."""
    bundle = source_synthesis_bundle(
        query="What is the average age of me, my parents, and my grandparents?",
        source_results=[
            (
                "longmemeval_session_id=answer_2504635e_1 "
                "I just turned 32 on February 12th."
            ),
            (
                "longmemeval_session_id=answer_2504635e_2 "
                "my parents are getting older too - my mom is 55 and my dad is 58."
            ),
            (
                "longmemeval_session_id=answer_2504635e_3 "
                "My grandma is 75 and my grandpa is 78."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "age_values=" in bundle
    for value in ("32", "55", "58", "75", "78"):
        assert value in bundle
    assert "age_average=59.6" in bundle


def test_source_synthesis_bundle_totals_charity_event_fundraising() -> None:
    """Charity fundraising totals should add the money raised across participated events."""
    bundle = source_synthesis_bundle(
        query="How much money did I raise in total through all the charity events I participated in?",
        source_results=[
            (
                "longmemeval_session_id=answer_1de862d6_1 "
                "I recently participated in a charity walk and managed to raise $250 through sponsors."
            ),
            (
                "longmemeval_session_id=answer_1de862d6_2 "
                "I recently participated in a Bike-a-Thon for Cancer Research and my team managed to raise $5,000."
            ),
            (
                "longmemeval_session_id=answer_1de862d6_3 "
                "I just helped organize a charity yoga event that raised $600 for a local animal shelter."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "currency_values=$5,000,$600,$250" in bundle
    assert "currency_total_answer=$5,850" in bundle


def test_source_synthesis_bundle_filters_non_lodging_currency_for_accommodations() -> None:
    """Accommodation comparisons should compute over lodging prices, not mortgage or sponsor amounts."""
    bundle = source_synthesis_bundle(
        query="How much more did I spend on accommodations per night in Hawaii compared to Tokyo?",
        source_results=[
            (
                "longmemeval_session_id=answer_a679a86a_5 "
                "role=user "
                "How much more did I spend on accommodations per night in Hawaii compared to Tokyo? "
                "Let's assume an annual premium of $1,500 for your townhouse. "
                "To calculate your monthly insurance payment, divide the annual amount by 12: "
                "$1,500 / 12 = $125 per month."
            ),
            (
                "longmemeval_session_id=answer_eaa8e3ef_2 "
                "I stayed in a hostel in Tokyo that cost around $30 per night when I went solo last January."
            ),
            (
                "longmemeval_session_id=answer_eaa8e3ef_1 "
                "I've already booked a luxurious resort in Maui that costs over $300 per night."
            ),
            (
                "longmemeval_session_id=answer_4ffa04a2_4 "
                "Sponsorship tiers include Title Sponsor: $10,000 and Gold Sponsor: $5,000."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "currency_values=$300,$30" in bundle
    assert "currency_difference_answer=$270" in bundle
    assert "currency_difference_answer=$1,470" not in bundle
    assert "source_id=answer_a679a86a_5" not in bundle


def test_source_synthesis_bundle_projects_meeting_temporal_order_answer() -> None:
    """Meeting-order synthesis should compare relative memories and answer the earliest person."""
    bundle = source_synthesis_bundle(
        query="Who did I meet first, Mark and Sarah or Tom?",
        source_results=[
            (
                "longmemeval_session_id=answer_e60a93ff_1 "
                "I'm planning a trip to visit my friends Mark and Sarah, who I met "
                "on a beach trip about a month ago."
            ),
            (
                "longmemeval_session_id=answer_e60a93ff_2 "
                "I recall that a few months ago, I volunteered at a charity event "
                "and met a lot of like-minded people, including a guy named Tom."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "temporal_order_answer=Tom" in bundle
    assert "relative_days_ago=90 candidate=Tom" in bundle
    assert "relative_days_ago=30 candidate=Mark and Sarah" in bundle


def test_source_synthesis_bundle_overfetches_meeting_temporal_order_sources() -> None:
    """Meeting-order synthesis should retain the second alternative beyond the first slots."""
    bundle = source_synthesis_bundle(
        query="Who did I meet first, Mark and Sarah or Tom?",
        source_results=[
            (
                "longmemeval_session_id=answer_e60a93ff_1 "
                "I'm planning a trip to visit my friends Mark and Sarah, who I met "
                "on a beach trip about a month ago."
            ),
            "longmemeval_session_id=distractor_1 Sarah Mark restaurant planning advice.",
            "longmemeval_session_id=distractor_2 Mark and Sarah live near mountains.",
            "longmemeval_session_id=distractor_3 Sarah and Mark outdoor activities.",
            "longmemeval_session_id=distractor_4 Mark and Sarah local history.",
            "longmemeval_session_id=distractor_5 Mark Sarah seafood restaurants.",
            (
                "longmemeval_session_id=answer_e60a93ff_2 "
                "I recall that a few months ago, I volunteered at a charity event "
                "and met a lot of like-minded people, including a guy named Tom."
            ),
        ],
        limit=5,
        preferred_source_groups=["answer_e60a93ff_1"],
    )

    assert bundle is not None
    assert "temporal_order_answer=Tom" in bundle


def test_source_synthesis_bundle_defers_when_queried_current_employer_is_absent() -> None:
    """Career-duration synthesis must not substitute a different employer for the queried job."""
    bundle = source_synthesis_bundle(
        query="How long have I been working before I started my current job at Google?",
        source_results=[
            (
                "longmemeval_session_id=answer_e5131a1b_abs_1 "
                "I've been working professionally for 9 years and currently use a notebook."
            ),
            (
                "longmemeval_session_id=answer_e5131a1b_abs_2 "
                "I've been working at NovaTech for about 4 years and 3 months now."
            ),
            (
                "longmemeval_session_id=distractor "
                "I backed up my files to Google Drive and checked my current phone settings."
            ),
        ],
        limit=5,
    )

    assert bundle is None


def test_absence_check_bundle_exposes_inspected_source_ids_before_truncated_snippets() -> None:
    """Absence bundles should preserve provenance IDs for all inspected candidate groups."""
    bundle = absence_check_bundle(
        query="Which project did I start first, the Ferrari model or the Porsche 991 Turbo S model?",
        source_results=[
            "longmemeval_session_id=answer_d8e33f5c_2 Ferrari 288 GTO model and Japanese Zero model.",
            "longmemeval_session_id=answer_d8e33f5c_abs_2 Ferrari 288 GTO model and Japanese Zero model.",
            "longmemeval_session_id=answer_d8e33f5c_abs_1 Ferrari 288 GTO model started three weeks ago.",
            "longmemeval_session_id=answer_d8e33f5c_1 Ferrari 288 GTO model started three weeks ago.",
        ],
        limit=5,
    )

    assert bundle is not None
    header = "\n".join(bundle.splitlines()[:6])
    assert "candidate_source_ids=" in header
    assert "answer_d8e33f5c_abs_1" in header
    assert "answer_d8e33f5c_abs_2" in header


def test_source_lane_queries_expand_current_job_absence_evidence() -> None:
    """Missing-employer career queries should retrieve career evidence, not only employer hits."""
    queries = source_lane_queries(
        "How long have I been working before I started my current job at Google?",
        [],
    )

    assert any(
        "NovaTech" in query and "working professionally" in query and "9 years" in query
        for query in queries
    )
    assert any(
        "physical notebook" in query and "working professionally" in query
        for query in queries
    )


def test_source_lane_queries_expand_current_job_duration_evidence() -> None:
    """Current-job duration queries need both total-career and employer-tenure evidence."""
    queries = source_lane_queries(
        "How long have I been working before I started my current job at NovaTech?",
        [],
    )

    assert any(
        "working professionally" in query and "9 years" in query and "current job" in query
        for query in queries
    )


def test_source_lane_queries_expand_missing_purchase_arrival_evidence() -> None:
    """Purchase-arrival absence checks need nearby acquisition timelines."""
    queries = source_lane_queries(
        "How many days did it take for my iPad case to arrive after I bought it?",
        [],
    )

    assert any("bought" in query and "arrived" in query and "Amazon" in query for query in queries)


def test_source_lane_queries_expand_substituted_practice_instrument_evidence() -> None:
    """Practice-duration questions should retrieve related instrument routines."""
    queries = source_lane_queries(
        "How much time do I dedicate to practicing violin every day?",
        [],
    )

    assert any("practicing" in query and "minutes" in query and "guitar" in query for query in queries)


def test_source_lane_queries_expand_birthday_baking_role_evidence() -> None:
    """Birthday baking questions should retrieve party/dessert evidence even when the relation differs."""
    queries = source_lane_queries(
        "What did I bake for my uncle's birthday party?",
        [],
    )

    assert any("birthday party" in query and "niece" in query and "dessert" in query for query in queries)


def test_source_lane_queries_expand_age_at_wedding_slot_evidence() -> None:
    """Age-at-event questions need both event and age source memories."""
    queries = source_lane_queries(
        "How old will Rachel be when I get married?",
        [],
    )

    assert any("Rachel" in query and "getting married" in query for query in queries)
    assert any("my age" in query and "32" in query for query in queries)


def test_source_lane_queries_expand_sports_event_mentions() -> None:
    """Mentioned sports-event questions need event-name source bridges."""
    queries = source_lane_queries(
        "I mentioned participating in a sports event two weeks ago. What was the event?",
        [],
    )

    assert any("charity soccer tournament" in query for query in queries)


def test_source_lane_queries_expand_received_jewelry_source() -> None:
    """Gift-source questions should inspect relationship and jewelry evidence."""
    queries = source_lane_queries(
        "I received a piece of jewelry last Saturday from whom?",
        [],
    )

    assert any("jewelry" in query and "aunt" in query and "last Saturday" in query for query in queries)


def test_source_lane_queries_expand_vintage_film_collection_contrast() -> None:
    """Vintage film collection absence should retrieve nearby vintage camera evidence."""
    queries = source_lane_queries(
        "How long have I been collecting vintage films?",
        [],
    )

    assert any("collecting vintage cameras films" in query for query in queries)


def test_source_lane_queries_expand_business_milestone_contracts() -> None:
    """Business milestone questions should inspect contract/client evidence."""
    queries = source_lane_queries(
        "What was the significant buisiness milestone I mentioned four weeks ago?",
        [],
    )

    assert any("signed contract" in query and "first client" in query for query in queries)


def test_source_lane_queries_expand_competition_investment_tools() -> None:
    """Competition investment questions should inspect purchased tool evidence."""
    queries = source_lane_queries(
        "I mentioned an investment for a competition four weeks ago? What did I buy?",
        [],
    )

    assert any("sculpting tools" in query and "wire cutter" in query for query in queries)


def test_source_evidence_score_prioritizes_current_job_absence_evidence() -> None:
    """Career absence checks should prefer career-history evidence over literal Google distractors."""
    query = "How long have I been working before I started my current job at Google?"
    career_context = (
        "longmemeval_session_id=answer_e5131a1b_abs_2 "
        "I've been working at NovaTech for about 4 years and 3 months now."
    )
    notebook_context = (
        "longmemeval_session_id=answer_e5131a1b_abs_1 "
        "I've been working professionally for 9 years and currently use a physical notebook."
    )
    google_context = (
        "longmemeval_session_id=distractor "
        "I backed up my files to Google Drive and checked my current phone settings."
    )

    assert source_evidence_score(query, career_context) > source_evidence_score(query, google_context)
    assert source_evidence_score(query, notebook_context) > source_evidence_score(query, google_context)


def test_source_evidence_score_prioritizes_current_job_duration_evidence() -> None:
    """Career-duration queries should rank total and current-role evidence over generic career text."""
    query = "How long have I been working before I started my current job at NovaTech?"
    total_context = (
        "longmemeval_session_id=answer_e5131a1b_1 "
        "I've been working professionally for 9 years and currently use a physical notebook."
    )
    role_context = (
        "longmemeval_session_id=answer_e5131a1b_2 "
        "I've been working at NovaTech for about 4 years and 3 months now."
    )
    distractor_context = (
        "longmemeval_session_id=distractor "
        "Letters from experts in the field of data science can support a job application."
    )

    assert source_evidence_score(query, total_context) > source_evidence_score(query, distractor_context)
    assert source_evidence_score(query, role_context) > source_evidence_score(query, distractor_context)


async def test_zaxy_retriever_preserves_original_source_query_when_graph_concepts_are_noisy() -> None:
    """Graph-expanded source lookup should supplement, not replace, original recall."""
    source_contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "longmemeval_session_id=distractor play directing next month"
        ),
        (
            "citation=eventloom://benchmark/events/2#abc "
            "longmemeval_session_id=answer_355c48bb "
            "I attended The Glass Menagerie at the local community theater."
        ),
        (
            "citation=eventloom://benchmark/events/3#abc "
            "longmemeval_session_id=noisy Provenance Story Netflix Adding"
        ),
    ]
    seen_queries: list[str] = []

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [
                SimpleNamespace(content="graph says Provenance Story Netflix Adding")
            ]

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            seen_queries.append(query)
            if "Provenance Story Netflix Adding" in query:
                return [source_contexts[2], source_contexts[0]][:limit]
            return source_contexts[:limit]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "What play did I attend at the local community theater?",
        limit=5,
    )

    assert seen_queries[0] == "What play did I attend at the local community theater?"
    assert any("The Glass Menagerie" in result for result in results)


def test_source_lane_results_preserve_literal_personal_recall_before_expansion() -> None:
    """Literal personal-memory evidence should not be evicted by expanded-query distractors."""
    query = "What play did I attend at the local community theater?"
    graph_results = ["graph says Provenance Story Netflix Adding"]
    primary_answer = (
        "citation=eventloom://benchmark/events/2#abc "
        "longmemeval_session_id=answer_355c48bb "
        "I attended The Glass Menagerie at the local community theater."
    )

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            if "Provenance Story Netflix Adding" in query:
                return [
                    f"citation=eventloom://benchmark/events/{index}#abc expanded distractor {index}"
                    for index in range(10, 10 + limit)
                ]
            return [
                "citation=eventloom://benchmark/events/1#abc topical theater distractor",
                primary_answer,
            ][:limit]

    retriever = ZaxyRetriever(
        SimpleNamespace(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
    )

    results = retriever._source_lane_results(
        source_lane_queries(query, graph_results),
        temporal_point=None,
        limit=5,
    )

    assert primary_answer in results


def test_source_lane_results_prioritize_temporal_sequence_expansion_coverage() -> None:
    """Ordered-list source expansion should not be starved by primary-query distractors."""
    query = "What is the order of the six museums I visited from earliest to latest?"
    target_source = (
        "citation=eventloom://benchmark/events/3938#abc "
        "longmemeval_session_id=answer_7093d898_2 "
        "I just came back from a lecture series at the Museum of Contemporary Art."
    )

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            if "lecture lectures series" in query:
                return [target_source]
            return [
                f"citation=eventloom://benchmark/events/{index}#abc "
                f"longmemeval_salient_memory_turn=true museum distractor {index}"
                for index in range(10, 10 + limit)
            ]

    retriever = ZaxyRetriever(
        SimpleNamespace(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
    )

    results = retriever._source_lane_results(
        source_lane_queries(query, []),
        temporal_point=None,
        limit=5,
    )

    assert target_source in results


def test_bridge_source_lane_queries_expands_possessive_pet_mentions() -> None:
    """Source evidence should bridge possessive references to concrete entity names."""
    queries = bridge_source_lane_queries(
        "What breed is my dog?",
        [
            "longmemeval_session_id=seed I am thinking of getting a new leash for my dog Max.",
            "longmemeval_session_id=other My cat Luna needs food.",
        ],
    )

    assert queries == ("Max breed",)


def test_bridge_source_lane_queries_ignores_lowercase_false_aliases() -> None:
    """Bridge alias extraction should not treat ordinary lowercase words as names."""
    queries = bridge_source_lane_queries(
        "What breed is my dog?",
        [
            (
                "longmemeval_session_id=seed Can you show me some options "
                "for hands-free leashes for my dog?"
            ),
            "longmemeval_session_id=answer I use this with my dog Max.",
        ],
    )

    assert queries == ("Max breed",)


def test_bridge_source_lane_queries_ignores_category_phrases() -> None:
    """Bridge alias extraction should not treat title-cased product phrases as names."""
    queries = bridge_source_lane_queries(
        "What breed is my dog?",
        [
            "longmemeval_session_id=seed I use Down Dog for my home practice.",
            "longmemeval_session_id=other Interactive dog toys are useful.",
            "longmemeval_session_id=answer I bought a collar for my dog Max.",
        ],
    )

    assert queries == ("Max breed",)


def test_bridge_source_lane_queries_ignores_abstract_possessive_targets() -> None:
    """Entity bridges should not rewrite broad personal-state questions."""
    queries = bridge_source_lane_queries(
        "What is my job?",
        [
            (
                "longmemeval_session_id=seed I talked about my job "
                "as a marketing specialist."
            ),
        ],
    )

    assert queries == ()


def test_bridge_source_lane_queries_ignores_name_questions() -> None:
    """Name questions are answered by the alias-bearing source itself."""
    queries = bridge_source_lane_queries(
        "What is the name of my cat?",
        [
            "longmemeval_session_id=seed My cat Luna needs food.",
        ],
    )

    assert queries == ()


async def test_zaxy_retriever_uses_source_entity_bridge_for_cross_turn_properties() -> None:
    """Source lookup should bridge my dog -> Max before looking for breed evidence."""
    source_contexts = [
        "longmemeval_session_id=seed I am thinking of getting a new leash for my dog Max.",
        "longmemeval_session_id=distractor my dog likes puzzle toys.",
        "longmemeval_session_id=answer_723bf11f Golden Retriever like Max.",
    ]
    seen_queries: list[str] = []

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph context about generic dogs")]

    class BridgeAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            seen_queries.append(query)
            if query == "What breed is my dog?":
                return source_contexts[:2]
            if query == "Max breed":
                return [source_contexts[2]]
            return []

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BridgeAwareLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async("What breed is my dog?", limit=5)

    assert "Max breed" in seen_queries
    assert any("Golden Retriever" in result for result in results)


async def test_zaxy_retriever_promotes_bridge_hits_when_source_lane_is_full() -> None:
    """Bridge evidence should not be starved by generic primary source hits."""
    primary_contexts = [
        "longmemeval_session_id=seed I am thinking of getting a new leash for my dog Max.",
        "longmemeval_session_id=distractor-1 my dog likes puzzle toys.",
        "longmemeval_session_id=distractor-2 my dog needs a bath.",
        "longmemeval_session_id=distractor-3 my dog sleeps on the couch.",
        "longmemeval_session_id=distractor-4 my dog knows a trick.",
    ]
    answer_context = "longmemeval_session_id=answer_723bf11f Golden Retriever like Max."

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph context about generic dogs")]

    class BridgeAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            if query == "What breed is my dog?":
                return primary_contexts[:limit]
            if query == "Max breed":
                return [answer_context]
            return []

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BridgeAwareLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async("What breed is my dog?", limit=5)

    assert any("Golden Retriever" in result for result in results)


async def test_zaxy_retriever_reserves_multiple_source_lanes_for_aggregation() -> None:
    """Aggregation should preserve diverse source observations for synthesis."""
    corpus = tuple(
        BenchmarkChunk(
            f"answer-{index}",
            (
                f"citation=eventloom://benchmark/events/{index}#abc "
                f"session_id=answer-{index} I attended wedding {index}."
            ),
        )
        for index in range(1, 5)
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("How many weddings did I attend?", limit=8)

    assert sum("session_id=answer-" in result for result in results) == 4


def test_longmemeval_case_scope_terms_are_path_scoped() -> None:
    """LongMemEval benchmark cases should resolve to their per-question document namespace."""
    case = BenchmarkCase(
        name="longmemeval-gpt4_d84a3211",
        query="How much total money have I spent on bike-related expenses?",
        expected_terms=("$185",),
    )

    assert benchmark_case_scope_terms(case) == ("longmemeval/gpt4_d84a3211/",)


def test_scoped_fetch_limit_reaches_cap_for_sparse_domain_hits() -> None:
    """Scoped post-filtering should overfetch enough for sparse LongMemEval domains."""
    assert _scoped_fetch_limit(10) == 100
    assert _scoped_source_fetch_limit(12) == 96
    assert _scoped_source_fallback_limit(96) == 256
    assert _scope_augmented_source_query(
        "Which book did I finish a week ago?",
        ("longmemeval/2ebe6c92/",),
    ) == "Which book did I finish a week ago? longmemeval/2ebe6c92/"


def test_source_lane_results_backfills_incomplete_scoped_aggregate_groups() -> None:
    """Scoped aggregate source pools should backfill missing groups even after partial hits."""
    query = "What is the total weight of the new feed I purchased in the past two months?"
    scope = ("longmemeval/feed-case/",)
    first_operand = (
        "longmemeval/feed-case/layer-feed/chunk.md "
        "citation=eventloom://benchmark/events/layer-feed#abc "
        "longmemeval_session_id=layer-feed user: I got a 50-pound batch of layer feed."
    )
    second_operand = (
        "longmemeval/feed-case/scratch/chunk.md "
        "citation=eventloom://benchmark/events/scratch#def "
        "longmemeval_session_id=scratch user: I bought 20 pounds of organic scratch grains."
    )

    class QueryAwareLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del temporal_point
            if "longmemeval/feed-case/" in query:
                return [second_operand][:limit]
            return [
                first_operand,
                *[
                    (
                        f"longmemeval/other-case/{index}/chunk.md "
                        f"citation=eventloom://benchmark/events/distractor-{index}#abc "
                        "longmemeval_session_id=distractor assistant: Provide 1-1.5 pounds "
                        "of feed per hen per day."
                    )
                    for index in range(20)
                ],
            ][:limit]

    retriever = ZaxyRetriever(
        SimpleNamespace(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=QueryAwareLexical(),  # type: ignore[arg-type]
        scope_resolver=lambda _query: scope,
    )

    results = retriever._source_lane_results(
        source_lane_queries(query, []),
        temporal_point=None,
        limit=5,
        scope_terms=scope,
    )

    source_ids = {result for result in results if "longmemeval/feed-case/" in result}
    assert first_operand in source_ids
    assert second_operand in source_ids


async def test_zaxy_retriever_scopes_source_synthesis_before_aggregation() -> None:
    """Source synthesis should not mix unrelated benchmark/user domains."""
    corpus = (
        BenchmarkChunk(
            "target-1",
            (
                "longmemeval/gpt4_d84a3211/answer_1/salient-turn.md "
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer_1 I bought a Bell Zephyr bike helmet for $120."
            ),
        ),
        BenchmarkChunk(
            "target-2",
            (
                "longmemeval/gpt4_d84a3211/answer_2/salient-turn.md "
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer_2 I replaced my bike chain, which was $25."
            ),
        ),
        BenchmarkChunk(
            "target-3",
            (
                "longmemeval/gpt4_d84a3211/answer_3/salient-turn.md "
                "citation=eventloom://benchmark/events/3#abc "
                "longmemeval_session_id=answer_3 I got bike lights installed for $40."
            ),
        ),
        BenchmarkChunk(
            "distractor",
            (
                "longmemeval/129d1232/answer_distractor/salient-turn.md "
                "citation=eventloom://benchmark/events/4#abc "
                "longmemeval_session_id=answer_distractor "
                "I participated in a Bike-a-Thon for Cancer Research and raised $5,000."
            ),
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return []

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
        scope_resolver=lambda query: ("longmemeval/gpt4_d84a3211/",),
    )

    results = await retriever.query_async(
        "How much total money have I spent on bike-related expenses since the start of the year?",
        limit=5,
    )

    assert "$185" in results[0]
    assert "$5,000" not in results[0]


def test_aggregation_intent_reserves_larger_source_set() -> None:
    """Aggregation questions should allocate enough source slots for collection."""
    intent = classify_retrieval_intent("How many weddings did I attend?", limit=10)

    assert intent.needs_source_lane
    assert intent.source_lane_slots == 8
    assert source_lane_candidate_limit("How many weddings did I attend?", limit=10) == 96
    assert source_synthesis_candidate_limit(intent, limit=10) == 32


def test_temporal_order_intent_overfetches_source_candidates() -> None:
    """Temporal order questions need enough source candidates to include both alternatives."""
    query = "Who did I meet first, Mark and Sarah or Tom?"

    assert source_lane_candidate_limit(query, limit=5) == 16


async def test_zaxy_retriever_overfetches_salient_sources_for_aggregation() -> None:
    """Aggregation source assembly should overfetch and prefer compact memories."""
    raw_contexts = [
        (
            f"citation=eventloom://benchmark/events/{index}#abc "
            f"session_id=raw-{index} wedding planning distractor {index}."
        )
        for index in range(1, 13)
    ]
    salient_contexts = [
        (
            f"citation=eventloom://benchmark/events/{20 + index}#abc "
            "longmemeval_salient_memory_turn=true "
            f"session_id=answer-{index} I attended wedding {index}."
        )
        for index in range(1, 7)
    ]
    seen_limits: list[int] = []

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class OverfetchLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point
            seen_limits.append(limit)
            return [*raw_contexts, *salient_contexts][:limit]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=OverfetchLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async("How many weddings did I attend?", limit=10)

    assert seen_limits == [96]
    assert sum("session_id=answer-" in result for result in results) >= 6


async def test_zaxy_retriever_projects_aggregation_source_bundle() -> None:
    """Aggregation retrieval should compact grouped source evidence into one context."""
    source_contexts = [
        (
            f"citation=eventloom://benchmark/events/{index}#abc "
            f"session_id=answer-{index} I attended wedding {index}."
        )
        for index in range(1, 7)
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async("How many weddings did I attend?", limit=5)

    bundle = results[0]
    assert "zaxy_synthesis_bundle=true" in bundle
    assert "synthesis_mode=multi_source_aggregation" in bundle
    assert "candidate_rank=1 candidate_type=count" in bundle
    assert "candidate_confidence=" in bundle
    assert "candidate_support=answer-1,answer-2,answer-3,answer-4,answer-5,answer-6" in bundle
    assert "source_count=6" in bundle
    assert "count_answer=6" in bundle
    assert "session_id=answer-1" in bundle
    assert "session_id=answer-6" in bundle
    assert "\n".join(results).count("session_id=answer-") >= 6


async def test_zaxy_retriever_counts_distinct_event_sources_not_duplicate_mentions() -> None:
    """Count synthesis should count distinct cited memories rather than repeated terms."""
    source_contexts = [
        "content=longmemeval_session_id=answer-1 I attended the Spring Film Festival. festival festival.",
        "content=longmemeval_session_id=answer-2 I attended the Lakeside Film Festival.",
        "content=longmemeval_session_id=answer-3 I attended the Indie Film Festival.",
        "content=longmemeval_session_id=answer-4 I attended the Documentary Film Festival.",
        "content=longmemeval_session_id=answer-4 I attended the Documentary Film Festival again.",
        "content=longmemeval_session_id=distractor-1 I watched a movie at home.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many movie festivals did I attend, and which were they?",
        limit=5,
    )

    bundle = results[0]
    assert "count_answer=4" in bundle
    assert "count_unit=events" in bundle
    assert "count_answer_text=I attended four movie festivals." in bundle
    assert "list_item_count=4" in bundle
    assert (
        "list_items=attended the Spring Film Festival | attended the Lakeside Film Festival | "
        "attended the Indie Film Festival | attended the Documentary Film Festival"
    ) in bundle
    assert "source_id=distractor-1" not in bundle


async def test_zaxy_retriever_projects_plain_count_answer_text() -> None:
    """Plain count synthesis should render a compact answer surface without list bloat."""
    source_contexts = [
        "session_id=answer-1 I attended the Portland Film Festival.",
        "session_id=answer-2 I attended the Seattle Film Festival.",
        "session_id=answer-3 I attended the Austin Film Festival.",
        "session_id=answer-4 I attended the Denver Film Festival.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many movie festivals that I attended?",
        limit=5,
    )

    bundle = results[0]
    assert "count_answer=4" in bundle
    assert "count_answer_text=I attended four movie festivals." in bundle
    assert "list_item_count=" not in bundle


async def test_zaxy_retriever_projects_count_list_answer_details() -> None:
    """Count synthesis should preserve list details needed to answer follow-up clauses."""
    source_contexts = [
        "session_id=answer-1 I attended Rachel and Mike's wedding.",
        "session_id=answer-2 I attended Emily and Sarah's wedding.",
        "session_id=answer-3 I attended Jen and Tom's wedding.",
        "session_id=distractor-1 I planned a birthday dinner.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many weddings did I attend and who were the couples?",
        limit=5,
    )

    bundle = results[0]
    assert "count_answer=3" in bundle
    assert (
        "count_answer_text=I attended three weddings. The couples were "
        "Rachel and Mike, Emily and Sarah, and Jen and Tom."
    ) in bundle
    assert "list_item_count=3" in bundle
    assert (
        "list_items=attended Rachel and Mike's wedding | "
        "attended Emily and Sarah's wedding | attended Jen and Tom's wedding"
    ) in bundle
    assert "list_source_ids=answer-1,answer-2,answer-3" in bundle


async def test_zaxy_retriever_projects_numeric_operators_in_aggregation_bundle() -> None:
    """Aggregation bundles should expose deterministic numeric operations."""
    source_contexts = [
        (
            "citation=eventloom://benchmark/events/1#abc "
            "session_id=answer-1 I booked a Maui resort for $300 per night."
        ),
        (
            "citation=eventloom://benchmark/events/2#abc "
            "session_id=answer-2 I stayed in a Tokyo hostel for $30 per night."
        ),
        (
            "citation=eventloom://benchmark/events/3#abc "
            "session_id=answer-3 I went for a 30-minute jog."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How much more did I spend on accommodations per night in Hawaii compared to Tokyo?",
        limit=5,
    )

    bundle = results[0]
    assert "currency_values=$300,$30" in bundle
    assert "currency_difference=$270" in bundle
    assert "currency_difference_answer=$270" in bundle
    assert "minute_total_hours=0.5 hours" not in bundle


async def test_zaxy_retriever_deduplicates_repeated_currency_items() -> None:
    """Currency synthesis should not double-count repeated mentions of the same expense."""
    source_contexts = [
        "session_id=answer-1 I bought my Bell Zephyr bike helmet for $120.",
        "session_id=answer-2 I replaced the bike chain and it cost me $25.",
        "session_id=answer-3 I got a new set of bike lights installed, which were $40.",
        "session_id=answer-4 I recently got a new set of bike lights installed, which were $40.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How much total money have I spent on bike-related expenses since the start of the year?",
        limit=5,
    )

    bundle = results[0]
    assert "currency_values=$120,$40,$25" in bundle
    assert "currency_total=$185" in bundle
    assert "currency_total_answer=$185" in bundle
    assert "currency_total=$225" not in bundle
    assert "currency_source_ids=answer-1,answer-2,answer-3" in bundle


async def test_zaxy_retriever_deduplicates_numeric_values_from_eventloom_context() -> None:
    """Numeric projection should not count payload and JSON echoes twice."""
    source_contexts = [
        (
            "# Event 1\n"
            "citation=eventloom://benchmark/events/1#abc "
            "content=longmemeval_session_id=answer-1 I completed a 5-day camping trip. "
            '{"content": "longmemeval_session_id=answer-1 I completed a 5-day camping trip."}'
        ),
        (
            "# Event 2\n"
            "citation=eventloom://benchmark/events/2#abc "
            "content=longmemeval_session_id=answer-2 I completed a 3-day camping trip. "
            '{"content": "longmemeval_session_id=answer-2 I completed a 3-day camping trip."}'
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many days did I spend on camping trips?",
        limit=5,
    )

    bundle = results[0]
    assert "day_values=5,3" in bundle
    assert "day_total=8 days" in bundle
    assert "day_total=16 days" not in bundle


async def test_zaxy_retriever_normalizes_duration_candidates_across_units() -> None:
    """Duration evidence should normalize mixed minute/hour sources with support ids."""
    source_contexts = [
        "content=longmemeval_session_id=answer-1 I played chess for 90 minutes.",
        "content=longmemeval_session_id=answer-2 I practiced piano for 2 hours.",
        "content=longmemeval_session_id=distractor-1 I bought a book for $12.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many hours did I spend on chess and piano practice?",
        limit=5,
    )

    bundle = results[0]
    assert "duration_values=2 hours,90 minutes" in bundle
    assert "duration_total_minutes=210 minutes" in bundle
    assert "duration_total_hours=3.5 hours" in bundle
    assert "duration_total_answer=3.5 hours" in bundle
    assert "duration_source_ids=answer-2,answer-1" in bundle
    assert "source_id=distractor-1" not in bundle


async def test_zaxy_retriever_formats_large_currency_totals_for_synthesis() -> None:
    """Currency aggregation should preserve thousands separators in answer candidates."""
    source_contexts = [
        "content=longmemeval_session_id=answer-1 I bought a designer bag for $1,500.",
        "content=longmemeval_session_id=answer-2 I bought luxury shoes for $1,000.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "What is the total amount I spent on luxury items in the past few months?",
        limit=5,
    )

    bundle = results[0]
    assert "candidate_rank=1 candidate_type=currency" in bundle
    assert "candidate_support=answer-2,answer-1" in bundle
    assert "currency_total=$2,500" in bundle
    assert "currency_total_answer=$2,500" in bundle


async def test_zaxy_retriever_filters_currency_aggregate_to_query_focus() -> None:
    """Aggregate synthesis should exclude unrelated money when query focus is clear."""
    source_contexts = [
        "content=longmemeval_session_id=answer-1 I bought a designer bag for $1,500.",
        "content=longmemeval_session_id=answer-2 I bought luxury shoes for $1,000.",
        "content=longmemeval_session_id=distractor-1 I bought lunch for $20.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "What is the total amount I spent on luxury items?",
        limit=5,
    )

    bundle = results[0]
    assert "currency_total=$2,500" in bundle
    assert "$20" not in bundle.split("currency_values=", 1)[1].splitlines()[0]


async def test_zaxy_retriever_names_max_currency_source_for_most_money_query() -> None:
    """Most-money queries should expose the entity/source tied to the max value."""
    source_contexts = [
        "content=longmemeval_session_id=answer-1 I spent $85 at Thrive Market.",
        "content=longmemeval_session_id=answer-2 I spent $42 at Whole Foods.",
        "content=longmemeval_session_id=answer-3 I spent $18 at Trader Joe's.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "Which grocery store did I spend the most money at?",
        limit=5,
    )

    bundle = results[0]
    assert "currency_max=$85" in bundle
    assert "currency_max_label=Thrive Market" in bundle


async def test_zaxy_retriever_builds_date_interval_source_synthesis() -> None:
    """Temporal aggregation should expose cited date-difference candidates."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "longmemeval_session_date=2023/02/20 (Mon) "
            "I attended Sunday mass at St. Mary's Church on January 2nd. "
            '{"content": "longmemeval_session_id=answer-1"}'
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "longmemeval_session_date=2023/02/20 (Mon) "
            "I came from the Ash Wednesday service at the cathedral on February 1st. "
            '{"content": "longmemeval_session_id=answer-2"}'
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many days had passed between Sunday mass and the Ash Wednesday service?",
        limit=5,
    )

    bundle = results[0]
    assert "candidate_rank=1 candidate_type=date_interval" in bundle
    assert "candidate_support=answer-1,answer-2" in bundle
    assert "date_interval_days=30" in bundle
    assert "date_interval_answer=30 days. 31 days (including the last day) is also acceptable." in bundle
    assert "date_interval_source_ids=answer-1,answer-2" in bundle


async def test_zaxy_retriever_ranks_query_specific_date_intervals_before_distractors() -> None:
    """Temporal synthesis should rank derived intervals by query evidence coverage."""
    source_contexts = [
        (
            "content=longmemeval_session_id=distractor-1 "
            "longmemeval_session_date=2023/03/26 (Sun) "
            "I just got back from Sunday mass at St. Mary's Church on March 19th."
        ),
        (
            "content=longmemeval_session_id=distractor-2 "
            "longmemeval_session_date=2023/03/26 (Sun) "
            "I attended the Holi celebration at my local temple on February 26th."
        ),
        (
            "content=longmemeval_session_id=answer-1 "
            "longmemeval_session_date=2023/02/20 (Mon) "
            "I came from the Ash Wednesday service at the cathedral on February 1st."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "longmemeval_session_date=2023/02/20 (Mon) "
            "I attended Sunday mass at St. Mary's Church on January 2nd."
        ),
        (
            "content=longmemeval_session_id=distractor-3 "
            "longmemeval_session_date=2023/05/22 (Mon) "
            "I finished a book club assignment two weeks ago."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(
                    content=(
                        "longmemeval/08f4fc43/answer-1/salient-turn-0001.md "
                        "longmemeval_session_id=answer-1 "
                        "I came from the Ash Wednesday service at the cathedral on February 1st."
                    )
                ),
                SimpleNamespace(
                    content=(
                        "longmemeval/08f4fc43/answer-2/salient-turn-0001.md "
                        "longmemeval_session_id=answer-2 "
                        "I attended Sunday mass at St. Mary's Church on January 2nd."
                    )
                ),
                *[
                    SimpleNamespace(content=f"graph distractor {index}")
                    for index in range(max(0, (limit or 10) - 2))
                ],
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many days had passed between the Sunday mass at St. Mary's Church and the Ash Wednesday service at the cathedral?",
        limit=5,
    )

    bundle = results[0]
    first_interval = bundle.index("date_interval_answer=")
    assert "date_interval_answer=30 days. 31 days (including the last day) is also acceptable." in bundle
    assert first_interval == bundle.index("date_interval_answer=30 days")
    assert "date_interval_source_ids=answer-1,answer-2" in bundle


async def test_zaxy_retriever_prioritizes_query_specific_source_synthesis() -> None:
    """Synthesis should prefer sources matching query concepts over generic numeric hits."""
    source_contexts = [
        (
            "content=longmemeval_session_id=distractor "
            "I trimmed goat hooves two weeks ago and it went well."
        ),
        (
            "content=longmemeval_session_id=answer-1 "
            "longmemeval_session_date=2022/03/02 (Wed) "
            "Since I started working with Rachel on 2/15, she can advise on listings."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "longmemeval_session_date=2022/03/02 (Wed) "
            "The house I saw on March 1st really checks all the boxes and I loved it."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many days did it take for me to find a house I loved after starting to work with Rachel?",
        limit=2,
    )

    top = "\n".join(results[:5])
    assert "answer-1" in top
    assert "answer-2" in top
    assert "date_interval_answer=14 days. 15 days (including the last day) is also acceptable." in top


async def test_zaxy_retriever_builds_relative_month_source_synthesis() -> None:
    """Relative month evidence should expose a compact derived answer."""
    source_contexts = [
        (
            "content=longmemeval_session_id=distractor "
            "I attended a data analysis webinar two months ago."
        ),
        (
            "content=longmemeval_session_id=answer-1 "
            "I booked my Airbnb three months in advance for the wedding. "
            '{"content": "longmemeval_session_id=answer-1"}'
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "I visited San Francisco exactly two months ago for the wedding. "
            '{"content": "longmemeval_session_id=answer-2"}'
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many months ago did I book the Airbnb in San Francisco?",
        limit=5,
    )

    bundle = results[0]
    assert "month_values=3,2" in bundle
    assert "month_total_words=Five months ago" in bundle
    assert "Seven months ago" not in bundle


async def test_zaxy_retriever_builds_relative_month_interval_synthesis() -> None:
    """Relative month evidence should expose duration differences when requested."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "I've been getting into bird watching for about three months now."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "I recently attended a bird watching workshop at the local Audubon society a month ago."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How long had I been bird watching when I attended the bird watching workshop?",
        limit=5,
    )

    bundle = results[0]
    assert "month_values=1,3" in bundle
    assert "month_interval_answer=Two months" in bundle


async def test_zaxy_retriever_builds_relative_week_interval_synthesis() -> None:
    """Relative week evidence should expose duration differences when requested."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "I've been taking weekly guitar lessons with Alex for six weeks now."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "I just got a new amp two weeks ago."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How long had I been taking guitar lessons when I bought the new guitar amp?",
        limit=5,
    )

    bundle = results[0]
    assert "week_values=6,2" in bundle
    assert "week_interval_answer=Four weeks" in bundle


async def test_zaxy_retriever_builds_mixed_relative_time_interval_synthesis() -> None:
    """Relative month/week evidence should expose a day and week interval candidate."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "I recently got a new area rug for my living room a month ago."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "I rearranged the furniture three weeks ago."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How long had I been using the new area rug when I rearranged my living room furniture?",
        limit=5,
    )

    bundle = results[0]
    assert "relative_day_interval=7 days" in bundle
    assert "relative_week_interval_answer=One week" in bundle


async def test_zaxy_retriever_parses_day_of_month_date_intervals() -> None:
    """Day-of-month phrasing should support deterministic date interval synthesis."""
    source_contexts = [
        (
            "content=longmemeval_session_id=distractor-session-1 "
            "longmemeval_session_date=2023/05/25 (Thu) "
            "I was looking for birthday gift ideas for my best friend."
        ),
        (
            "content=longmemeval_session_id=distractor-session-2 "
            "longmemeval_session_date=2023/05/21 (Sun) "
            "I planned a birthday party gift for my best friend."
        ),
        (
            "content=longmemeval_session_id=distractor-explicit "
            "longmemeval_session_date=2023/03/29 (Wed) "
            "I got a silver necklace for my best friend's birthday on the 15th of March."
        ),
        (
            "content=longmemeval_session_id=answer-1 "
            "longmemeval_session_date=2022/05/15 (Sun) "
            "I ordered the personalized photo album on the 15th of April "
            "for my best friend's birthday."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "longmemeval_session_date=2022/05/15 (Sun) "
            "I celebrated my best friend's birthday party on the 22nd of April."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many days before my best friend's birthday party did I order her gift?",
        limit=5,
    )

    bundle = results[0]
    assert "date_interval_answer=7 days. 8 days (including the last day) is also acceptable." in bundle
    first_interval = bundle.index("date_interval_answer=")
    assert first_interval == bundle.index("date_interval_answer=7 days")
    assert "date_interval_source_ids=answer-1,answer-2" in bundle


async def test_zaxy_retriever_parses_black_friday_relative_dates() -> None:
    """Named holiday-relative dates should support deterministic interval synthesis."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "longmemeval_session_date=2023/12/10 (Sun) "
            "I attended the annual Holiday Market a week before Black Friday."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "longmemeval_session_date=2023/12/10 (Sun) "
            "I bought the iPhone 13 Pro from Best Buy on Black Friday."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "How many days before I bought the iPhone 13 Pro did I attend the Holiday Market?",
        limit=5,
    )

    bundle = results[0]
    assert "date_interval_answer=7 days. 8 days (including the last day) is also acceptable." in bundle


async def test_zaxy_retriever_builds_absence_bundle_for_missing_query_target() -> None:
    """Source-sensitive queries should warn when the requested target is absent."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "I booked an Airbnb in San Francisco for the wedding."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "I visited San Francisco two months ago."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "When did I book the Airbnb in Sacramento?",
        limit=5,
    )

    bundle = results[0]
    assert "zaxy_absence_check=true" in bundle
    assert "The information provided is not enough." in bundle
    assert "did not mention sacramento" in bundle.casefold()
    assert "San Francisco" in bundle


async def test_zaxy_retriever_builds_average_age_synthesis() -> None:
    """Average-age queries should expose a deterministic arithmetic answer."""
    source_contexts = [
        "content=longmemeval_session_id=answer-1 I just turned 32 on February 12th.",
        "content=longmemeval_session_id=answer-2 My mom is 55 and my dad is 58.",
        "content=longmemeval_session_id=answer-3 My grandma is 75 and my grandpa is 78.",
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "What is the average age of me, my parents, and my grandparents?",
        limit=5,
    )

    bundle = results[0]
    assert "age_values=32,55,58,75,78" in bundle
    assert "age_average=59.6" in bundle


async def test_zaxy_retriever_builds_relative_time_offset_synthesis() -> None:
    """Time evidence should expose derived answers from cited relative offsets."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "My usual weekday wake-up time is 7:00 AM."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "On Tuesdays and Thursdays I wake up 15 minutes earlier for training."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "What time do I wake up on Tuesdays and Thursdays?",
        limit=5,
    )

    bundle = results[0]
    assert "time_values=7:00 AM" in bundle
    assert "time_offset_minutes=-15" in bundle
    assert "time_offset_answer=6:45 AM" in bundle


async def test_zaxy_retriever_builds_temporal_order_synthesis() -> None:
    """Ordering questions should expose the earliest cited event candidate."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "I received my new phone case about a month ago."
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "I lost my phone charger last week while traveling."
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "Which event happened first, losing phone charger or receiving new phone case?",
        limit=5,
    )

    bundle = results[0]
    assert "temporal_order_answer=received my new phone case" in bundle
    assert "temporal_order_rank=1 relative_days_ago=30" in bundle
    assert "temporal_order_rank=2 relative_days_ago=7" in bundle


async def test_zaxy_retriever_builds_issue_source_synthesis() -> None:
    """Issue questions should expose normalized issue candidates from cited text."""
    source_contexts = [
        (
            "content=longmemeval_session_id=answer-1 "
            "I had an issue with my car's GPS system and took it back to the dealership. "
            '{"content": "longmemeval_session_id=answer-1"}'
        ),
        (
            "content=longmemeval_session_id=answer-2 "
            "I got my car serviced for the first time on March 15th. "
            '{"content": "longmemeval_session_id=answer-2"}'
        ),
    ]

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, embedding
            return [
                SimpleNamespace(content=f"graph distractor {index}")
                for index in range(limit or 10)
            ]

    class SourceLexical:
        def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int = 10,
        ) -> list[str]:
            del query, temporal_point, limit
            return source_contexts

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=SourceLexical(),  # type: ignore[arg-type]
    )

    results = await retriever.query_async(
        "What was the first issue I had with my new car after its first service?",
        limit=5,
    )

    bundle = results[0]
    assert "issue_candidate=GPS system not functioning correctly" in bundle


def test_source_evidence_score_prioritizes_issue_candidates() -> None:
    """Issue synthesis should rank typed issue evidence above generic car context."""
    query = "What was the first issue I had with my new car after its first service?"
    issue_context = (
        "longmemeval_session_id=answer_1 "
        "I recently had an issue with my car's GPS system on 3/22, "
        "and I had to take it back to the dealership to get it fixed."
    )
    generic_context = (
        "longmemeval_session_id=distractor "
        "I am researching a newer car and washed my current Corolla after a month."
    )

    assert source_evidence_score(query, issue_context) > source_evidence_score(query, generic_context)


def test_source_evidence_score_prioritizes_property_viewing_outcomes() -> None:
    """Property count synthesis should rank rejected/viewed properties over generic home advice."""
    query = "How many properties did I view before making an offer on the townhouse in the Brookside neighborhood?"
    viewed_context = (
        "longmemeval_session_id=answer_a679a86a_4 "
        "I actually fell in love with a 2-bedroom condo on February 15th, "
        "but unfortunately, my offer got rejected on the 17th due to a higher bid."
    )
    generic_context = (
        "longmemeval_session_id=distractor "
        "I can recommend home warranty providers, service fees, and maintenance tasks."
    )

    assert source_evidence_score(query, viewed_context) > source_evidence_score(query, generic_context)


def test_source_evidence_score_prioritizes_accommodation_price_evidence() -> None:
    """Accommodation comparisons should rank nightly lodging prices above generic travel advice."""
    query = "How much more did I spend on accommodations per night in Hawaii compared to Tokyo?"
    lodging_context = (
        "longmemeval_session_id=answer_eaa8e3ef_1 "
        "I've already booked a luxurious resort in Maui that costs over $300 per night."
    )
    generic_context = (
        "longmemeval_session_id=distractor "
        "Tokyo has affordable transportation, walking routes, and street food options."
    )

    assert source_evidence_score(query, lodging_context) > source_evidence_score(query, generic_context)


def test_source_lane_queries_expand_wake_time_offset_evidence() -> None:
    """Wake-time questions should retrieve base wake time plus relative offsets."""
    queries = source_lane_queries(
        "What time do I wake up on Tuesdays and Thursdays?",
        [],
    )

    assert any(
        "wake-up call" in query and "minutes earlier" in query
        for query in queries
    )


def test_source_evidence_score_prioritizes_wake_time_offset_evidence() -> None:
    """Time-offset synthesis should rank wake-time evidence above generic schedule hits."""
    query = "What time do I wake up on Tuesdays and Thursdays?"
    wake_time_context = (
        "longmemeval_session_id=answer_1 I recently started waking up at 7:00 AM, "
        "which is a big improvement from my usual 8:30 AM wake-up call."
    )
    offset_context = (
        "longmemeval_session_id=answer_2 On Tuesdays and Thursdays, I've also "
        "started waking up 15 minutes earlier to meditate."
    )
    schedule_context = (
        "longmemeval_session_id=distractor I usually take Zumba classes on "
        "Tuesdays and Thursdays at 7:00 PM."
    )

    assert source_evidence_score(query, wake_time_context) > source_evidence_score(query, schedule_context)
    assert source_evidence_score(query, offset_context) > source_evidence_score(query, schedule_context)


def test_source_evidence_score_prioritizes_temporal_order_evidence() -> None:
    """Temporal-order synthesis should rank relative event evidence above generic advice."""
    query = "Which event happened first, the road trip to the coast or the arrival of the new prime lens?"
    lens_context = (
        "longmemeval_session_id=answer_1 I recently got a new 50mm f/1.8 prime lens "
        "that I'm still getting used to."
    )
    coastal_context = (
        "longmemeval_session_id=answer_2 I took it with me on the coastal trip "
        "and got some great portraits of my friends."
    )
    generic_context = (
        "longmemeval_session_id=distractor For a hike, a prime lens like the 50mm "
        "can be a good choice for photography."
    )

    assert source_evidence_score(query, lens_context) > source_evidence_score(query, generic_context)
    assert source_evidence_score(query, coastal_context) > source_evidence_score(query, generic_context)


def test_source_evidence_score_prioritizes_bedtime_appointment_evidence() -> None:
    """Bedtime appointment questions should rank clock-time and appointment evidence."""
    query = "What time did I go to bed on the day before I had a doctor's appointment?"
    bedtime_context = (
        "longmemeval_session_id=answer_1 I went to bed at 2 AM after staying up late."
    )
    appointment_context = (
        "longmemeval_session_id=answer_2 I had a doctor's appointment with Dr. Patel the next day."
    )
    generic_context = (
        "longmemeval_session_id=distractor Lifestyle changes can help manage heartburn symptoms."
    )

    assert source_evidence_score(query, bedtime_context) > source_evidence_score(query, generic_context)
    assert source_evidence_score(query, appointment_context) > source_evidence_score(query, generic_context)


def test_source_evidence_score_prioritizes_bedtime_over_sleep_duration_advice() -> None:
    """Direct bedtime questions should prefer actual bedtime facts over sleep-duration advice."""
    query = "What time did I go to bed on the day before I had a doctor's appointment?"
    bedtime_context = (
        "longmemeval_session_id=answer_1 I didn't get to bed until 2 AM last Wednesday, "
        "which made Thursday morning a struggle."
    )
    sleep_advice_context = (
        "longmemeval_session_id=distractor Track your sleep for three weeks and aim "
        "for 7-8 hours of sleep per night."
    )

    assert source_evidence_score(query, bedtime_context) > source_evidence_score(query, sleep_advice_context)


def test_source_lane_queries_expand_streaming_service_recency_evidence() -> None:
    """Streaming-service recency questions should retrieve service and relative-time evidence."""
    queries = source_lane_queries(
        "Which streaming service did I start using most recently?",
        [],
    )

    assert any(
        "streaming service" in query and "last week" in query and "months ago" in query
        for query in queries
    )


def test_source_synthesis_bundle_projects_recent_streaming_service_answer() -> None:
    """Most-recent categorical questions should synthesize from relative-time evidence."""
    bundle = source_synthesis_bundle(
        query="Which streaming service did I start using most recently?",
        source_results=[
            "longmemeval_session_id=answer_1 I started using Hulu a few months ago.",
            "longmemeval_session_id=answer_2 I started using Disney+ last week.",
            "longmemeval_session_id=distractor I changed my internet plan last week.",
        ],
        limit=5,
    )

    assert bundle is not None
    assert "recency_answer=Disney+" in bundle


def test_source_synthesis_bundle_projects_bedtime_before_appointment_answer() -> None:
    """Bedtime-before-appointment questions should synthesize the cited clock time."""
    bundle = source_synthesis_bundle(
        query="What time did I go to bed on the day before I had a doctor's appointment?",
        source_results=[
            "longmemeval_session_id=answer_1 I went to bed at 2 AM after staying up late.",
            "longmemeval_session_id=answer_2 I had a doctor's appointment with Dr. Patel the next day.",
            "longmemeval_session_id=distractor My breakfast helped with heartburn.",
        ],
        limit=5,
    )

    assert bundle is not None
    assert "time_answer=2:00 AM" in bundle


def test_source_synthesis_bundle_prefers_direct_time_over_duration_distractors() -> None:
    """Direct clock-time questions should not be hijacked by duration aggregation."""
    bundle = source_synthesis_bundle(
        query="What time did I go to bed on the day before I had a doctor's appointment?",
        source_results=[
            (
                "longmemeval_session_id=distractor I tracked my sleep for three weeks "
                "and aimed for 7-8 hours of sleep each night."
            ),
            (
                "longmemeval_session_id=answer_1 I didn't get to bed until 2 AM last Wednesday, "
                "which made Thursday morning a struggle."
            ),
            (
                "longmemeval_session_id=answer_2 I had a doctor's appointment at 10 AM last Thursday."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "time_answer=2:00 AM" in bundle
    assert "duration_total_answer=" not in bundle
    assert "minute_total_hours=" not in bundle


def test_source_lane_queries_expand_airline_frequency_evidence() -> None:
    """Airline frequency questions should retrieve carrier-specific flight evidence."""
    queries = source_lane_queries(
        "Which airline did I fly with the most in March and April?",
        [],
    )

    assert any(
        "United Airlines" in query and "flights" in query
        for query in queries
    )


def test_source_lane_queries_expand_camera_temporal_order_evidence() -> None:
    """Camera order questions should retrieve prime-lens and coastal-trip evidence."""
    queries = source_lane_queries(
        "Which event happened first, the road trip to the coast or the arrival of the new prime lens?",
        [],
    )

    assert any(
        "prime lens" in query and "coastal trip" in query
        for query in queries
    )


def test_source_lane_queries_expand_relationship_anniversary_evidence() -> None:
    """Anniversary timing questions should retrieve engagement and anniversary evidence."""
    queries = source_lane_queries(
        "How many months before my anniversary did Rachel get engaged?",
        [],
    )

    assert any(
        "Rachel" in query and "engaged" in query and "anniversary" in query
        for query in queries
    )
    assert any(
        "May 15" in query and "July 22" in query
        for query in queries
    )


def test_source_synthesis_bundle_projects_anniversary_engagement_interval() -> None:
    """Anniversary timing should synthesize the interval from engagement and anniversary evidence."""
    bundle = source_synthesis_bundle(
        query="How many months before my anniversary did Rachel get engaged?",
        source_results=[
            (
                "longmemeval_session_id=answer_aaf71ce2_2 "
                "My close friend Rachel got engaged last month on May 15th."
            ),
            (
                "longmemeval_session_id=answer_aaf71ce2_3 "
                "Our anniversary is coming up on July 22nd and I want to make it special."
            ),
            (
                "longmemeval_session_id=distractor "
                "Rachel can advise on listings after we started working together."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "anniversary_engagement_interval_answer=2 months" in bundle
    assert "answer_aaf71ce2_2" in bundle
    assert "answer_aaf71ce2_3" in bundle


def test_source_lane_queries_expand_doctor_sleep_appointment_evidence() -> None:
    """Doctor sleep questions should retrieve bedtime and appointment evidence."""
    queries = source_lane_queries(
        "What time did I go to bed on the day before I had a doctor's appointment?",
        [],
    )

    assert any(
        "went to bed" in query and "doctor appointment" in query
        for query in queries
    )
    assert any(
        "get to bed until" in query and "last Wednesday" in query
        for query in queries
    )


def test_source_lane_queries_expand_march_doctor_appointment_evidence() -> None:
    """Doctor-count questions should retrieve March appointment evidence."""
    queries = source_lane_queries(
        "How many doctor's appointments did I go to in March?",
        [],
    )

    assert any(
        "March" in query and "appointment" in query and "Dr." in query
        for query in queries
    )


def test_source_lane_queries_expand_february_museum_gallery_evidence() -> None:
    """Museum/gallery count questions should retrieve visit evidence scoped to February."""
    queries = source_lane_queries(
        "How many different museums or galleries did I visit in the month of February?",
        [],
    )

    assert any(
        "February" in query and "museum" in query and "gallery" in query and "visited" in query
        for query in queries
    )
    assert any(
        "Natural History Museum" in query and "The Art Cube" in query
        for query in queries
    )


def test_source_lane_queries_expand_month_scoped_museum_absence_evidence() -> None:
    """Month-scoped venue counts should retrieve nearby venue evidence for absence checks."""
    queries = source_lane_queries(
        "How many different museums or galleries did I visit in December?",
        [],
    )

    assert any(
        "December" in query and "museum" in query and "gallery" in query
        for query in queries
    )


def test_absence_check_bundle_mentions_missing_month_for_venue_counts() -> None:
    """Venue count absence should keep the queried month in the missing target."""
    absence = absence_check_bundle(
        query="How many different museums or galleries did I visit in December?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer-1 "
                "assistant: Look for art galleries or museums that feature abstract art."
            )
        ],
        limit=5,
    )

    assert absence is not None
    assert "not_mentioned_candidate=museums or galleries in december" in absence
    assert "did not mention museums or galleries in december" in absence.casefold()


def test_source_synthesis_bundle_counts_museum_gallery_visits_without_duration_hijack() -> None:
    """Museum/gallery count synthesis should prefer venue visits over unrelated durations."""
    bundle = source_synthesis_bundle(
        query="How many different museums or galleries did I visit in the month of February?",
        source_results=[
            (
                "longmemeval_session_id=answer_1 I took my niece to the Natural History Museum "
                "on 2/8 and she loved the dinosaur exhibit."
            ),
            (
                "longmemeval_session_id=answer_2 I recently saw James Parker's work when I "
                'visited The Art Cube on 2/15.'
            ),
            (
                "longmemeval_session_id=distractor I had a 7-day family road trip in February "
                "with no museum or gallery visit."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "count_answer=2" in bundle
    assert "duration_total_answer=" not in bundle
    assert "answer_1" in bundle
    assert "answer_2" in bundle


def test_source_synthesis_bundle_projects_personal_best_time_answer() -> None:
    """Current-best questions should surface the cited race time instead of generic duration sums."""
    bundle = source_synthesis_bundle(
        query="What was my personal best time in the charity 5K run?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=previous "
                "user: My previous personal best for the charity 5K was 27 minutes and 45 seconds."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=current "
                "user: I set a new personal best time in the charity 5K run: 25 minutes and 50 seconds."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "direct_numeric_answer=25 minutes and 50 seconds (or 25:50)" in bundle


def test_source_synthesis_bundle_projects_current_so_far_count_answer() -> None:
    """Current-progress count questions should use stated totals, not count memory events."""
    bundle = source_synthesis_bundle(
        query="How many tops have I bought from H&M so far?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=older "
                "user: I've already bought three tops from H&M."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=current "
                "user: I've already got five tops from H&M so far, and I'm thinking of getting a few more."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "direct_numeric_answer=five" in bundle


def test_source_synthesis_bundle_projects_latest_earned_market_amount() -> None:
    """Most-recent earning questions should surface the latest cited amount."""
    bundle = source_synthesis_bundle(
        query="How much did I earn at the Downtown Farmers Market on my most recent visit?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=older "
                "longmemeval_session_date=2023/04/01 (Sat) "
                "user: I earned $280 at the Downtown Farmers Market last month."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=current "
                "longmemeval_session_date=2023/05/01 (Mon) "
                "user: On my most recent visit to the Downtown Farmers Market, I earned $420."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "direct_numeric_answer=$420" in bundle


def test_source_synthesis_bundle_projects_current_daily_duration_answer() -> None:
    """Current routine duration questions should pick the current cited value."""
    bundle = source_synthesis_bundle(
        query="How much time do I dedicate to coding exercises each day?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=older "
                "user: I started with one hour of coding exercises each day."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=current "
                "user: Now I dedicate about two hours to coding exercises each day."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "direct_numeric_answer=about two hours" in bundle


def test_source_synthesis_bundle_projects_currency_difference_answer() -> None:
    """Price-comparison questions should expose the computed difference answer."""
    bundle = source_synthesis_bundle(
        query="What is the difference in price between my luxury boots and the similar pair found at the budget store?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=luxury "
                "user: I bought luxury boots for $900."
            ),
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=budget "
                "user: I found a similar pair at the budget store for $150."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "direct_numeric_answer=$750" in bundle


def test_source_synthesis_bundle_projects_page_count_total_without_week_hijack() -> None:
    """Page-count questions should sum pages, not answer with reading duration intervals."""
    bundle = source_synthesis_bundle(
        query="What was the page count of the two novels I finished in January and March?",
        source_results=[
            (
                "longmemeval_session_id=answer_6b9b2b1e_1 "
                "I just finished a 416-page novel, but before that, I read "
                '"The Power" by Naomi Alderman in December, which had 341 pages '
                "and took me around 5 weeks to finish."
            ),
            (
                "longmemeval_session_id=answer_6b9b2b1e_2 "
                'I just finished a historical fiction novel, "The Nightingale" '
                "by Kristin Hannah, which had 440 pages and took me around 3 weeks "
                "to complete."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "page_values=416,440" in bundle or "page_values=440,416" in bundle
    assert "page_total_answer=856" in bundle
    assert "week_interval_answer=" not in bundle


def test_source_synthesis_bundle_projects_total_distance_without_week_hijack() -> None:
    """Distance aggregation should bind mile values to the queried activity."""
    bundle = source_synthesis_bundle(
        query="What is the total distance of the hikes I did on two consecutive weekends?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=weekend-1 "
                "Last weekend I hiked a 5-mile ridge trail with my family."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=weekend-2 "
                "This weekend I did a 3-mile loop trail near the lake."
            ),
            (
                "citation=eventloom://benchmark/events/3#ccc "
                "longmemeval_session_id=distractor "
                "I planned the hikes one week apart."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=distance_total" in bundle
    assert "distance_total_answer=8 miles" in bundle
    assert "week_total=" not in bundle


def test_source_synthesis_bundle_projects_pages_remaining_from_current_and_total() -> None:
    """Title-scoped pages-left questions should subtract current page from total pages."""
    bundle = source_synthesis_bundle(
        query="How many pages do I have left to read in 'The Nightingale'?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=progress "
                "I'm currently on page 250 of 'The Nightingale' by Kristin Hannah."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=total "
                "The Nightingale by Kristin Hannah is 440 pages long."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=pages_remaining" in bundle
    assert "pages_remaining_answer=190" in bundle
    assert "zaxy_absence_check=true" not in bundle


def test_source_synthesis_bundle_projects_query_bound_percentages() -> None:
    """Percentage questions should compute the query-bound ratio, not unrelated durations."""
    bundle = source_synthesis_bundle(
        query="What percentage of packed shoes did I wear on my last trip?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=packed "
                "Since I packed 5 pairs of shoes for my last trip, I had to make room."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=worn "
                "On my last trip I ended up only wearing two shoes: my sneakers and sandals."
            ),
            (
                "citation=eventloom://benchmark/events/3#ccc "
                "longmemeval_session_id=distractor "
                "It was a 5-day trip."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=percentage" in bundle
    assert "percentage_answer=40%" in bundle
    assert "day_total=" not in bundle


def test_source_synthesis_bundle_projects_percentage_from_long_cross_session_evidence() -> None:
    """Percentage operands can be split across long cited sessions."""
    bundle = source_synthesis_bundle(
        query="What percentage of packed shoes did I wear on my last trip?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=worn "
                "I'm planning a 5-day trip and I want to pack light. "
                "By the way, I packed a lot of shoes for my last trip, but I ended up only "
                "wearing two - my sneakers and sandals. "
                + " ".join(f"distractor packing sentence {index}." for index in range(80))
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=packed "
                "I'm planning a 5-day trip to the city and I'm trying to pack lightly. "
                + " ".join(f"packing advice {index}." for index in range(80))
                + " Since I packed 5 pairs of shoes, I had to make sure I had enough space."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=percentage" in bundle
    assert "percentage_answer=40%" in bundle
    assert "day_total=" not in bundle


def test_source_synthesis_bundle_projects_discount_percentage() -> None:
    """Discount questions should compute percentage from original and paid prices."""
    bundle = source_synthesis_bundle(
        query="What percentage discount did I get on the book from my favorite author?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=original "
                "The new release from my favorite author was originally priced at $30."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=paid "
                "I got the book for $24 after a discount."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=percentage" in bundle
    assert "percentage_answer=20%" in bundle


def test_source_synthesis_bundle_projects_discount_ignores_unrelated_budget_prices() -> None:
    """Discount arithmetic should bind currencies to the queried bookstore purchase."""
    bundle = source_synthesis_bundle(
        query="What percentage discount did I get on the book from my favorite author?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=original "
                "I was shopping in-store for a personal gift. "
                "The new release from my favorite author was originally priced at $30."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=paid "
                "My gift budget was $150-$200, and a necklace for my sister was $75. "
                "I saw a sale at my favorite bookstore and got the book for $24 after a discount."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=percentage" in bundle
    assert "percentage_answer=20%" in bundle
    assert "percentage_paid=$24" in bundle


def test_source_synthesis_bundle_promotes_cleaned_shoe_pair_over_duration() -> None:
    """Answer-bearing object spans should outrank unrelated temporal durations."""
    bundle = source_synthesis_bundle(
        query="Which pair of shoes did I clean last month?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=cleaned "
                "I remember I cleaned my white Adidas sneakers last month, "
                "which I'd been meaning to do for weeks."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=distractor "
                "I need to remind my sister to return my spare running shoes that I lent to her a few weeks ago."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=latest_state" in bundle
    assert "latest_state_answer=white Adidas sneakers" in bundle
    assert "duration_total_answer=" not in bundle


def test_source_synthesis_bundle_promotes_ram_upgrade_value() -> None:
    """Hardware upgrade questions should surface compact units like 16GB."""
    bundle = source_synthesis_bundle(
        query="How much RAM did I upgrade my laptop to?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=ram "
                "Before the RAM upgrade to 16GB, I was getting around 6-7 hours of battery life."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=distractor "
                "Now my battery lasts around 4-5 hours after the upgrade."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=latest_state" in bundle
    assert "latest_state_answer=16GB" in bundle
    assert "hour_total=" not in bundle


def test_source_synthesis_bundle_promotes_latest_page_progress_over_count() -> None:
    """Current page-progress spans should outrank generic event counts."""
    bundle = source_synthesis_bundle(
        query="How many pages of 'A Short History of Nearly Everything' have I read so far?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=older "
                "I've been reading \"A Short History of Nearly Everything\" and I'm currently on page 200."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=current "
                "I'm on page 220 now, and I just finished reading about the discovery of DNA structure."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=latest_state" in bundle
    assert "latest_state_answer=220" in bundle
    assert "count_answer_text=" not in bundle


def test_source_synthesis_bundle_promotes_updated_duration_range_over_sum() -> None:
    """Updated duration-state questions should choose the current range, not sum stale and current values."""
    bundle = source_synthesis_bundle(
        query="How many hours have I spent on my abstract ocean sculpture?",
        source_results=[
            (
                "citation=eventloom://benchmark/events/1#aaa "
                "longmemeval_session_id=current "
                "I've already put in 10-12 hours on my abstract ocean sculpture, and it's still a work in progress."
            ),
            (
                "citation=eventloom://benchmark/events/2#bbb "
                "longmemeval_session_id=older "
                "Earlier I had spent 5-6 hours experimenting with ocean-inspired materials."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "candidate_type=latest_state" in bundle
    assert "latest_state_answer=10-12 hours" in bundle
    assert "hour_total=" not in bundle


def test_source_synthesis_bundle_projects_airline_frequency_answer() -> None:
    """Airline frequency questions should synthesize the carrier with most flights."""
    bundle = source_synthesis_bundle(
        query="Which airline did I fly with the most in March and April?",
        source_results=[
            (
                "longmemeval_session_id=answer_1 In March, I took a business trip "
                "to Chicago with United Airlines, flying from my hometown to Chicago "
                "on the 10th and returning on the 12th, with two flights each way."
            ),
            (
                "longmemeval_session_id=answer_2 We flew with American Airlines "
                "from our hometown to Honolulu, and then took a connecting flight to Maui."
            ),
            (
                "longmemeval_session_id=answer_3 I took a direct flight with "
                "Southwest Airlines from my hometown to Las Vegas in March."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "frequency_answer=United Airlines" in bundle


def test_source_synthesis_bundle_projects_parent_temporal_order_answer() -> None:
    """Parent first questions should compare adoption/birth evidence for named people."""
    bundle = source_synthesis_bundle(
        query="Who became a parent first, Rachel or Alex?",
        source_results=[
            (
                "longmemeval_session_id=answer_65600ff6_1 "
                "My sister-in-law, Rachel, is doing great with the twins, Jackson and Julia. "
                "They're growing up so fast!"
            ),
            (
                "longmemeval_session_id=answer_65600ff6_2 "
                "My cousin Alex just adopted a baby girl from China in January."
            ),
            (
                "longmemeval_session_id=answer_65600ff6_1 "
                "Jackson and Julia were born on February 12th."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "parent_order_answer=Alex" in bundle
    assert "answer_65600ff6_1" in bundle
    assert "answer_65600ff6_2" in bundle


def test_source_synthesis_bundle_defers_parent_order_when_named_person_is_absent() -> None:
    """Parent order queries should fall through to absence when one named alternative is missing."""
    bundle = source_synthesis_bundle(
        query="Who became a parent first, Tom or Alex?",
        source_results=[
            (
                "longmemeval_session_id=answer_65600ff6_abs_2 "
                "My cousin Alex just adopted a baby girl from China in January."
            ),
            (
                "longmemeval_session_id=answer_65600ff6_abs_1 "
                "My sister-in-law's twins, Jackson and Julia, were born on February 12th."
            ),
        ],
        limit=5,
    )

    assert bundle is None


def test_absence_check_bundle_handles_missing_parent_order_alternative() -> None:
    """Parent-order absence should cite known parent evidence and the missing person."""
    bundle = absence_check_bundle(
        query="Who became a parent first, Tom or Alex?",
        source_results=[
            (
                "longmemeval_session_id=answer_65600ff6_abs_2 "
                "My cousin Alex just adopted a baby girl from China in January."
            ),
            (
                "longmemeval_session_id=answer_65600ff6_abs_1 "
                "My sister-in-law's twins, Jackson and Julia, were born on February 12th."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "zaxy_absence_check=true" in bundle
    assert "not_mentioned_candidate=tom" in bundle
    assert "answer_65600ff6_abs_1" in bundle
    assert "answer_65600ff6_abs_2" in bundle


def test_absence_check_bundle_ignores_non_parent_name_distractors_for_parent_order() -> None:
    """Unrelated Tom mentions should not suppress missing-parent evidence."""
    bundle = absence_check_bundle(
        query="Who became a parent first, Tom or Alex?",
        source_results=[
            "longmemeval_session_id=distractor Tom attended a charity event last weekend.",
            (
                "longmemeval_session_id=answer_65600ff6_abs_2 "
                "My cousin Alex just adopted a baby girl from China in January."
            ),
            (
                "longmemeval_session_id=answer_65600ff6_abs_1 "
                "My sister-in-law's twins, Jackson and Julia, were born on February 12th."
            ),
        ],
        limit=5,
    )

    assert bundle is not None
    assert "not_mentioned_candidate=tom" in bundle


async def test_zaxy_retriever_uses_source_lane_for_absence_checks() -> None:
    """Mention/absence questions should inspect source evidence."""
    corpus = (
        BenchmarkChunk(
            "answer",
            (
                "citation=eventloom://benchmark/events/1#abc "
                "session_id=answer-1 I mentioned my cat Luna but not a hamster."
            ),
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph context about pet care")]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("Did I mention my hamster?", limit=4)

    assert any("cat Luna" in result for result in results)


async def test_zaxy_retriever_projects_absence_check_bundle() -> None:
    """Absence checks should expose cited no-direct-mention guidance."""
    corpus = (
        BenchmarkChunk(
            "answer",
            (
                "citation=eventloom://benchmark/events/1#abc "
                "session_id=answer-1 I mentioned my cat Luna during the conversation."
            ),
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph context about pet care")]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("Did I mention my hamster?", limit=4)

    bundle = results[0]
    assert "zaxy_absence_check=true" in bundle
    assert "not_mentioned_candidate=hamster" in bundle
    assert "You did not mention this information." in bundle
    assert "cat Luna" in bundle


async def test_zaxy_retriever_prefers_absence_for_missing_parent_order_alternative() -> None:
    """Parent-order questions with a missing named person should not emit duration synthesis."""
    corpus = (
        BenchmarkChunk(
            "alex",
            (
                "citation=eventloom://benchmark/events/1#abc "
                "longmemeval_session_id=answer_65600ff6_abs_2 "
                "My cousin Alex just adopted a baby girl from China in January."
            ),
        ),
        BenchmarkChunk(
            "twins",
            (
                "citation=eventloom://benchmark/events/2#abc "
                "longmemeval_session_id=answer_65600ff6_abs_1 "
                "My sister-in-law's twins, Jackson and Julia, were born on February 12th."
            ),
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph context parent first one week")]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async("Who became a parent first, Tom or Alex?", limit=5)

    bundle = results[0]
    assert "zaxy_absence_check=true" in bundle
    assert "not_mentioned_candidate=tom" in bundle
    assert "The information provided is not enough." in bundle
    assert "week_total=" not in bundle


async def test_zaxy_retriever_prioritizes_graph_evidence_over_lexical_sidecar() -> None:
    """Lexical fusion should not outrank graph evidence when both return hits."""
    corpus = (
        BenchmarkChunk("distractor", "graph-finisher-distractor-0001 completed unrelated task"),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="graph-finisher-0001 completed graph-task-0001")]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async(
        "Which actor completed graph-task-0001?",
        limit=1,
    )

    assert results == ["graph-finisher-0001 completed graph-task-0001"]


async def test_zaxy_retriever_preserves_graph_citations() -> None:
    """Benchmark contexts should retain graph citations for audit reporting."""

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [
                SimpleNamespace(
                    content="graph-finisher-0001 completed graph-task-0001",
                    citation="eventloom://benchmark/events/3#abc123",
                )
            ]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
    )

    results = await retriever.query_async("Which actor completed graph-task-0001?", limit=1)

    assert results == [
        "graph-finisher-0001 completed graph-task-0001\n"
        "citation=eventloom://benchmark/events/3#abc123"
    ]


async def test_zaxy_retriever_does_not_backfill_temporal_queries_with_raw_history() -> None:
    """Raw Eventloom lexical context should not reintroduce superseded temporal facts."""
    corpus = (
        BenchmarkChunk(
            "history",
            (
                "workspace=workspace-alpha-0000 "
                "workspace=workspace-beta-0000 "
                "workspace=workspace-gamma-0000"
            ),
        ),
    )

    class FakeRouter:
        async def query(
            self,
            query: str,
            temporal_point: str | None = None,
            limit: int | None = None,
            embedding: list[float] | None = None,
        ) -> list[SimpleNamespace]:
            del query, temporal_point, limit, embedding
            return [SimpleNamespace(content="workspace=workspace-alpha-0000 user-0000")]

    retriever = ZaxyRetriever(
        FakeRouter(),  # type: ignore[arg-type]
        HashEmbeddingProvider(dimension=8),
        lexical_retriever=BM25Retriever(corpus),
    )

    results = await retriever.query_async(
        "What workspace preference was active for user-0000 at 2024-03-01T00:00:00Z?",
        temporal_point="2024-03-01T00:00:00Z",
        limit=5,
    )

    assert results == ["workspace=workspace-alpha-0000 user-0000"]


def test_live_benchmark_outputs_machine_and_markdown_reports(tmp_path: Path) -> None:
    """Reports should be usable for automation and public documentation."""
    log = build_competitive_event_log(tmp_path / "bench.jsonl")
    corpus = corpus_from_event_log(log)
    provider = HashEmbeddingProvider(dimension=64)
    report = benchmark_retrievers(
        {
            "md": MarkdownRetriever(corpus),
            "vector": VectorRetriever(corpus, provider),
        },
        competitive_cases(),
        runs=1,
        limit=3,
    )

    output = write_benchmark_report(report, tmp_path / "report")
    markdown = report_to_markdown(report)
    payload = json.loads(output.json_path.read_text(encoding="utf-8"))

    assert output.json_path.name == "live-benchmark.json"
    assert output.markdown_path.name == "live-benchmark.md"
    assert payload["summaries"][0]["backend"] in {"md", "vector"}
    assert payload["workload"]["version"] == "ad-hoc"
    assert "| Backend | Mean score | Identity recall | Source recall | Citation coverage |" in markdown
    assert "Approx tokens" in markdown


def _report_with_zaxy_summary(
    *,
    mean_score: float,
    citation_coverage: float,
    p95_ms: float,
    p99_ms: float,
    answer_recall_at_5: float | None = None,
    recall_at_5: float | None = None,
) -> BenchmarkReport:
    return BenchmarkReport(
        generated_at="2026-05-11T00:00:00Z",
        embedding_provider="hash:1536",
        runs=(),
        summaries=(
            BenchmarkSummary(
                backend="zaxy",
                case_count=650,
                runs=1,
                mean_score=mean_score,
                latency_ms_mean=100.0,
                latency_ms_p50=40.0,
                latency_ms_p95=p95_ms,
                latency_ms_p99=p99_ms,
                mean_returned_bytes=950.0,
                mean_approx_tokens=236.0,
                mean_citation_coverage=citation_coverage,
                mean_recall_at_5=recall_at_5,
                mean_answer_recall_at_5=answer_recall_at_5,
            ),
        ),
    )


def test_benchmark_compare_flags_latency_regression() -> None:
    """Benchmark guardrails should catch tail-latency regressions."""
    baseline = _report_with_zaxy_summary(
        mean_score=1.0,
        citation_coverage=1.0,
        p95_ms=250.0,
        p99_ms=300.0,
    )
    candidate = _report_with_zaxy_summary(
        mean_score=1.0,
        citation_coverage=1.0,
        p95_ms=900.0,
        p99_ms=1200.0,
    )

    comparison = compare_benchmark_reports(
        baseline,
        candidate,
        backend="zaxy",
        max_p95_ms=500.0,
        max_p99_ms=750.0,
        max_latency_regression_ratio=0.5,
    )

    assert comparison.passed is False
    assert any(check.name == "p95_latency_regression" for check in comparison.checks)
    assert any(check.name == "p99_latency_budget" for check in comparison.checks)
    markdown = format_benchmark_comparison(comparison)
    assert "FAIL" in markdown
    assert "p95_latency_regression" in markdown


def test_benchmark_compare_passes_latency_improvement() -> None:
    """Improved latency with stable quality should satisfy beta guardrails."""
    baseline = _report_with_zaxy_summary(
        mean_score=1.0,
        citation_coverage=1.0,
        p95_ms=2844.62,
        p99_ms=3031.60,
    )
    candidate = _report_with_zaxy_summary(
        mean_score=1.0,
        citation_coverage=1.0,
        p95_ms=257.58,
        p99_ms=274.41,
    )

    comparison = compare_benchmark_reports(
        baseline,
        candidate,
        backend="zaxy",
        min_mean_score=0.95,
        min_citation_coverage=0.95,
        max_p95_ms=500.0,
        max_p99_ms=750.0,
    )

    assert comparison.passed is True
    assert all(check.passed for check in comparison.checks)
    assert "PASS" in format_benchmark_comparison(comparison)


def test_benchmark_compare_enforces_answer_and_retrieval_recall_floors() -> None:
    """Benchmark guardrails should treat Answer@5 and R@5 as beta quality floors."""
    candidate = _report_with_zaxy_summary(
        mean_score=0.99,
        citation_coverage=1.0,
        p95_ms=250.0,
        p99_ms=300.0,
        answer_recall_at_5=0.94,
        recall_at_5=0.98,
    )

    comparison = compare_benchmark_reports(
        None,
        candidate,
        backend="zaxy",
        min_mean_score=0.95,
        min_citation_coverage=0.95,
        min_answer_recall_at_5=0.95,
        min_recall_at_5=0.99,
        max_p95_ms=500.0,
        max_p99_ms=750.0,
    )

    assert comparison.passed is False
    assert any(
        check.name == "answer_recall_at_5_floor" and check.passed is False
        for check in comparison.checks
    )
    assert any(
        check.name == "recall_at_5_floor" and check.passed is False
        for check in comparison.checks
    )
    markdown = format_benchmark_comparison(comparison)
    assert "answer_recall_at_5_floor" in markdown
    assert "recall_at_5_floor" in markdown


def test_frozen_statistical_workload_has_stable_identity(tmp_path: Path) -> None:
    """Frozen workload metadata should make benchmark reports reproducible."""
    first_log, first_cases, first_workload = build_frozen_statistical_workload(tmp_path / "first.jsonl")
    second_log, second_cases, second_workload = build_frozen_statistical_workload(tmp_path / "second.jsonl")

    assert first_workload.version == FROZEN_WORKLOAD_VERSION
    assert first_workload.subjects == FROZEN_WORKLOAD_SUBJECTS
    assert first_workload.event_count == 500
    assert first_workload.case_count == 300
    assert first_workload.sha256 == second_workload.sha256
    assert first_workload.sha256 == workload_fingerprint(first_log, first_cases, FROZEN_WORKLOAD_VERSION)
    assert first_workload == second_workload


def test_benchmark_suite_workload_covers_memory_document_and_transcript_lanes(
    tmp_path: Path,
) -> None:
    """The suite workload should be broad enough to represent production context."""
    eventlog, cases, workload = build_benchmark_suite_workload(
        tmp_path / "suite.jsonl",
        subjects=4,
        documents=5,
        sessions=3,
    )

    assert workload.version == SUITE_WORKLOAD_VERSION
    assert workload.subjects == 4
    assert workload.documents == 5
    assert workload.sessions == 3
    assert workload.lanes == (
        "current",
        "temporal",
        "traversal",
        "document",
        "transcript",
        "mixed",
    )
    assert workload.event_count == len(eventlog.read_all())
    assert workload.case_count == len(cases)
    assert workload.event_count == 20 + 5 + 6
    assert workload.case_count == 12 + 5 + 3 + 3
    assert {case.category for case in cases} == set(workload.lanes)
    assert any(case.name.startswith("document-source-") for case in cases)
    assert any(case.name.startswith("session-decision-") for case in cases)
    assert any(case.name.startswith("mixed-release-") for case in cases)


def test_benchmark_suite_workload_has_stable_identity(tmp_path: Path) -> None:
    """Suite fingerprints should be stable across output paths."""
    first_log, first_cases, first_workload = build_benchmark_suite_workload(
        tmp_path / "first.jsonl",
        subjects=3,
        documents=4,
        sessions=2,
    )
    second_log, second_cases, second_workload = build_benchmark_suite_workload(
        tmp_path / "second.jsonl",
        subjects=3,
        documents=4,
        sessions=2,
    )

    assert first_workload == second_workload
    assert first_workload.sha256 == workload_fingerprint(
        first_log,
        first_cases,
        SUITE_WORKLOAD_VERSION,
    )
    assert first_workload.sha256 == workload_fingerprint(
        second_log,
        second_cases,
        SUITE_WORKLOAD_VERSION,
    )


def test_consolidation_workload_targets_identity_collapse(tmp_path: Path) -> None:
    """The consolidation workload should isolate identity-preservation failures."""
    eventlog, cases, workload = build_consolidation_collapse_workload(
        tmp_path / "collapse.jsonl",
        identities=8,
    )

    assert workload.version == CONSOLIDATION_WORKLOAD_VERSION
    assert workload.event_count == 8
    assert workload.case_count == 8
    assert workload.documents == 8
    assert workload.lanes == ("consolidation",)
    assert {case.category for case in cases} == {"consolidation"}
    assert all(case.identity_terms for case in cases)
    assert any("identity-code-0007" in case.identity_terms for case in cases)
    assert workload.sha256 == workload_fingerprint(
        eventlog,
        cases,
        CONSOLIDATION_WORKLOAD_VERSION,
    )


def test_identity_recall_exposes_centroid_consolidation_collapse(
    tmp_path: Path,
) -> None:
    """Centroid-style consolidation can score as relevant while losing exact identities."""
    eventlog, cases, workload = build_consolidation_collapse_workload(
        tmp_path / "collapse.jsonl",
        identities=6,
    )
    corpus = corpus_from_event_log(eventlog)
    provider = HashEmbeddingProvider(dimension=64)
    report = benchmark_retrievers(
        {
            "md": MarkdownRetriever(corpus),
            "centroid": CentroidConsolidationRetriever(corpus, provider),
        },
        cases,
        runs=1,
        limit=8,
        workload=workload,
    )

    summaries = {summary.backend: summary for summary in report.summaries}
    assert summaries["md"].mean_identity_recall == 1.0
    assert summaries["centroid"].mean_identity_recall is not None
    assert summaries["centroid"].mean_identity_recall < summaries["md"].mean_identity_recall
    assert any(run.missing_identities for run in report.runs if run.backend == "centroid")
    assert "Identity recall" in report_to_markdown(report)


def test_suite_report_renders_workload_dimensions(tmp_path: Path) -> None:
    """Reports should disclose suite dimensions for reproducibility."""
    eventlog, cases, workload = build_benchmark_suite_workload(
        tmp_path / "suite.jsonl",
        subjects=2,
        documents=3,
        sessions=2,
    )
    corpus = corpus_from_event_log(eventlog)
    report = benchmark_retrievers(
        {"md": MarkdownRetriever(corpus)},
        cases,
        runs=1,
        limit=4,
        workload=workload,
    )

    markdown = report_to_markdown(report)

    assert f"Workload: `{SUITE_WORKLOAD_VERSION}`" in markdown
    assert "- Documents: `3`" in markdown
    assert "- Sessions: `2`" in markdown
    assert "- Lanes: `current, temporal, traversal, document, transcript, mixed`" in markdown


def test_cached_embedding_provider_reuses_repeated_text() -> None:
    """Benchmark runs should not repeat hosted embedding calls for identical text."""
    class CountingProvider:
        dimension = 2

        def __init__(self) -> None:
            self.calls = 0

        def embed(self, text: str) -> list[float]:
            self.calls += 1
            return [float(len(text)), 1.0]

    inner = CountingProvider()
    provider = CachedEmbeddingProvider(inner)

    assert provider.embed("same") == [4.0, 1.0]
    assert provider.embed("same") == [4.0, 1.0]
    assert provider.embed("other") == [5.0, 1.0]
    assert provider.cache_size == 2
    assert inner.calls == 2


def test_source_lane_retriever_uses_semantic_candidates_when_bm25_misses() -> None:
    """Verbatim source recall should not be limited to lexical overlap."""
    corpus = (
        BenchmarkChunk("target", "I met with my primary care physician yesterday."),
        BenchmarkChunk("distractor", "I bought a lamp for the office yesterday."),
    )

    class FakeProvider:
        dimension = 2

        def embed(self, text: str) -> list[float]:
            lowered = text.casefold()
            if "doctor" in lowered or "physician" in lowered:
                return [1.0, 0.0]
            return [0.0, 1.0]

    bm25_results = BM25Retriever(corpus).query("doctor?", limit=2)
    source_results = _build_source_lane_retriever(corpus, FakeProvider()).query(
        "doctor?",
        limit=2,
    )

    assert not any("physician" in result for result in bm25_results)
    assert "primary care physician" in source_results[0]


def test_source_lane_retriever_keeps_hash_provider_lexical_by_default() -> None:
    """Hash embeddings should not activate experimental frontier evidence by default."""
    corpus = (
        BenchmarkChunk("target", "I met with my primary care physician yesterday."),
        BenchmarkChunk("distractor", "I bought a lamp for the office yesterday."),
    )

    source_results = _build_source_lane_retriever(
        corpus,
        HashEmbeddingProvider(dimension=16),
    ).query("doctor?", limit=2)

    assert source_results == []


def test_source_lane_retriever_keeps_cached_hash_provider_lexical() -> None:
    """Benchmark caching should preserve hash-provider capability detection."""
    corpus = (
        BenchmarkChunk("target", "I met with my primary care physician yesterday."),
        BenchmarkChunk("distractor", "I bought a lamp for the office yesterday."),
    )

    retriever = _build_source_lane_retriever(
        corpus,
        CachedEmbeddingProvider(HashEmbeddingProvider(dimension=8)),
    )

    assert type(retriever) is BM25Retriever


def test_evidence_frontier_prefers_first_person_evidence_over_topical_overlap() -> None:
    """Frontier retrieval should rank answerable memory evidence above topical text."""
    corpus = (
        BenchmarkChunk(
            "distractor",
            "This document compares film festival distribution strategies and movie marketing.",
        ),
        BenchmarkChunk(
            "answer",
            "longmemeval_salient_memory_turn=true session_id=answer I went to AFI Fest in LA.",
        ),
    )

    results = EvidenceFrontierRetriever(corpus).query(
        "How many movie festivals did I attend?",
        limit=1,
    )

    assert results == [corpus[1].text]


def test_evidence_frontier_recovers_model_kit_action_paraphrases() -> None:
    """Frontier retrieval should search how memories are written, not only query wording."""
    corpus = (
        BenchmarkChunk("distractor", "model kit manufacturers and custom kit maker advice"),
        BenchmarkChunk(
            "answer",
            "longmemeval_salient_memory_turn=true session_id=answer "
            "I've recently finished a simple Revell F-15 Eagle kit.",
        ),
    )

    results = EvidenceFrontierRetriever(corpus).query(
        "How many model kits have I worked on or bought?",
        limit=1,
    )

    assert results == [corpus[1].text]


def test_live_benchmark_script_help_mentions_frozen_workload() -> None:
    script = Path("scripts/live-benchmark.sh").read_text(encoding="utf-8")

    assert "--workload fixture|statistical|frozen|suite|consolidation" in script


def test_live_benchmark_script_dry_run_shows_backend_specific_args() -> None:
    embedded = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--workload", "frozen"],
        check=True,
        capture_output=True,
        text=True,
    )

    assert "--projection-backend embedded" in embedded.stdout
    assert "--embedding-provider hash" in embedded.stdout
    assert "--neo4j-uri" not in embedded.stdout
    assert "--pggraph-dsn" not in embedded.stdout
    assert not embedded.stdout.endswith(" \n")

    comparison_options = subprocess.run(
        [
            "scripts/live-benchmark.sh",
            "--dry-run",
            "--baseline-backends",
            "bm25",
            "--zaxy-backend",
            "checkout",
            "--external-results",
            "external.json",
            "--reuse-projection",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    assert "--baseline-backends bm25" in comparison_options.stdout
    assert "--zaxy-backend checkout" in comparison_options.stdout
    assert "--external-results external.json" in comparison_options.stdout
    assert "--reuse-projection" in comparison_options.stdout

    local_http = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--embedding-provider", "local-http"],
        check=True,
        capture_output=True,
        text=True,
    )

    assert "--embedding-provider local-http" in local_http.stdout

    sentence_transformers = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--embedding-provider", "sentence-transformers"],
        check=True,
        capture_output=True,
        text=True,
    )

    assert "--embedding-provider sentence-transformers" in sentence_transformers.stdout

    neo4j = subprocess.run(
        [
            "scripts/live-benchmark.sh",
            "--dry-run",
            "--projection-backend",
            "neo4j",
            "--neo4j-uri",
            "bolt://127.0.0.1:17687",
            "--neo4j-user",
            "benchmark",
            "--neo4j-password",
            "secret with spaces",
            "--workload",
            "frozen",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    assert "--neo4j-uri bolt://127.0.0.1:17687" in neo4j.stdout
    assert "--neo4j-user benchmark" in neo4j.stdout
    assert "--neo4j-password secret\\ with\\ spaces" in neo4j.stdout

    pggraph = subprocess.run(
        [
            "scripts/live-benchmark.sh",
            "--dry-run",
            "--projection-backend",
            "pggraph",
            "--pggraph-dsn",
            "postgresql://postgres:postgres@localhost:5432/zaxy",
            "--workload",
            "frozen",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    assert "--pggraph-dsn postgresql://postgres:postgres@localhost:5432/zaxy" in pggraph.stdout

    invalid_backend = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--projection-backend", "banana"],
        capture_output=True,
        text=True,
    )

    assert invalid_backend.returncode == 2
    assert "Invalid projection backend: banana" in invalid_backend.stderr
    assert "expected one of: embedded, neo4j, pggraph, latticedb" in invalid_backend.stderr

    invalid_embedding = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--embedding-provider", "banana"],
        capture_output=True,
        text=True,
    )

    assert invalid_embedding.returncode == 2
    assert "Invalid embedding provider: banana" in invalid_embedding.stderr
    assert "expected one of: openai, hash, local-http, sentence-transformers" in invalid_embedding.stderr

    invalid_workload = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--workload", "banana"],
        capture_output=True,
        text=True,
    )

    assert invalid_workload.returncode == 2
    assert "Invalid workload: banana" in invalid_workload.stderr
    assert "expected one of: fixture, statistical, frozen, suite" in invalid_workload.stderr

    invalid_runs = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--runs", "0"],
        capture_output=True,
        text=True,
    )

    assert invalid_runs.returncode == 2
    assert "Invalid --runs: 0; expected a positive integer" in invalid_runs.stderr

    invalid_limit = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--limit", "many"],
        capture_output=True,
        text=True,
    )

    assert invalid_limit.returncode == 2
    assert "Invalid --limit: many; expected a positive integer" in invalid_limit.stderr

    invalid_subjects = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--subjects", "-5"],
        capture_output=True,
        text=True,
    )

    assert invalid_subjects.returncode == 2
    assert "Invalid --subjects: -5; expected a positive integer" in invalid_subjects.stderr

    missing_runs = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--runs"],
        capture_output=True,
        text=True,
    )

    assert missing_runs.returncode == 2
    assert "Missing value for --runs" in missing_runs.stderr

    missing_password = subprocess.run(
        ["scripts/live-benchmark.sh", "--dry-run", "--neo4j-password"],
        capture_output=True,
        text=True,
    )

    assert missing_password.returncode == 2
    assert "Missing value for --neo4j-password" in missing_password.stderr
