# SPDX-License-Identifier: AGPL-3.0-only
# Copyright (c) 2026 sol pbc

"""Tests for the unified journal index."""

import hashlib
import json
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import Mock

import pytest

from solstone.convey.chat_stream import append_chat_event
from solstone.think.indexer import sanitize_fts_query
from solstone.think.indexer.journal import (
    _build_where_clause,
    extract_temporal_references,
    get_journal_index,
    search_counts,
    search_journal,
)


class TestSanitizeFtsQuery:
    """Tests for FTS5 query sanitization."""

    def test_simple_words(self):
        """Simple words get NEAR proximity formulation."""
        assert sanitize_fts_query("foo bar baz") == (
            "NEAR(foo bar baz, 10) OR (foo AND bar AND baz)",
            None,
            None,
        )

    def test_preserves_or_operator(self):
        """OR operator is preserved."""
        assert sanitize_fts_query("foo OR bar") == ("foo OR bar", None, None)

    def test_preserves_and_operator(self):
        """AND operator is preserved."""
        assert sanitize_fts_query("foo AND bar") == ("foo AND bar", None, None)

    def test_preserves_not_operator(self):
        """NOT operator is preserved."""
        assert sanitize_fts_query("foo NOT bar") == ("foo NOT bar", None, None)

    def test_preserves_asterisk_prefix_match(self):
        """Asterisk for prefix matching is preserved."""
        assert sanitize_fts_query("test*") == ("test*", None, None)

    def test_preserves_quoted_phrases(self):
        """Quoted phrases are preserved."""
        assert sanitize_fts_query('"public benefit"') == (
            '"public benefit"',
            None,
            None,
        )

    def test_complex_query_with_or_and_quotes(self):
        """Complex query with OR and quoted phrases."""
        result = sanitize_fts_query('solstone OR pbc OR "public benefit"')
        assert result == ('solstone OR pbc OR "public benefit"', None, None)

    def test_dot_replaced_with_space(self):
        """Dots are replaced with spaces."""
        assert sanitize_fts_query("config.json") == (
            "NEAR(config json, 10) OR (config AND json)",
            None,
            None,
        )

    def test_colon_replaced_with_space(self):
        """Colons are replaced with spaces."""
        assert sanitize_fts_query("foo:bar") == (
            "NEAR(foo bar, 10) OR (foo AND bar)",
            None,
            None,
        )

    def test_special_chars_replaced_with_space(self):
        """Various special characters are replaced with spaces."""
        assert sanitize_fts_query("a@b#c$d") == (
            "NEAR(a b c d, 10) OR (a AND b AND c AND d)",
            None,
            None,
        )

    def test_preserves_apostrophe(self):
        """Apostrophes in contractions are preserved."""
        assert sanitize_fts_query("what's up") == (
            'NEAR("what\'s" up, 10) OR ("what\'s" AND up)',
            None,
            None,
        )

    def test_unbalanced_quote_removed(self):
        """Unbalanced quotes are removed entirely."""
        assert sanitize_fts_query('"unbalanced') == ("unbalanced", None, None)

    def test_unbalanced_quote_removes_all(self):
        """When quotes are unbalanced, all quotes are removed."""
        assert sanitize_fts_query('foo "bar" baz "qux') == (
            "NEAR(foo bar baz qux, 10) OR (foo AND bar AND baz AND qux)",
            None,
            None,
        )

    def test_balanced_quotes_preserved(self):
        """Balanced quotes are kept."""
        assert sanitize_fts_query('"foo" "bar"') == ('"foo" "bar"', None, None)

    def test_near_two_words(self):
        """Two plain words get NEAR proximity formulation."""
        assert sanitize_fts_query("git commit") == (
            "NEAR(git commit, 10) OR (git AND commit)",
            None,
            None,
        )

    def test_near_three_words(self):
        """Three plain words get NEAR proximity formulation."""
        assert sanitize_fts_query("meeting with Alice") == (
            "NEAR(meeting with Alice, 10) OR (meeting AND with AND Alice)",
            None,
            None,
        )

    def test_near_with_prefix(self):
        """Prefix matching works within NEAR formulation."""
        assert sanitize_fts_query("test* foo") == (
            "NEAR(test* foo, 10) OR (test* AND foo)",
            None,
            None,
        )

    def test_single_word_no_near(self):
        """Single word does not get NEAR treatment."""
        assert sanitize_fts_query("hello") == ("hello", None, None)

    def test_empty_query(self):
        """Empty query returns empty string."""
        assert sanitize_fts_query("") == ("", None, None)

    def test_near_normalizes_whitespace(self):
        """Extra whitespace in input is normalized in NEAR output."""
        assert sanitize_fts_query("foo  bar") == (
            "NEAR(foo bar, 10) OR (foo AND bar)",
            None,
            None,
        )


def test_build_where_clause_binds_match_param():
    """FTS MATCH query text is bound as the first positional parameter."""
    where_clause, params = _build_where_clause("it's")

    assert where_clause == "chunks MATCH ?"
    assert params[0] == '"it\'s"'


class TestTemporalExtraction:
    """Tests for temporal date extraction from queries."""

    REF = datetime(2024, 1, 15)  # Monday

    def test_yesterday(self):
        result = sanitize_fts_query("meeting yesterday", self.REF)
        assert result == ("meeting", "20240114", "20240114")

    def test_today(self):
        result = sanitize_fts_query("meeting today", self.REF)
        assert result == ("meeting", "20240115", "20240115")

    def test_last_week(self):
        result = sanitize_fts_query("meeting last week", self.REF)
        assert result == ("meeting", "20240108", "20240114")

    def test_this_week(self):
        result = sanitize_fts_query("meeting this week", self.REF)
        assert result == ("meeting", "20240115", "20240121")

    def test_last_month(self):
        result = sanitize_fts_query("meeting last month", self.REF)
        assert result == ("meeting", "20231201", "20231231")

    def test_this_month(self):
        result = sanitize_fts_query("meeting this month", self.REF)
        assert result == ("meeting", "20240101", "20240131")

    def test_last_monday(self):
        # ref is Monday, so "last monday" = 7 days ago
        result = sanitize_fts_query("meeting last Monday", self.REF)
        assert result == ("meeting", "20240108", "20240108")

    def test_last_tuesday(self):
        result = sanitize_fts_query("meeting last Tuesday", self.REF)
        assert result == ("meeting", "20240109", "20240109")

    def test_last_wednesday(self):
        result = sanitize_fts_query("meeting last Wednesday", self.REF)
        assert result == ("meeting", "20240110", "20240110")

    def test_last_thursday(self):
        result = sanitize_fts_query("meeting last Thursday", self.REF)
        assert result == ("meeting", "20240111", "20240111")

    def test_last_friday(self):
        result = sanitize_fts_query("meeting last Friday", self.REF)
        assert result == ("meeting", "20240112", "20240112")

    def test_last_saturday(self):
        result = sanitize_fts_query("meeting last Saturday", self.REF)
        assert result == ("meeting", "20240113", "20240113")

    def test_last_sunday(self):
        result = sanitize_fts_query("meeting last Sunday", self.REF)
        assert result == ("meeting", "20240114", "20240114")

    def test_over_the_weekend(self):
        result = sanitize_fts_query("meeting over the weekend", self.REF)
        assert result == ("meeting", "20240113", "20240114")

    def test_on_the_weekend(self):
        result = sanitize_fts_query("meeting on the weekend", self.REF)
        assert result == ("meeting", "20240113", "20240114")

    def test_case_insensitive(self):
        result = sanitize_fts_query("meeting Last Monday", self.REF)
        assert result == ("meeting", "20240108", "20240108")

    def test_temporal_only_query(self):
        """Pure temporal query produces empty FTS string + date filter."""
        result = sanitize_fts_query("yesterday", self.REF)
        assert result == ("", "20240114", "20240114")

    def test_no_temporal_reference(self):
        """Query without temporal words returns None dates."""
        result = sanitize_fts_query("machine learning", self.REF)
        assert result == (
            "NEAR(machine learning, 10) OR (machine AND learning)",
            None,
            None,
        )

    def test_temporal_at_start(self):
        result = sanitize_fts_query("yesterday meeting with Alice", self.REF)
        assert result == (
            "NEAR(meeting with Alice, 10) OR (meeting AND with AND Alice)",
            "20240114",
            "20240114",
        )

    def test_temporal_in_middle(self):
        result = sanitize_fts_query("project last week update", self.REF)
        assert result == (
            "NEAR(project update, 10) OR (project AND update)",
            "20240108",
            "20240114",
        )

    def test_quoted_temporal_not_extracted(self):
        """Temporal words inside quotes are not extracted."""
        result = sanitize_fts_query('"last week" meeting', self.REF)
        assert result == ('"last week" meeting', None, None)

    def test_multiple_temporal_first_wins(self):
        """When multiple temporal references exist, first one wins."""
        result = sanitize_fts_query("yesterday last week meeting", self.REF)
        assert result == (
            "NEAR(last week meeting, 10) OR (last AND week AND meeting)",
            "20240114",
            "20240114",
        )

    def test_extract_temporal_references_directly(self):
        """Test extract_temporal_references for the cleaned query."""
        cleaned, day_from, day_to = extract_temporal_references(
            "meeting last week", self.REF
        )
        assert cleaned == "meeting"
        assert day_from == "20240108"
        assert day_to == "20240114"


@pytest.fixture
def journal_fixture(tmp_path, monkeypatch):
    """Create a temporary journal with test data."""
    journal = tmp_path
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))

    # Create daily insight
    day = journal / "chronicle" / "20240101"
    day.mkdir(parents=True)
    agents_dir = day / "talents"
    agents_dir.mkdir()
    (agents_dir / "flow.md").write_text("# Flow Summary\n\nWorked on project alpha.\n")

    # Create segment with agent output
    stream_dir = day / "default"
    stream_dir.mkdir()
    segment = stream_dir / "100000_300"
    segment.mkdir()
    (segment / "talents").mkdir()
    (segment / "talents" / "screen.md").write_text(
        "# Screen Summary\n\nViewed documentation.\n"
    )
    # Add stream.json for segment stream metadata
    from solstone.think.streams import write_segment_stream

    write_segment_stream(str(segment), "default", None, None, 1)
    # Add second agent file for cross-file segment testing
    (segment / "talents" / "activity.md").write_text(
        "# Activity Summary\n\nMet with Scott Ward about Acme deal.\n"
    )

    # Create evening segment for time_bucket testing
    evening_segment = stream_dir / "200000_300"
    evening_segment.mkdir()
    (evening_segment / "talents").mkdir()
    (evening_segment / "talents" / "screen.md").write_text(
        "# Evening Screen\n\nReviewed evening reports.\n"
    )
    write_segment_stream(str(evening_segment), "default", None, None, 1)

    # Create facet events
    events_dir = journal / "facets" / "work" / "events"
    events_dir.mkdir(parents=True)
    event = {
        "type": "meeting",
        "start": "09:00:00",
        "end": "09:30:00",
        "title": "Standup",
        "summary": "Daily sync meeting",
        "facet": "work",
        "agent": "meetings",
        "occurred": True,
    }
    (events_dir / "20240101.jsonl").write_text(json.dumps(event))

    # Create facet entities
    entities_dir = journal / "facets" / "work" / "entities"
    entities_dir.mkdir(parents=True)
    entity = {
        "name": "Project Alpha",
        "type": "project",
        "description": "Main project",
    }
    (entities_dir / "20240101.jsonl").write_text(json.dumps(entity))

    # Create facet news
    news_dir = journal / "facets" / "work" / "news"
    news_dir.mkdir(parents=True)
    (news_dir / "20240101.md").write_text(
        "# News\n\nImportant update about the project.\n"
    )

    return journal


@pytest.fixture
def relax_fixture(tmp_path, monkeypatch):
    """Create a focused journal for search relaxation and rerank tests."""
    journal = tmp_path
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))

    import solstone.think.utils as think_utils

    think_utils._journal_path_cache = None

    from solstone.think.indexer.journal import scan_journal
    from solstone.think.streams import write_segment_stream

    day = journal / "chronicle" / "20240101"
    day.mkdir(parents=True)

    segment = day / "default" / "100000_300"
    talents = segment / "talents"
    talents.mkdir(parents=True)
    (talents / "activity.md").write_text(
        "# Activity\n\nMet with Scott Ward about Acme deal.\n",
        encoding="utf-8",
    )
    (talents / "screen.md").write_text(
        "# Screen\n\nViewed documentation.\n",
        encoding="utf-8",
    )
    write_segment_stream(str(segment), "default", None, None, 1)

    segment2 = day / "default" / "110000_300"
    talents2 = segment2 / "talents"
    talents2.mkdir(parents=True)
    (talents2 / "a_first.md").write_text(
        "# Zephyr\n\nMongoose census complete.\n",
        encoding="utf-8",
    )
    (talents2 / "b_second.md").write_text(
        "Quokka sightings logged.\n",
        encoding="utf-8",
    )
    write_segment_stream(str(segment2), "default", None, None, 1)

    day_talents = day / "talents"
    day_talents.mkdir()
    (day_talents / "apostrophe.md").write_text(
        "# Apostrophe\n\nOwner's decision log mentions budget.\n",
        encoding="utf-8",
    )
    (day_talents / "android.md").write_text(
        "# Android\n\nANDROID migration checklist.\n",
        encoding="utf-8",
    )
    (day_talents / "oracle.md").write_text(
        "# Oracle\n\nORACLE support matrix.\n",
        encoding="utf-8",
    )
    (day_talents / "rerank_a.md").write_text(
        "# Rerank A\n\nsharedrank alpha lowmark.\n",
        encoding="utf-8",
    )
    (day_talents / "rerank_b.md").write_text(
        "# Rerank B\n\nsharedrank beta midmark.\n",
        encoding="utf-8",
    )
    (day_talents / "rerank_c.md").write_text(
        "# Rerank C\n\nsharedrank gamma highmark.\n",
        encoding="utf-8",
    )
    (day_talents / "single.md").write_text(
        "# Single\n\nsinglematch only here.\n",
        encoding="utf-8",
    )

    events_dir = journal / "facets" / "work" / "events"
    events_dir.mkdir(parents=True)
    event = {
        "type": "meeting",
        "start": "09:00:00",
        "end": "09:30:00",
        "title": "Standup",
        "summary": "Daily sync meeting",
        "facet": "work",
        "occurred": True,
    }
    (events_dir / "20240101.jsonl").write_text(json.dumps(event), encoding="utf-8")

    yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
    yesterday_talents = journal / "chronicle" / yesterday / "talents"
    yesterday_talents.mkdir(parents=True)
    (yesterday_talents / "temporal.md").write_text(
        "# Temporal\n\nCalypso ladder recall happened.\n",
        encoding="utf-8",
    )

    scan_journal(str(journal), full=True)
    return {"journal": journal, "yesterday": yesterday}


def test_scan_journal(journal_fixture):
    """Test scanning journal creates index."""
    from solstone.think.indexer.journal import scan_journal

    changed = scan_journal(str(journal_fixture), verbose=True).changed
    assert changed is True

    # Index file should exist
    index_path = journal_fixture / "indexer" / "journal.sqlite"
    assert index_path.exists()


def test_known_agents_returns_indexed_agents(journal_fixture):
    """known_agents() returns distinct non-empty lowercase agent names."""
    import solstone.think.utils as think_utils
    from solstone.think.indexer.journal import known_agents, scan_journal

    think_utils._journal_path_cache = None
    scan_journal(str(journal_fixture), full=True)
    agents = known_agents()
    assert agents  # non-empty
    assert "flow" in agents
    assert all(a and a == a.lower() for a in agents)


def test_known_agents_empty_on_unscanned_journal(tmp_path, monkeypatch):
    """A fresh journal with no built index yields an empty set."""
    from solstone.think.indexer.journal import known_agents

    monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
    import solstone.think.utils as think_utils

    think_utils._journal_path_cache = None
    try:
        assert known_agents() == set()
    finally:
        think_utils._journal_path_cache = None


def test_search_journal_outputs(journal_fixture):
    """Test searching returns agent output chunks."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    total, results = search_journal("project alpha")
    assert total >= 1
    # Should find the flow output mentioning "project alpha"
    found = any("alpha" in r["text"].lower() for r in results)
    assert found


def test_search_journal_apostrophe_terms(journal_fixture):
    """Apostrophe-bearing search terms are valid against a real scanned index."""
    from solstone.think.indexer.journal import scan_journal

    talents_dir = journal_fixture / "chronicle" / "20240101" / "talents"
    (talents_dir / "apostrophes.md").write_text(
        "# Apostrophe Search\n\n"
        "it's indexed exactly here. "
        "Bob O'Brien said don't panic. "
        "O'Brien brought dogs to the review.\n"
    )

    scan_journal(str(journal_fixture), full=True)

    for query in ("it's", "O'Brien", "don't panic", "O'Brien AND dogs"):
        total, results = search_journal(query)
        assert total >= 1
        assert results


def test_search_journal_yesterdays_meeting_apostrophe(journal_fixture):
    """Wall-clock temporal extraction still matches apostrophe residue."""
    from solstone.think.indexer.journal import scan_journal

    yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
    talents_dir = journal_fixture / "chronicle" / yesterday / "talents"
    talents_dir.mkdir(parents=True, exist_ok=True)
    (talents_dir / "flow.md").write_text(
        "# Yesterday\n\nReviewed yesterday's meeting notes.\n"
    )

    scan_journal(str(journal_fixture), full=True)

    total, results = search_journal("yesterday's meeting")
    assert total >= 1
    assert any("yesterday's meeting" in result["text"].lower() for result in results)


def test_segment_sense_json_searchable_and_exact(journal_copy):
    """Segment sense JSON is indexed with its dedicated formatter."""
    from solstone.think.formatters import find_formattable_files
    from solstone.think.indexer.journal import scan_journal, search_journal

    day = "20990115"
    segment_dir = journal_copy / "chronicle" / day / "default" / "120000_300"
    talents_dir = segment_dir / "talents"
    talents_dir.mkdir(parents=True, exist_ok=True)
    (talents_dir / "sense.json").write_text(
        json.dumps(
            {
                "density": "active",
                "content_type": "coding",
                "activity_summary": (
                    "Discussed the Monazite Comet Handoff unique regression seed."
                ),
                "entities": [
                    {
                        "type": "Project",
                        "name": "Zephyr Quartz Index",
                        "role": "mentioned",
                        "source": "screen",
                        "context": "Anchored the Monazite Comet Handoff.",
                    }
                ],
                "facets": [
                    {"facet": "work", "activity": "search indexing", "level": "high"}
                ],
                "speculative_facet": None,
                "meeting_detected": True,
                "speakers": ["Alice Smith", "Bob Chen"],
                "recommend": {"screen_record": True, "speaker_attribution": True},
                "emotional_register": "focused",
            }
        ),
        encoding="utf-8",
    )
    rel_path = f"{day}/default/120000_300/talents/sense.json"

    formattable = find_formattable_files(str(journal_copy))
    assert rel_path in formattable

    scan_journal(str(journal_copy), full=True)

    total, name_results = search_journal("Zephyr Quartz Index")
    assert total >= 1
    assert any(result["metadata"]["path"] == rel_path for result in name_results)

    total, description_results = search_journal("Monazite Comet Handoff")
    assert total >= 1
    matching = [
        result
        for result in description_results
        if result["metadata"]["path"] == rel_path
    ]
    assert matching
    assert any("Monazite Comet Handoff" in result["text"] for result in matching)
    assert all(result["metadata"]["agent"] == "sense" for result in matching)


def test_search_journal_events(journal_fixture):
    """Test searching returns event chunks."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    total, results = search_journal("Standup", agent="event")
    assert total >= 1
    assert any("Standup" in r["text"] for r in results)


def test_search_journal_filter_by_day(journal_fixture):
    """Test filtering search by day."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Search with day filter
    total, results = search_journal("", day="20240101")
    assert total >= 1
    for r in results:
        assert r["metadata"]["day"] == "20240101"


def test_search_journal_filter_by_facet(journal_fixture):
    """Test filtering search by facet."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Search with facet filter
    total, results = search_journal("", facet="work")
    assert total >= 1
    for r in results:
        assert r["metadata"]["facet"] == "work"


def test_search_journal_filter_by_agent(journal_fixture):
    """Test filtering search by agent."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Search events by agent
    total, results = search_journal("", agent="event")
    assert total >= 1
    for r in results:
        assert r["metadata"]["agent"] == "event"


def test_search_journal_facet_case_insensitive(journal_fixture):
    """Test facet filtering is case-insensitive."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Search with uppercase facet filter should find lowercase-indexed data
    total_upper, results_upper = search_journal("", facet="WORK")
    total_lower, _ = search_journal("", facet="work")
    total_mixed, _ = search_journal("", facet="Work")

    assert total_upper == total_lower == total_mixed
    assert total_upper >= 1
    # All results should have lowercase facet in metadata
    for r in results_upper:
        assert r["metadata"]["facet"] == "work"


def test_search_journal_agent_case_insensitive(journal_fixture):
    """Test agent filtering is case-insensitive."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Search with uppercase agent filter should find lowercase-indexed data
    total_upper, results_upper = search_journal("", agent="EVENT")
    total_lower, _ = search_journal("", agent="event")
    total_mixed, _ = search_journal("", agent="Event")

    assert total_upper == total_lower == total_mixed
    assert total_upper >= 1
    # All results should have lowercase agent in metadata
    for r in results_upper:
        assert r["metadata"]["agent"] == "event"


def test_time_bucket_population(journal_fixture):
    """Test that indexed chunks have correct time_bucket values."""
    from solstone.think.indexer.journal import get_journal_index, scan_journal

    scan_journal(str(journal_fixture), verbose=True, full=True)
    conn, _ = get_journal_index(str(journal_fixture))

    # Segment chunks should have correct time buckets
    morning_rows = conn.execute(
        "SELECT time_bucket FROM chunks WHERE agent='segment' AND path LIKE '%100000_300%'"
    ).fetchall()
    assert all(r[0] == "morning" for r in morning_rows)

    evening_rows = conn.execute(
        "SELECT time_bucket FROM chunks WHERE agent='segment' AND path LIKE '%200000_300%'"
    ).fetchall()
    assert all(r[0] == "evening" for r in evening_rows)

    # Non-segment content should have empty time_bucket
    entity_rows = conn.execute(
        "SELECT time_bucket FROM chunks WHERE path LIKE 'entity_search:%'"
    ).fetchall()
    assert all(r[0] == "" for r in entity_rows)

    conn.close()


def test_search_filter_by_time_bucket(journal_fixture):
    """Test filtering search by time_bucket."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture), verbose=True, full=True)

    # Morning filter should find 100000 segment content
    total, results = search_journal("", time_bucket="morning")
    assert total >= 1
    # All segment results should be from morning segments
    for r in results:
        if r["metadata"]["agent"] == "segment":
            assert "100000_300" in r["metadata"]["path"]

    # Evening filter should find 200000 segment content
    total, results = search_journal("", time_bucket="evening")
    assert total >= 1
    for r in results:
        if r["metadata"]["agent"] == "segment":
            assert "200000_300" in r["metadata"]["path"]

    # Non-matching bucket should return no segment content
    total, results = search_journal("", time_bucket="afternoon")
    segment_results = [r for r in results if r["metadata"]["agent"] == "segment"]
    assert len(segment_results) == 0


def test_time_bucket_non_segment_empty(journal_fixture):
    """Test that non-segment content (day-level agents, facet files) has empty time_bucket."""
    from solstone.think.indexer.journal import get_journal_index, scan_journal

    scan_journal(str(journal_fixture), verbose=True, full=True)
    conn, _ = get_journal_index(str(journal_fixture))

    # Day-level flow agent should have empty time_bucket
    flow_rows = conn.execute(
        "SELECT time_bucket FROM chunks WHERE agent='flow'"
    ).fetchall()
    assert all(r[0] == "" for r in flow_rows)

    # Event chunks should have empty time_bucket
    event_rows = conn.execute(
        "SELECT time_bucket FROM chunks WHERE agent='event'"
    ).fetchall()
    assert all(r[0] == "" for r in event_rows)

    conn.close()


def test_index_caching(journal_fixture):
    """Test that unchanged files are not re-indexed."""
    from solstone.think.indexer.journal import scan_journal

    # First scan indexes files
    changed = scan_journal(str(journal_fixture)).changed
    assert changed is True

    # Second scan should be a no-op (all cached)
    changed = scan_journal(str(journal_fixture)).changed
    assert changed is False


def test_is_historical_day():
    """Test _is_historical_day helper function."""
    from solstone.think.indexer.journal import _is_historical_day

    # Non-day paths are never historical
    assert _is_historical_day("facets/work/events/20240101.jsonl") is False
    assert _is_historical_day("imports/123/summary.md") is False
    assert _is_historical_day("apps/home/talents/foo.md") is False

    # Future dates are not historical
    assert _is_historical_day("29991231/talents/flow.md") is False

    # Path without slash is not historical
    assert _is_historical_day("20240101") is False
    assert _is_historical_day("") is False

    # Day paths before today are historical (tested with a very old date)
    assert _is_historical_day("20000101/talents/flow.md") is True


def test_scan_journal_full_mode(journal_fixture):
    """Test full mode includes all files including historical days."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    # Full scan should include everything
    changed = scan_journal(str(journal_fixture), full=True).changed
    assert changed is True

    # Should find content from historical day
    total, results = search_journal("project alpha")
    assert total >= 1


def test_find_formattable_files(journal_fixture):
    """Test file discovery function finds only indexed content."""
    from solstone.think.formatters import find_formattable_files

    files = find_formattable_files(str(journal_fixture))

    # Should find various file types
    paths = set(files.keys())

    # Daily agent outputs
    assert "20240101/talents/flow.md" in paths

    # Segment agent outputs
    assert "20240101/default/100000_300/talents/screen.md" in paths

    # Facet content
    assert "facets/work/events/20240101.jsonl" in paths
    assert "facets/work/entities/20240101.jsonl" in paths
    assert "facets/work/news/20240101.md" in paths


def test_find_formattable_files_includes_weekly_reflection(journal_copy):
    """Test tracked fixture reflections are included in indexed file discovery."""
    from solstone.think.formatters import find_formattable_files

    fixture_path = Path("tests/fixtures/journal/reflections/weekly/20260308.md")
    target_path = journal_copy / "reflections" / "weekly" / "20260308.md"
    target_path.parent.mkdir(parents=True, exist_ok=True)
    target_path.write_text(fixture_path.read_text(encoding="utf-8"), encoding="utf-8")

    files = find_formattable_files(str(journal_copy))

    assert "reflections/weekly/20260308.md" in files


def test_search_journal_empty_query(journal_fixture):
    """Test search with empty query returns all results."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Empty query should return all chunks
    total, results = search_journal("")
    assert total > 0


def test_search_journal_pagination(journal_fixture):
    """Test search pagination."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Get first page
    total, results1 = search_journal("", limit=2, offset=0)

    # Get second page
    _, results2 = search_journal("", limit=2, offset=2)

    # Results should be different (if enough data)
    if total > 2:
        ids1 = {r["id"] for r in results1}
        ids2 = {r["id"] for r in results2}
        assert ids1 != ids2


def test_search_journal_date_range(journal_fixture):
    """Test filtering search by date range."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    scan_journal(str(journal_fixture))

    # Search with date range that includes our test day
    total, results = search_journal("", day_from="20240101", day_to="20240101")
    assert total >= 1
    for r in results:
        assert r["metadata"]["day"] == "20240101"

    # Search with date range that excludes our test day
    total, results = search_journal("", day_from="20240102", day_to="20240105")
    assert total == 0


def test_search_counts_date_range(journal_fixture):
    """Test search_counts with date range filtering."""
    from solstone.think.indexer.journal import scan_journal, search_counts

    scan_journal(str(journal_fixture))

    # Counts with date range including test data
    counts = search_counts("", day_from="20240101", day_to="20240101")
    assert counts["total"] >= 1
    assert "20240101" in counts["days"]

    # Counts with date range excluding test data
    counts = search_counts("", day_from="20240102", day_to="20240105")
    assert counts["total"] == 0


def _joined_text(results):
    return "\n".join(result["text"] for result in results)


def _ids(results):
    return [result["id"] for result in results]


def test_collapse_drops_aggregate_when_child_matches(relax_fixture):
    total, results = search_journal("Acme", relax=False)

    assert total == 1
    assert all(result["metadata"]["agent"] != "segment" for result in results)
    assert _ids(results) == ["20240101/default/100000_300/talents/activity.md:0"]


def test_collapse_keeps_aggregate_when_no_child_matches(relax_fixture):
    total, results = search_journal("Zephyr Quokka", relax=False)

    assert total == 1
    assert len(results) == 1
    result = results[0]
    assert result["metadata"]["agent"] == "segment"
    assert result["metadata"]["path"] == "20240101/default/110000_300"
    assert result["metadata"]["idx"] == 1


def test_collapse_total_parity_and_reduction(relax_fixture):
    q = "Acme"

    total, _ = search_journal(q)
    counts = search_counts(q)

    assert total == counts["total"] == 1

    conn, _ = get_journal_index()
    raw = conn.execute(
        "SELECT count(*) FROM chunks WHERE chunks MATCH ?", ["Acme"]
    ).fetchone()[0]
    conn.close()

    assert raw == 2
    assert search_journal(q)[0] < raw


def test_collapse_pagination_no_duplicate_aggregate(relax_fixture):
    q = "Acme OR documentation"

    assert search_journal(q)[0] == 2

    page0_total, page0 = search_journal(q, limit=1, offset=0)
    page1_total, page1 = search_journal(q, limit=1, offset=1)
    page2_total, page2 = search_journal(q, limit=1, offset=2)

    assert page0_total == page1_total == page2_total == 2
    assert page2 == []

    pages = page0 + page1 + page2
    ids = _ids(pages)
    assert set(ids) == {
        "20240101/default/100000_300/talents/activity.md:0",
        "20240101/default/100000_300/talents/screen.md:0",
    }
    assert len(ids) == len(set(ids))
    assert all(result["metadata"]["agent"] != "segment" for result in pages)


def test_search_counts_relaxed_flag(relax_fixture):
    relaxed = search_counts("what was the Acme deal", relax=True)

    assert relaxed["relaxed"] is True
    assert search_counts("what was the Acme deal", relax=False)["relaxed"] is False
    assert search_counts("Acme", relax=True)["relaxed"] is False
    assert relaxed["total"] == search_journal("what was the Acme deal", relax=True)[0]


def test_relax_all_stopwords_without_filter_never_browses(relax_fixture):
    assert search_journal("what did i do", relax=True) == (0, [])


def test_relax_keeps_tier_zero_matches_identical(relax_fixture):
    plain = search_journal("singlematch", relax=False)
    relaxed = search_journal("singlematch", relax=True)

    assert relaxed == plain


def test_relax_stopword_ladder_rescues_natural_question(relax_fixture):
    unrelaxed_total, unrelaxed_results = search_journal(
        "what was the Acme deal", relax=False
    )
    relaxed_total, relaxed_results = search_journal(
        "what was the Acme deal", relax=True
    )
    browse_total, _ = search_journal("", relax=False)

    assert (unrelaxed_total, unrelaxed_results) == (0, [])
    assert 0 < relaxed_total < browse_total
    assert "Acme deal" in _joined_text(relaxed_results)


def test_relax_or_tier_rescues_terms_in_different_chunks(relax_fixture):
    unrelaxed_total, unrelaxed_results = search_journal(
        "documentation Standup", relax=False
    )
    relaxed_total, relaxed_results = search_journal("documentation Standup", relax=True)
    text = _joined_text(relaxed_results)

    assert (unrelaxed_total, unrelaxed_results) == (0, [])
    assert relaxed_total >= 2
    assert "documentation" in text
    assert "Standup" in text


def test_relax_or_tier_quotes_apostrophe_terms(relax_fixture):
    total, results = search_journal("owner's Standup", relax=True)
    text = _joined_text(results)

    assert total >= 2
    assert "Owner's decision" in text
    assert "Standup" in text


def test_relax_temporal_content_threads_day_filter(relax_fixture):
    yesterday = relax_fixture["yesterday"]
    total, results = search_journal("what was Calypso yesterday", relax=True)

    assert total >= 1
    assert any("Calypso ladder recall" in result["text"] for result in results)
    assert all(result["metadata"]["day"] == yesterday for result in results)


def test_relax_power_user_queries_stay_empty(relax_fixture):
    assert search_journal("Acme AND Standup", relax=True) == (0, [])
    assert search_journal('"Acme deal" Standup', relax=True) == (0, [])


def test_relax_operator_detection_is_token_level(relax_fixture):
    total, results = search_journal("ANDROID ORACLE", relax=True)
    text = _joined_text(results)

    assert total >= 2
    assert "ANDROID migration" in text
    assert "ORACLE support" in text


def test_relax_preserves_facet_and_date_range_filters(relax_fixture):
    facet_total, facet_results = search_journal(
        "what was Standup", facet="work", relax=True
    )
    range_total, range_results = search_journal(
        "what was the Acme deal",
        day_from="20240101",
        day_to="20240101",
        relax=True,
    )

    assert facet_total >= 1
    assert all(result["metadata"]["facet"] == "work" for result in facet_results)
    assert range_total >= 1
    assert all(result["metadata"]["day"] == "20240101" for result in range_results)


def test_relax_counts_and_results_choose_same_tier_for_all_tiers(relax_fixture):
    for query in (
        "what was the Acme deal",
        "documentation Standup",
        "what did i do yesterday",
    ):
        total, _ = search_journal(query, relax=True)
        counts = search_counts(query, relax=True)

        assert total > 0
        assert counts["total"] == total


def test_rerank_reorders_top_fts_pool(monkeypatch, relax_fixture):
    total, bm25_results = search_journal("sharedrank", limit=3, rerank=False)
    assert total == 3
    score_by_text = {
        result["text"]: float(index)
        for index, result in enumerate(bm25_results, start=1)
    }

    def stub(query, texts):
        assert query == "sharedrank"
        return [score_by_text[text] for text in texts]

    monkeypatch.setattr("solstone.think.indexer.journal.score", stub)

    _, reranked = search_journal("sharedrank", limit=3, rerank=True)

    assert _ids(reranked) == list(reversed(_ids(bm25_results)))
    assert all("rerank_score" in result for result in reranked)


def test_rerank_score_none_falls_back_to_bm25(monkeypatch, relax_fixture):
    _, bm25_results = search_journal("sharedrank", limit=3, rerank=False)
    monkeypatch.setattr("solstone.think.indexer.journal.score", lambda *_args: None)

    _, results = search_journal("sharedrank", limit=3, rerank=True)

    assert results == bm25_results
    assert all("rerank_score" not in result for result in results)


def test_rerank_skips_without_calling_score(monkeypatch, relax_fixture):
    def fail_score(*_args):
        raise AssertionError("score should not be called")

    monkeypatch.setattr("solstone.think.indexer.journal.score", fail_score)

    search_journal("sharedrank", limit=2, offset=49, rerank=True)
    search_journal("", limit=2, rerank=True)
    search_journal("singlematch", rerank=True)
    search_journal("yesterday", rerank=True)


def test_rerank_equal_scores_keep_bm25_order(monkeypatch, relax_fixture):
    _, bm25_results = search_journal("sharedrank", limit=3, rerank=False)

    def stub(_query, texts):
        return [1.0 for _text in texts]

    monkeypatch.setattr("solstone.think.indexer.journal.score", stub)

    _, results = search_journal("sharedrank", limit=3, rerank=True)

    assert _ids(results) == _ids(bm25_results)
    assert all(result["rerank_score"] == 1.0 for result in results)


def test_search_tool_keeps_rerank_without_relax(monkeypatch):
    from collections import Counter

    from solstone.think.tools import search as search_tool

    calls = {}

    def fake_search(query, limit, offset, **kwargs):
        calls["search"] = (query, limit, offset, kwargs)
        return 0, []

    def fake_counts(query, **kwargs):
        calls["counts"] = (query, kwargs)
        return {
            "total": 0,
            "facets": Counter(),
            "agents": Counter(),
            "days": Counter(),
            "streams": Counter(),
        }

    monkeypatch.setattr(search_tool, "search_journal_impl", fake_search)
    monkeypatch.setattr(search_tool, "search_counts_impl", fake_counts)

    search_tool.search_journal("needle", facet="work")

    assert "relax" not in calls["search"][3]
    assert calls["search"][3]["rerank"] is True
    assert calls["search"][3]["facet"] == "work"
    assert "relax" not in calls["counts"][1]
    assert "rerank" not in calls["counts"][1]


def test_call_tool_keeps_rerank_without_relax(monkeypatch):
    from collections import Counter

    from solstone.think.indexer import journal as journal_index
    from solstone.think.tools import call as call_tool

    calls = {}

    def fake_search(query, limit, offset, **kwargs):
        calls["search"] = (query, limit, offset, kwargs)
        return 0, []

    def fake_counts(query, **kwargs):
        calls["counts"] = (query, kwargs)
        return {
            "total": 0,
            "facets": Counter(),
            "agents": Counter(),
            "days": Counter(),
            "streams": Counter(),
        }

    monkeypatch.setattr(journal_index, "search_journal", fake_search)
    monkeypatch.setattr(journal_index, "search_counts", fake_counts)

    call_tool.search(
        query="needle",
        query_opt=None,
        limit=5,
        offset=2,
        day=None,
        day_from=None,
        day_to=None,
        facet=None,
        agent=None,
        stream=None,
        time_bucket=None,
        json_output=False,
    )

    assert "relax" not in calls["search"][3]
    assert calls["search"][3]["rerank"] is True
    assert "relax" not in calls["counts"][1]
    assert "rerank" not in calls["counts"][1]


def test_voice_tool_flips_relax_and_rerank(monkeypatch):
    from solstone.think.voice import tools as voice_tools

    calls = {}

    def fake_search(query, **kwargs):
        calls["search"] = (query, kwargs)
        return 0, []

    monkeypatch.setattr(voice_tools, "search_journal", fake_search)

    result = voice_tools.handle_journal_search(
        {"query": "needle", "facet": "work", "days": 7, "limit": 3},
        app=None,
    )

    assert result["count"] == 0
    assert calls["search"][1]["relax"] is True
    assert calls["search"][1]["rerank"] is True
    assert calls["search"][1]["facet"] == "work"


def test_search_journal_returns_counts(monkeypatch):
    """Test search tool returns counts aggregation."""
    from solstone.think.tools import search as search_tools

    monkeypatch.setattr(search_tools, "search_journal_impl", Mock(return_value=(0, [])))
    monkeypatch.setattr(
        search_tools,
        "search_counts_impl",
        Mock(
            return_value={
                "facets": [("work", 2)],
                "agents": [("flow", 2)],
                "days": [("20240101", 2)],
            }
        ),
    )

    result = search_tools.search_journal("test")

    # Should have counts structure
    assert "counts" in result
    counts = result["counts"]
    assert "facets" in counts
    assert "agents" in counts
    assert "recent_days" in counts
    assert "top_days" in counts
    assert "bucketed_days" in counts

    # recent_days should have 7 entries (including zeros)
    assert len(counts["recent_days"]) == 7


def test_search_journal_returns_query_echo(monkeypatch):
    """Test search tool returns query echo."""
    from solstone.think.tools import search as search_tools

    monkeypatch.setattr(search_tools, "search_journal_impl", Mock(return_value=(0, [])))
    monkeypatch.setattr(
        search_tools,
        "search_counts_impl",
        Mock(return_value={"facets": [], "agents": [], "days": []}),
    )

    result = search_tools.search_journal("test query", facet="work", agent="flow")

    assert "query" in result
    assert result["query"]["text"] == "test query"
    assert result["query"]["filters"]["facet"] == "work"
    assert result["query"]["filters"]["agent"] == "flow"


def test_search_journal_results_include_path(monkeypatch):
    """Test search tool results include path and idx."""
    from solstone.think.tools import search as search_tools

    monkeypatch.setattr(
        search_tools,
        "search_journal_impl",
        Mock(
            return_value=(
                1,
                [
                    {
                        "text": "result",
                        "metadata": {"path": "facets/work/result.md", "idx": 3},
                    }
                ],
            )
        ),
    )
    monkeypatch.setattr(
        search_tools,
        "search_counts_impl",
        Mock(return_value={"facets": [], "agents": [], "days": []}),
    )

    result = search_tools.search_journal("")

    item = result["results"][0]
    assert item["path"] == "facets/work/result.md"
    assert item["idx"] == 3
    assert item["id"] == "facets/work/result.md:3"


def test_search_journal_truncates_large_results(monkeypatch):
    """Test search tool truncates oversized result text."""
    from unittest.mock import patch

    from solstone.think.tools.search import _MAX_RESULT_TEXT, search_journal

    big_text = "x" * 10_000
    fake_results = [
        {
            "text": big_text,
            "metadata": {
                "day": "20240101",
                "facet": "",
                "agent": "test",
                "path": "a.md",
                "idx": 0,
            },
            "score": 1.0,
        }
    ]
    fake_counts = {"facets": [], "agents": [], "days": []}

    with (
        patch(
            "solstone.think.tools.search.search_journal_impl",
            return_value=(1, fake_results),
        ),
        patch(
            "solstone.think.tools.search.search_counts_impl", return_value=fake_counts
        ),
    ):
        result = search_journal("test")

    text = result["results"][0]["text"]
    assert len(text) < _MAX_RESULT_TEXT + 200  # truncated + note
    assert "truncated from 10,000 chars" in text


def test_bucket_day_counts():
    """Test day bucketing logic."""
    from datetime import datetime, timedelta

    from solstone.think.tools.search import _bucket_day_counts

    today = datetime.now()

    # Create test data with various dates
    day_counts = {}

    # Add recent days (within last 7 days)
    for i in range(3):
        d = (today - timedelta(days=i)).strftime("%Y%m%d")
        day_counts[d] = 5 + i

    # Add older days (more than 7 days ago)
    for i in range(10, 25):
        d = (today - timedelta(days=i)).strftime("%Y%m%d")
        day_counts[d] = 2

    result = _bucket_day_counts(day_counts)

    # recent_days should have 7 entries
    assert len(result["recent_days"]) == 7

    # top_days should have entries
    assert len(result["top_days"]) > 0

    # bucketed_days should have entries for older days
    assert len(result["bucketed_days"]) > 0

    # Bucketed day keys should be in YYYYMMDD-YYYYMMDD format
    for key in result["bucketed_days"]:
        assert "-" in key
        parts = key.split("-")
        assert len(parts) == 2
        assert len(parts[0]) == 8
        assert len(parts[1]) == 8


def test_light_scan_removes_deleted_facet_content(journal_fixture):
    """Test that light scan detects and removes deleted facet files."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    # Initial scan
    scan_journal(str(journal_fixture), full=True)

    # Verify event is indexed
    total, _ = search_journal("Standup", agent="event")
    assert total >= 1

    # Delete the facet event file
    events_file = journal_fixture / "facets" / "work" / "events" / "20240101.jsonl"
    events_file.unlink()

    # Light rescan should detect the deletion (facet content is in scope)
    changed = scan_journal(str(journal_fixture), full=False).changed
    assert changed is True

    # Event should no longer be searchable
    total, _ = search_journal("Standup", agent="event")
    assert total == 0


def test_light_scan_removes_deleted_today_segment(tmp_path, monkeypatch):
    """Test that light scan detects and removes deleted content from today."""
    from datetime import datetime

    from solstone.think.indexer.journal import scan_journal, search_journal

    journal = tmp_path
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))

    # Create content for today (which is in light scan scope)
    today = datetime.now().strftime("%Y%m%d")
    day_dir = journal / today
    day_dir.mkdir(parents=True)
    agents_dir = day_dir / "talents"
    agents_dir.mkdir()
    output_file = agents_dir / "flow.md"
    output_file.write_text("# Today Flow\n\nWorked on unique_today_content.\n")

    # Initial scan
    scan_journal(str(journal), full=False)

    # Verify content is indexed
    total, _ = search_journal("unique_today_content")
    assert total >= 1

    # Delete the output file
    output_file.unlink()

    # Light rescan should detect the deletion
    changed = scan_journal(str(journal), full=False).changed
    assert changed is True

    # Content should no longer be searchable
    total, _ = search_journal("unique_today_content")
    assert total == 0


def test_light_scan_preserves_historical_content(tmp_path, monkeypatch):
    """Test that light scan does NOT remove historical day content from index."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    journal = tmp_path
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))

    # Create historical day content
    day_dir = journal / "chronicle" / "20200101"
    day_dir.mkdir(parents=True)
    agents_dir = day_dir / "talents"
    agents_dir.mkdir()
    output_file = agents_dir / "flow.md"
    output_file.write_text("# Historical Flow\n\nWorked on historical_content.\n")

    # Full scan to index historical content
    scan_journal(str(journal), full=True)

    # Verify content is indexed
    total, _ = search_journal("historical_content")
    assert total >= 1

    # Delete the historical file
    output_file.unlink()

    # Light rescan should NOT remove the historical content (out of scope)
    changed = scan_journal(str(journal), full=False).changed
    # No changes because the historical path is out of scope
    assert changed is False

    # Content should still be searchable (not removed)
    total, _ = search_journal("historical_content")
    assert total >= 1


def test_full_scan_removes_historical_content(tmp_path, monkeypatch):
    """Test that full scan removes deleted historical day content."""
    from solstone.think.indexer.journal import scan_journal, search_journal

    journal = tmp_path
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))

    # Create historical day content
    day_dir = journal / "chronicle" / "20200101"
    day_dir.mkdir(parents=True)
    agents_dir = day_dir / "talents"
    agents_dir.mkdir()
    output_file = agents_dir / "flow.md"
    output_file.write_text("# Historical Flow\n\nWorked on historical_full_test.\n")

    # Full scan to index historical content
    scan_journal(str(journal), full=True)

    # Verify content is indexed
    total, _ = search_journal("historical_full_test")
    assert total >= 1

    # Delete the historical file
    output_file.unlink()

    # Full rescan SHOULD remove the historical content
    changed = scan_journal(str(journal), full=True).changed
    assert changed is True

    # Content should no longer be searchable
    total, _ = search_journal("historical_full_test")
    assert total == 0


def test_index_file_valid(journal_fixture):
    """Test indexing a single valid file."""
    from solstone.think.indexer.journal import index_file, search_journal

    # Index a specific file
    result = index_file(str(journal_fixture), "20240101/talents/flow.md", verbose=True)
    assert result is True

    # Should be searchable
    total, results = search_journal("project alpha")
    assert total >= 1


def test_index_file_absolute_path(journal_fixture):
    """Test indexing with absolute path."""
    from solstone.think.indexer.journal import index_file, search_journal

    abs_path = str(journal_fixture / "chronicle" / "20240101" / "talents" / "flow.md")
    result = index_file(str(journal_fixture), abs_path, verbose=True)
    assert result is True

    # Should be searchable
    total, _ = search_journal("project alpha")
    assert total >= 1


def test_scan_journal_never_stores_chronicle_prefix(journal_fixture):
    """Chronicle is an on-disk prefix only, never a stored relative path."""
    from solstone.think.indexer.journal import scan_journal

    scan_journal(str(journal_fixture), full=True)

    conn, _ = get_journal_index(str(journal_fixture))
    try:
        assert (
            conn.execute(
                "SELECT count(*) FROM files WHERE path LIKE 'chronicle/%'"
            ).fetchone()[0]
            == 0
        )
        assert (
            conn.execute(
                "SELECT count(*) FROM chunks WHERE path LIKE 'chronicle/%'"
            ).fetchone()[0]
            == 0
        )
    finally:
        conn.close()


def test_index_file_updates_existing(journal_fixture):
    """Test that re-indexing a file replaces existing chunks."""
    from solstone.think.indexer.journal import index_file, search_journal

    # Index the file
    index_file(str(journal_fixture), "20240101/talents/flow.md")

    # Get initial count
    total1, _ = search_journal("project alpha")

    # Re-index the same file
    index_file(str(journal_fixture), "20240101/talents/flow.md")

    # Count should be the same (not doubled)
    total2, _ = search_journal("project alpha")
    assert total2 == total1


def test_index_file_not_found(journal_fixture):
    """Test indexing non-existent file raises error."""
    from solstone.think.indexer.journal import index_file

    with pytest.raises(FileNotFoundError, match="File not found"):
        index_file(str(journal_fixture), "nonexistent/file.md")


def test_index_file_outside_journal(journal_fixture, tmp_path_factory):
    """Test indexing file outside journal raises error."""
    from solstone.think.indexer.journal import index_file

    # Create a file in a separate temp directory (outside the journal)
    outside_dir = tmp_path_factory.mktemp("outside")
    outside_file = outside_dir / "outside.md"
    outside_file.write_text("# Outside\n\nThis is outside the journal.\n")

    with pytest.raises(ValueError, match="outside journal directory"):
        index_file(str(journal_fixture), str(outside_file))


def test_index_file_no_formatter(journal_fixture):
    """Test indexing file without formatter raises error."""
    from solstone.think.indexer.journal import index_file

    # Create a file with no formatter (e.g., .txt)
    txt_file = journal_fixture / "chronicle" / "20240101" / "notes.txt"
    txt_file.write_text("Just some text notes.\n")

    with pytest.raises(ValueError, match="No formatter found"):
        index_file(str(journal_fixture), str(txt_file))


# --- Stream indexing tests ---


def test_extract_stream_segment_path(tmp_path):
    """_extract_stream reads stream.json from segment directories."""
    from solstone.think.indexer.journal import _extract_stream
    from solstone.think.streams import write_segment_stream

    # Create a segment with stream marker
    seg_dir = tmp_path / "chronicle" / "20240101" / "default" / "123456_300"
    seg_dir.mkdir(parents=True)
    write_segment_stream(seg_dir, "archon", None, None, 1)

    result = _extract_stream(
        str(tmp_path), "20240101/default/123456_300/talents/work/flow.md"
    )
    assert result == "archon"


def test_extract_stream_non_segment_path(tmp_path):
    """_extract_stream returns None for non-segment paths."""
    from solstone.think.indexer.journal import _extract_stream

    result = _extract_stream(str(tmp_path), "20240101/talents/flow.md")
    assert result is None

    result = _extract_stream(str(tmp_path), "facets/work/events/20240101.jsonl")
    assert result is None


def test_extract_stream_missing_marker(tmp_path):
    """_extract_stream returns None when stream.json doesn't exist."""
    from solstone.think.indexer.journal import _extract_stream

    seg_dir = tmp_path / "chronicle" / "20240101" / "default" / "123456_300"
    seg_dir.mkdir(parents=True)

    result = _extract_stream(
        str(tmp_path), "20240101/default/123456_300/talents/work/flow.md"
    )
    assert result is None


def test_search_tool_stream_filter(monkeypatch):
    """Agent search tool accepts and passes stream filter."""
    from solstone.think.tools import search as search_tools

    search = Mock(return_value=(1, [{"text": "result", "metadata": {}}]))
    counts = Mock(return_value={"facets": [], "agents": [], "days": [("20240101", 1)]})
    monkeypatch.setattr(search_tools, "search_journal_impl", search)
    monkeypatch.setattr(search_tools, "search_counts_impl", counts)

    result = search_tools.search_journal("", stream="default")

    assert "results" in result
    assert result["total"] == 1
    assert result["query"]["filters"]["stream"] == "default"
    search.assert_called_once_with("", 10, 0, rerank=True, stream="default")
    counts.assert_called_once_with("", stream="default")


def test_prune_chunks_by_stream(monkeypatch, tmp_path):
    """prune_chunks_by_stream removes only the target stream's chunks."""
    from solstone.think.indexer.journal import (
        get_journal_index,
        prune_chunks_by_stream,
        scan_journal,
    )
    from solstone.think.streams import write_segment_stream

    journal = tmp_path / "journal"
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))

    share_seg = journal / "chronicle" / "20240101" / "import.share" / "100000_300"
    share_talents = share_seg / "talents"
    share_talents.mkdir(parents=True)
    (share_talents / "screen.md").write_text(
        "# Share\n\nImported share source content.\n",
        encoding="utf-8",
    )
    write_segment_stream(share_seg, "import.share", None, None, 1)

    other_seg = journal / "chronicle" / "20240101" / "import.apple" / "101000_300"
    other_talents = other_seg / "talents"
    other_talents.mkdir(parents=True)
    (other_talents / "screen.md").write_text(
        "# Apple\n\nImported apple source content.\n",
        encoding="utf-8",
    )
    write_segment_stream(other_seg, "import.apple", None, None, 1)

    scan_journal(str(journal), full=True)
    conn, _ = get_journal_index(str(journal))
    share_count = conn.execute(
        "SELECT count(*) FROM chunks WHERE stream=?",
        ("import.share",),
    ).fetchone()[0]
    other_count = conn.execute(
        "SELECT count(*) FROM chunks WHERE stream=?",
        ("import.apple",),
    ).fetchone()[0]
    share_paths = [
        row[0]
        for row in conn.execute(
            "SELECT DISTINCT path FROM chunks WHERE stream=?",
            ("import.share",),
        ).fetchall()
    ]
    conn.close()

    assert share_count > 0
    assert other_count > 0
    assert share_paths

    result = prune_chunks_by_stream("import.share", str(journal))

    assert result["chunks"] == share_count
    assert result["files"] == len(share_paths)

    conn, _ = get_journal_index(str(journal))
    assert (
        conn.execute(
            "SELECT count(*) FROM chunks WHERE stream=?",
            ("import.share",),
        ).fetchone()[0]
        == 0
    )
    assert (
        conn.execute(
            "SELECT count(*) FROM chunks WHERE stream=?",
            ("import.apple",),
        ).fetchone()[0]
        == other_count
    )
    for path in share_paths:
        assert (
            conn.execute(
                "SELECT count(*) FROM files WHERE path=?",
                (path,),
            ).fetchone()[0]
            == 0
        )
    conn.close()


def test_delete_segment_index_rows_removes_only_target_segment(monkeypatch, tmp_path):
    """delete_segment_index_rows remains an in-process index maintenance helper."""
    from solstone.think.indexer.journal import (
        delete_segment_index_rows,
        get_journal_index,
        scan_journal,
    )
    from solstone.think.streams import write_segment_stream

    journal = tmp_path / "journal"
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))

    target_rel = "20240101/default/100000_300"
    target_seg = journal / "chronicle" / target_rel
    target_talents = target_seg / "talents"
    target_talents.mkdir(parents=True)
    (target_talents / "screen.md").write_text(
        "# Target\n\nTarget segment content.\n",
        encoding="utf-8",
    )
    write_segment_stream(target_seg, "default", None, None, 1)

    other_rel = "20240101/default/101000_300"
    other_seg = journal / "chronicle" / other_rel
    other_talents = other_seg / "talents"
    other_talents.mkdir(parents=True)
    (other_talents / "screen.md").write_text(
        "# Other\n\nOther segment content.\n",
        encoding="utf-8",
    )
    write_segment_stream(other_seg, "default", None, None, 1)

    scan_journal(str(journal), full=True)
    conn, _ = get_journal_index(str(journal))
    target_chunks = conn.execute(
        "SELECT count(*) FROM chunks WHERE path = ? OR path LIKE ?",
        (target_rel, f"{target_rel}/%"),
    ).fetchone()[0]
    target_files = conn.execute(
        "SELECT count(*) FROM files WHERE path LIKE ?",
        (f"{target_rel}/%",),
    ).fetchone()[0]
    other_chunks = conn.execute(
        "SELECT count(*) FROM chunks WHERE path = ? OR path LIKE ?",
        (other_rel, f"{other_rel}/%"),
    ).fetchone()[0]
    conn.close()

    assert target_chunks > 0
    assert target_files > 0
    assert other_chunks > 0

    result = delete_segment_index_rows(str(journal), target_rel)

    assert result == {"chunks": target_chunks, "files": target_files, "error": None}

    conn, _ = get_journal_index(str(journal))
    assert (
        conn.execute(
            "SELECT count(*) FROM chunks WHERE path = ? OR path LIKE ?",
            (target_rel, f"{target_rel}/%"),
        ).fetchone()[0]
        == 0
    )
    assert (
        conn.execute(
            "SELECT count(*) FROM files WHERE path LIKE ?",
            (f"{target_rel}/%",),
        ).fetchone()[0]
        == 0
    )
    assert (
        conn.execute(
            "SELECT count(*) FROM chunks WHERE path = ? OR path LIKE ?",
            (other_rel, f"{other_rel}/%"),
        ).fetchone()[0]
        == other_chunks
    )
    conn.close()


class TestSegmentChunks:
    """Tests for segment-level concatenated FTS5 chunks."""

    def test_segment_chunks_created(self, journal_fixture):
        """scan_journal creates segment chunks with agent='segment'."""
        from solstone.think.indexer.journal import get_journal_index, scan_journal

        scan_journal(str(journal_fixture), verbose=True, full=True)
        conn, _ = get_journal_index(str(journal_fixture))
        rows = conn.execute(
            "SELECT content, path, day, facet, agent, stream FROM chunks WHERE agent='segment'"
        ).fetchall()
        conn.close()
        assert len(rows) >= 1
        content, path, day, facet, agent, stream = rows[0]
        assert content
        assert path == "20240101/default/100000_300"
        assert day == "20240101"
        assert facet == ""
        assert agent == "segment"
        assert stream == "default"

    def test_segment_chunk_contains_all_agent_content(self, journal_fixture):
        """Segment chunk content includes text from all agent files."""
        from solstone.think.indexer.journal import get_journal_index, scan_journal

        scan_journal(str(journal_fixture), verbose=True, full=True)
        conn, _ = get_journal_index(str(journal_fixture))
        rows = conn.execute(
            "SELECT content FROM chunks WHERE agent='segment'"
        ).fetchall()
        conn.close()
        all_content = " ".join(r[0] for r in rows)
        assert "Viewed documentation" in all_content
        assert "Scott Ward" in all_content

    def test_segment_chunk_searchable(self, journal_fixture):
        """Segment chunks are searchable via an explicit segment filter."""
        from solstone.think.indexer.journal import scan_journal, search_journal

        scan_journal(str(journal_fixture), verbose=True, full=True)
        total, results = search_journal("Scott Ward", agent="segment")
        assert total >= 1
        segment_results = [r for r in results if r["metadata"]["agent"] == "segment"]
        assert len(segment_results) >= 1

    def test_segment_chunk_cross_file_search(self, journal_fixture):
        """Segment-filtered search finds segment chunks for child terms."""
        from solstone.think.indexer.journal import scan_journal, search_journal

        scan_journal(str(journal_fixture), verbose=True, full=True)
        total1, results1 = search_journal("documentation", agent="segment")
        total2, results2 = search_journal("Acme deal", agent="segment")
        assert total1 >= 1
        assert total2 >= 1
        seg1 = [r for r in results1 if r["metadata"]["agent"] == "segment"]
        seg2 = [r for r in results2 if r["metadata"]["agent"] == "segment"]
        assert len(seg1) >= 1
        assert len(seg2) >= 1

    def test_agent_filter_returns_only_segments(self, journal_fixture):
        """agent='segment' filter returns only segment chunks."""
        from solstone.think.indexer.journal import scan_journal, search_journal

        scan_journal(str(journal_fixture), verbose=True, full=True)
        total, results = search_journal("", agent="segment")
        assert total >= 1
        for r in results:
            assert r["metadata"]["agent"] == "segment"

    def test_existing_agent_chunks_unchanged(self, journal_fixture):
        """Segment chunks are additive — agent-level chunks still exist."""
        from solstone.think.indexer.journal import get_journal_index, scan_journal

        scan_journal(str(journal_fixture), verbose=True, full=True)
        conn, _ = get_journal_index(str(journal_fixture))
        screen_chunks = conn.execute(
            "SELECT count(*) FROM chunks WHERE path='20240101/default/100000_300/talents/screen.md'"
        ).fetchone()[0]
        activity_chunks = conn.execute(
            "SELECT count(*) FROM chunks WHERE path='20240101/default/100000_300/talents/activity.md'"
        ).fetchone()[0]
        segment_chunks = conn.execute(
            "SELECT count(*) FROM chunks WHERE agent='segment'"
        ).fetchone()[0]
        conn.close()
        assert screen_chunks >= 1
        assert activity_chunks >= 1
        assert segment_chunks >= 1

    def test_idempotent_scan(self, journal_fixture):
        """Running scan_journal twice produces same segment chunk count."""
        from solstone.think.indexer.journal import get_journal_index, scan_journal

        scan_journal(str(journal_fixture), verbose=True, full=True)
        conn, _ = get_journal_index(str(journal_fixture))
        count1 = conn.execute(
            "SELECT count(*) FROM chunks WHERE agent='segment'"
        ).fetchone()[0]
        conn.close()
        scan_journal(str(journal_fixture), verbose=True, full=True)
        conn, _ = get_journal_index(str(journal_fixture))
        count2 = conn.execute(
            "SELECT count(*) FROM chunks WHERE agent='segment'"
        ).fetchone()[0]
        conn.close()
        assert count1 == count2


def test_chat_turn_is_searchable_after_rescan(journal_fixture):
    from solstone.think.indexer.journal import scan_journal, search_journal

    append_chat_event(
        "owner_message",
        text="Tell me about the nebula phrase",
        app="sol",
        path="/app/sol",
        facet="work",
    )
    append_chat_event(
        "sol_message",
        use_id="1713628000000",
        text="The unique nebula phrase is now in chat history.",
        notes="done",
        requested_target=None,
        requested_task=None,
    )

    scan_journal(str(journal_fixture), full=True)
    total, results = search_journal("unique nebula phrase")

    assert total >= 1
    assert any("unique nebula phrase" in result["text"].lower() for result in results)


def test_weekly_reflection_is_searchable_after_rescan(journal_copy):
    from solstone.think.indexer.journal import scan_journal, search_journal

    fixture_path = Path("tests/fixtures/journal/reflections/weekly/20260308.md")
    target_path = journal_copy / "reflections" / "weekly" / "20260308.md"
    target_path.parent.mkdir(parents=True, exist_ok=True)
    target_path.write_text(fixture_path.read_text(encoding="utf-8"), encoding="utf-8")

    scan_journal(str(journal_copy), full=True)
    total, results = search_journal("boardroom balcony inflection")

    assert total >= 1
    assert any(
        "boardroom balcony inflection" in result["text"].lower() for result in results
    )


def test_scan_journal_is_pure_wrt_entity_state(journal_copy):
    """scan_journal must not mutate journal/entities/ state."""
    from solstone.think.indexer.journal import scan_journal

    journal_path = Path(journal_copy)
    today = datetime.now().strftime("%Y%m%d")
    segment_dir = (
        journal_path / "chronicle" / today / "default" / "120000_300" / "talents"
    )
    segment_dir.mkdir(parents=True)
    (segment_dir / "sense.json").write_text(
        json.dumps(
            {
                "density": "active",
                "content_type": "coding",
                "activity_summary": "Unique regression seed for pure indexing.",
                "entities": [
                    {
                        "name": "Zephyr Quartz Index",
                        "type": "Project",
                        "role": "mentioned",
                        "source": "screen",
                        "context": "Used only for indexer purity coverage.",
                    }
                ],
                "facets": [
                    {"facet": "work", "activity": "search indexing", "level": "high"}
                ],
                "speculative_facet": None,
                "meeting_detected": False,
                "speakers": [],
                "recommend": {"screen_record": False, "speaker_attribution": False},
                "emotional_register": "focused",
            }
        ),
        encoding="utf-8",
    )

    def snapshot_entities(root: Path) -> list[tuple[str, str]]:
        entries = []
        for path in sorted((root / "entities").rglob("*")):
            if not path.is_file():
                continue
            rel = path.relative_to(root).as_posix()
            digest = hashlib.sha256(path.read_bytes()).hexdigest()
            entries.append((rel, digest))
        return entries

    snap_before = snapshot_entities(journal_path)
    scan_journal(str(journal_path), full=True)
    snap_between = snapshot_entities(journal_path)
    scan_journal(str(journal_path), full=True)
    snap_after = snapshot_entities(journal_path)

    assert snap_before == snap_between == snap_after, (
        "scan_journal() mutated journal/entities/ — see docs/coding-standards.md § L6"
    )


def test_scan_journal_rescan_does_not_record_entity_ambiguities(journal_copy):
    """Index rescans must not create domain ambiguity rows for low-confidence names."""
    from solstone.think.entities import load_ambiguities
    from solstone.think.indexer.journal import scan_journal

    journal_path = Path(journal_copy)
    for entity_id, name in [
        ("sarah_connor", "Sarah Connor"),
        ("sarah_lee", "Sarah Lee"),
    ]:
        entity_dir = journal_path / "entities" / entity_id
        entity_dir.mkdir(parents=True, exist_ok=True)
        (entity_dir / "entity.json").write_text(
            json.dumps(
                {
                    "id": entity_id,
                    "name": name,
                    "type": "Person",
                    "created_at": "2026-01-01T00:00:00Z",
                    "updated_at": "2026-01-01T00:00:00Z",
                }
            ),
            encoding="utf-8",
        )

    today = datetime.now().strftime("%Y%m%d")
    segment_dir = (
        journal_path / "chronicle" / today / "default" / "121000_300" / "talents"
    )
    segment_dir.mkdir(parents=True)
    (segment_dir / "sense.json").write_text(
        json.dumps(
            {
                "density": "active",
                "content_type": "meeting",
                "activity_summary": "Sarah discussed a low-confidence indexer case.",
                "entities": [
                    {
                        "name": "Sarah",
                        "type": "Person",
                        "role": "mentioned",
                        "source": "screen",
                        "context": "Ambiguous against Sarah Connor and Sarah Lee.",
                    }
                ],
                "facets": [
                    {"facet": "work", "activity": "search indexing", "level": "high"}
                ],
                "speculative_facet": None,
                "meeting_detected": False,
                "speakers": [],
                "recommend": {"screen_record": False, "speaker_attribution": False},
                "emotional_register": "focused",
            }
        ),
        encoding="utf-8",
    )

    ambiguity_path = journal_path / "entities" / "ambiguities.jsonl"
    assert not ambiguity_path.exists()

    scan_journal(str(journal_path), full=True)
    scan_journal(str(journal_path), full=True)

    assert not ambiguity_path.exists()
    assert load_ambiguities() == []
