"""Tests for canonical turn ownership, precedence, and repair."""

from __future__ import annotations

import ast
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, replace
from pathlib import Path
from types import SimpleNamespace
from typing import TYPE_CHECKING
from unittest.mock import ANY, MagicMock, patch

import pytest
from agno.db.base import SessionType
from agno.run.agent import RunOutput
from agno.run.team import TeamRunOutput
from agno.session.agent import AgentSession
from agno.session.summary import SessionSummary
from agno.session.team import TeamSession

from mindroom import constants
from mindroom.bot import AgentBot
from mindroom.config.main import Config
from mindroom.conversation_state_writer import ConversationStateWriter, ConversationStateWriterDeps
from mindroom.handled_turns import (
    SourceEventMetadata,
    TurnRecord,
    TurnRecordCodec,
    _reset_handled_turn_ledger_runtime,
)
from mindroom.history.storage import (
    read_scope_seen_event_ids,
    read_scope_state,
    update_scope_seen_event_ids,
    write_scope_state,
)
from mindroom.history.types import HistoryScope, HistoryScopeState
from mindroom.matrix.users import AgentMatrixUser
from mindroom.message_target import MessageTarget
from mindroom.text_ingress_dispatch import _run_claimed_response
from mindroom.turn_store import TurnStore, TurnStoreDeps
from tests.conftest import TEST_PASSWORD, bind_runtime_paths, runtime_paths_for, test_runtime_paths

if TYPE_CHECKING:
    from collections.abc import Callable


def _store(
    tmp_path: Path,
    *,
    on_terminal_turn_persisted: Callable[[tuple[str, ...]], None] | None = None,
) -> TurnStore:
    return TurnStore(
        TurnStoreDeps(
            agent_name="agent",
            tracking_base_path=tmp_path,
            state_writer=MagicMock(),
            resolver=MagicMock(),
            tool_runtime=MagicMock(),
            on_terminal_turn_persisted=on_terminal_turn_persisted,
        ),
    )


def _load_with_recovery(
    store: TurnStore,
    *,
    original_event_id: str,
    recovery_record: TurnRecord | None,
) -> TurnRecord | None:
    room = MagicMock(room_id="!room:example.org")
    with patch.object(store, "_load_persisted_turn_record", return_value=recovery_record):
        return store.load_turn(
            room=room,
            thread_id=None,
            original_event_id=original_event_id,
            requester_user_id="@user:example.org",
        )


@dataclass
class _FakeAgentStorage:
    session: AgentSession | TeamSession | None
    upserted_session: AgentSession | TeamSession | None = None

    def get_session(self, session_id: str, _session_type: object) -> AgentSession | TeamSession | None:
        if self.session is None or self.session.session_id != session_id:
            return None
        return self.session

    def upsert_session(self, session: AgentSession | TeamSession) -> None:
        self.upserted_session = session

    def close(self) -> None:
        return None


def _store_with_storage(
    tmp_path: Path,
    storage: _FakeAgentStorage,
    *,
    agent_name: str = "agent",
) -> TurnStore:
    state_writer = MagicMock()
    state_writer.create_storage.return_value = storage
    state_writer.session_type_for_scope.return_value = SessionType.AGENT
    state_writer.history_scope.return_value = HistoryScope(kind="agent", scope_id="agent")
    return TurnStore(
        TurnStoreDeps(
            agent_name=agent_name,
            tracking_base_path=tmp_path,
            state_writer=state_writer,
            resolver=MagicMock(),
            tool_runtime=MagicMock(),
        ),
    )


def _owned_turn_record(target: MessageTarget) -> TurnRecord:
    return TurnRecord.create(
        ["$user_msg"],
        response_event_id="$reply",
        response_owner="agent",
        requester_id="@user:example.org",
        history_scope=HistoryScope(kind="agent", scope_id="agent"),
        conversation_target=target,
    )


def _prepare_redaction(
    store: TurnStore,
    target: MessageTarget,
    *,
    redacted_event_id: str = "$user_msg",
) -> bool:
    """Tombstone one source and run the next response's locked cleanup gate."""
    store.mark_source_redacted(redacted_event_id)
    return store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$later",),
    )


def test_only_terminal_turn_notifies_exact_indexed_event_ids(tmp_path: Path) -> None:
    """Pending TurnStore state must not retire callback work that startup cannot resume."""
    notifications: list[tuple[str, ...]] = []
    store = _store(tmp_path, on_terminal_turn_persisted=notifications.append)
    pending = TurnRecord.create(
        ["$source"],
        discovery_event_ids=["$alias"],
        completed=False,
    )

    store.record_pending_turn(pending)
    assert notifications == []

    store.record_turn(pending)
    store._ledger.flush()

    assert notifications == [("$source", "$alias")]


def test_user_stop_durably_terminates_the_turn_that_owns_the_response(tmp_path: Path) -> None:
    """A user stop must become durable turn truth before dispatch settles it."""
    notifications: list[tuple[str, ...]] = []
    store = _store(tmp_path, on_terminal_turn_persisted=notifications.append)
    target = MessageTarget.resolve("!room:example.org", None, "$source")
    pending = TurnRecord.create(
        ["$source"],
        response_event_id="$reply",
        completed=False,
        response_owner="agent",
        requester_id="@user:example.org",
        conversation_target=target,
    )
    store.record_pending_turn(pending)

    stop_receipt_order = 2
    stopped = store.record_user_stopped_response("$reply", stop_receipt_order)

    assert stopped is not None
    assert stopped.completed is True
    assert stopped.user_stop_receipt_order == stop_receipt_order
    assert stopped.user_stop_settled_receipt_order is None
    assert store.is_durably_handled("$source") is True
    assert notifications == [("$source",)]
    retried = store.record_user_stopped_response("$reply", stop_receipt_order)
    assert retried is not None
    assert retried.completed is True
    assert retried.user_stop_receipt_order == stop_receipt_order
    assert retried.user_stop_settled_receipt_order is None
    assert notifications == [("$source",)]

    finalized = store.record_user_stopped_response(
        "$reply",
        stop_receipt_order,
        delivery_settled=True,
    )
    assert finalized is not None
    assert finalized.user_stop_settled_receipt_order == stop_receipt_order
    assert notifications == [("$source",)]


def test_settled_user_stop_retry_reenters_durable_persistence(tmp_path: Path) -> None:
    """A failed settled-STOP write must not make an in-memory retry look durable."""
    store = _store(tmp_path)
    target = MessageTarget.resolve("!room:example.org", None, "$source")
    store.record_pending_turn(
        TurnRecord.create(
            ["$source"],
            response_event_id="$reply",
            completed=False,
            response_owner="agent",
            requester_id="@user:example.org",
            conversation_target=target,
        ),
    )
    stop_receipt_order = 2
    store.record_user_stopped_response("$reply", stop_receipt_order)
    persist_attempts = 0

    def fail_persist(_records: tuple[TurnRecord, ...]) -> None:
        nonlocal persist_attempts
        persist_attempts += 1
        msg = "disk unavailable"
        raise OSError(msg)

    with (
        patch.object(store._ledger, "_persist_records", side_effect=fail_persist),
        patch.object(store._ledger, "_schedule_persist_retry_locked"),
    ):
        with pytest.raises(OSError, match="disk unavailable"):
            store.record_user_stopped_response(
                "$reply",
                stop_receipt_order,
                delivery_settled=True,
            )
        assert persist_attempts == 2

        with pytest.raises(OSError, match="disk unavailable"):
            store.record_user_stopped_response(
                "$reply",
                stop_receipt_order,
                delivery_settled=True,
            )
        assert persist_attempts == 4


@pytest.mark.parametrize("invalid_receipt_order", [0, -1, True])
def test_user_stop_rejects_invalid_receipt_order(tmp_path: Path, invalid_receipt_order: int) -> None:
    """STOP ordering requires a real positive durable admission sequence."""
    with pytest.raises(ValueError, match="receipt order must be positive"):
        _store(tmp_path).record_user_stopped_response("$reply", invalid_receipt_order)


def test_locked_pending_response_preparation_suppresses_a_concurrent_user_stop(tmp_path: Path) -> None:
    """A response queued before STOP must recheck terminal truth after taking its lock."""
    store = _store(tmp_path)
    target = MessageTarget.resolve("!room:example.org", None, "$source")
    pending = TurnRecord.create(
        ["$source"],
        response_event_id="$reply",
        completed=False,
        response_owner="agent",
        requester_id="@user:example.org",
        conversation_target=target,
    )
    store.record_pending_turn(pending)
    assert (
        store.prepare_pending_response_source(
            target=target,
            source_event_ids=("$source",),
            terminal_source_event_ids=("$source",),
        )
        is False
    )

    store.record_user_stopped_response("$reply", 2)

    assert (
        store.prepare_pending_response_source(
            target=target,
            source_event_ids=("$source",),
            terminal_source_event_ids=("$source",),
        )
        is True
    )


def test_locked_edit_preparation_uses_stop_order_and_settles_superseded_delivery(tmp_path: Path) -> None:
    """Only later edits run, and they durably supersede an older STOP delivery."""
    store = _store(tmp_path)
    target = MessageTarget.resolve("!room:example.org", None, "$source")
    store.record_turn(
        TurnRecord.create(
            ["$source"],
            response_event_id="$reply",
            response_owner="agent",
            requester_id="@user:example.org",
            conversation_target=target,
            user_stop_receipt_order=2,
        ),
    )

    assert store.prepare_edit_response_source(
        target=target,
        source_event_ids=("$source",),
        response_event_id="$reply",
        edit_receipt_order=1,
    )
    stopped = store.get_turn_record("$source")
    assert stopped is not None
    assert stopped.latest_edit_receipt_order is None
    assert stopped.user_stop_settled_receipt_order is None

    assert not store.prepare_edit_response_source(
        target=target,
        source_event_ids=("$source",),
        response_event_id="$reply",
        edit_receipt_order=3,
    )
    reopened = store.get_turn_record("$source")
    assert reopened is not None
    assert reopened.latest_edit_receipt_order == 3
    assert reopened.user_stop_settled_receipt_order == 2


def test_pending_delivery_intent_does_not_require_model_history_scope(tmp_path: Path) -> None:
    """Response ownership and target distinguish delivery intent from a raw visible echo."""
    store = _store(tmp_path)
    source_event_ids = ("$source",)
    visible_echo = TurnRecord.create(
        source_event_ids,
        response_event_id="$echo",
        completed=False,
    )
    store.record_pending_turn(visible_echo)
    assert store.has_pending_response_intent(source_event_ids) is False

    target = MessageTarget.resolve("!room:example.org", None, "$source")
    delivery_intent = store.attach_response_context(
        visible_echo,
        history_scope=None,
        conversation_target=target,
    )
    store.record_pending_turn(delivery_intent)

    assert store.has_pending_response_intent(source_event_ids) is True
    record = store.get_turn_record("$source")
    assert record is not None
    assert record.response_owner == "agent"
    assert record.conversation_target == target
    assert record.history_scope is None


def test_terminal_turn_notifies_only_after_durable_write(tmp_path: Path) -> None:
    """Dispatch truth must remain until its replacing TurnStore write reaches disk."""
    notified = threading.Event()
    persist_started = threading.Event()
    release_persist = threading.Event()
    store = _store(tmp_path, on_terminal_turn_persisted=lambda _event_ids: notified.set())
    original_persist = store._ledger._persist_records

    def blocking_persist(records: tuple[TurnRecord, ...]) -> None:
        persist_started.set()
        assert release_persist.wait(timeout=2)
        original_persist(records)

    with patch.object(store._ledger, "_persist_records", side_effect=blocking_persist):
        record_thread = threading.Thread(
            target=store.record_turn,
            args=(TurnRecord.create(["$source"], response_event_id="$response"),),
        )
        record_thread.start()
        try:
            assert persist_started.wait(timeout=2)
            assert not notified.wait(timeout=0.1)
            record_thread.join(timeout=0.1)
            assert not record_thread.is_alive()
        finally:
            release_persist.set()
            record_thread.join(timeout=2)

    assert not record_thread.is_alive()
    assert notified.wait(timeout=2)


def test_pending_turn_claim_allows_only_one_concurrent_owner(tmp_path: Path) -> None:
    """Overlapping delivery of one source event must start one response."""
    store = _store(tmp_path)
    turn = TurnRecord.create(["$source"], completed=False)
    barrier = threading.Barrier(8)
    claims = [False] * barrier.parties

    def claim(index: int) -> None:
        barrier.wait()
        claims[index] = store.try_claim_turn(turn)

    threads = [threading.Thread(target=claim, args=(index,)) for index in range(barrier.parties)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    assert sum(claims) == 1
    store.release_pending_turn_claim(turn)
    assert store.try_claim_turn(turn) is True


def test_discovery_alias_allows_original_but_excludes_second_relay(tmp_path: Path) -> None:
    """One human original may overlap its relay, but two relays for it may not."""
    store = _store(tmp_path)
    original = TurnRecord.create(["$human"], completed=False)
    first_relay = TurnRecord.create(["$relay-one"], discovery_event_ids=["$human"], completed=False)
    duplicate_relay = TurnRecord.create(["$relay-two"], discovery_event_ids=["$human"], completed=False)

    assert store.try_claim_turn(original) is True
    assert store.try_claim_turn(first_relay) is True
    assert store.try_claim_turn(duplicate_relay) is False

    store.release_pending_turn_claim(first_relay)
    assert store.try_claim_turn(duplicate_relay) is True
    store.release_pending_turn_claim(duplicate_relay)
    store.release_pending_turn_claim(original)


@pytest.mark.parametrize("completed_claim", [False, True])
def test_same_turn_can_reclaim_its_handled_alias(tmp_path: Path, *, completed_claim: bool) -> None:
    """A relay turn must remain claimable for its own edit or restart drain."""
    store = _store(tmp_path)
    completed = TurnRecord.create(["$relay"], discovery_event_ids=["$human"])
    store.record_turn(completed)
    claim = replace(completed, completed=completed_claim)

    assert store.try_claim_turn(claim) is True

    store.release_pending_turn_claim(claim)


def test_pending_coalesced_turn_can_reclaim_tombstoned_alias(tmp_path: Path) -> None:
    """A sibling edit remains claimable after another alias is tombstoned."""
    store = _store(tmp_path)
    pending = TurnRecord.create(
        ["$relay-one", "$relay-two"],
        discovery_event_ids=["$human-one", "$human-two"],
        redacted_source_event_ids=["$human-two"],
        completed=False,
    )
    store.record_pending_turn(pending)

    assert store.is_handled("$human-two") is True
    assert store.try_claim_turn(pending) is True

    store.release_pending_turn_claim(pending)


@pytest.mark.asyncio
async def test_turn_settlement_waits_for_pending_claim_release(tmp_path: Path) -> None:
    """A waiter should remain blocked until response ownership reaches its existing release seam."""
    store = _store(tmp_path)
    turn = TurnRecord.create(["$source"], completed=False)
    assert store.try_claim_turn(turn) is True
    wait_started = asyncio.Event()

    async def wait_for_settlement() -> None:
        wait_started.set()
        await store.wait_for_turn_settled(turn.indexed_event_ids)

    waiter = asyncio.create_task(wait_for_settlement())
    await wait_started.wait()
    assert not waiter.done()

    store.release_pending_turn_claim(turn)
    await waiter


@pytest.mark.asyncio
async def test_distinct_physical_claims_can_share_alias_until_both_settle(tmp_path: Path) -> None:
    """A discovery alias coordinates settlement without rejecting its physical relay."""
    store = _store(tmp_path)
    original = TurnRecord.create(["$human"], completed=False)
    relay = TurnRecord.create(["$relay"], discovery_event_ids=["$human"], completed=False)
    assert store.try_claim_turn(original) is True
    assert store.try_claim_turn(relay) is True
    first_wait_started = asyncio.Event()

    async def wait_for_alias(started: asyncio.Event) -> None:
        started.set()
        await store.wait_for_turn_settled(("$human",))

    waiter = asyncio.create_task(wait_for_alias(first_wait_started))
    await first_wait_started.wait()
    assert not waiter.done()

    store.release_pending_turn_claim(original)
    second_wait_started = asyncio.Event()
    second_waiter = asyncio.create_task(wait_for_alias(second_wait_started))
    await second_wait_started.wait()
    assert not second_waiter.done()
    store.release_pending_turn_claim(relay)
    await asyncio.gather(waiter, second_waiter)


def test_turn_settlement_wait_does_not_consume_default_executor(tmp_path: Path) -> None:
    """Claim settlement must progress while every default-executor worker is occupied."""
    store = _store(tmp_path)
    turn = TurnRecord.create(["$source"], completed=False)

    async def probe() -> None:
        loop = asyncio.get_running_loop()
        loop.set_default_executor(ThreadPoolExecutor(max_workers=1))
        worker_started = asyncio.Event()
        release_worker = threading.Event()

        def occupy_worker() -> None:
            loop.call_soon_threadsafe(worker_started.set)
            release_worker.wait()

        blocker = asyncio.create_task(asyncio.to_thread(occupy_worker))
        await worker_started.wait()
        try:
            assert store.try_claim_turn(turn) is True
            wait_started = asyncio.Event()

            async def wait_for_settlement() -> None:
                wait_started.set()
                await store.wait_for_turn_settled(turn.indexed_event_ids)

            waiter = asyncio.create_task(wait_for_settlement())
            await wait_started.wait()
            assert not waiter.done()
            store.release_pending_turn_claim(turn)
            async with asyncio.timeout(1):
                await waiter
        finally:
            release_worker.set()
            await blocker

    asyncio.run(probe())


@pytest.mark.asyncio
async def test_failed_claimed_response_releases_turn_for_replay(tmp_path: Path) -> None:
    """A failed owner must not permanently suppress a later delivery."""
    store = _store(tmp_path)
    turn = TurnRecord.create(["$source"], completed=False)
    assert store.try_claim_turn(turn) is True

    async def fail() -> None:
        msg = "response failed"
        raise RuntimeError(msg)

    controller = SimpleNamespace(deps=SimpleNamespace(turn_store=store))
    with pytest.raises(RuntimeError, match="response failed"):
        await _run_claimed_response(controller, turn, fail())

    assert store.try_claim_turn(turn) is True


@pytest.mark.asyncio
async def test_terminal_turn_keeps_claim_until_response_task_finishes(tmp_path: Path) -> None:
    """Terminal persistence must not reopen a source before task cleanup completes."""
    store = _store(tmp_path)
    turn = TurnRecord.create(["$source"], completed=False)
    other_turn = TurnRecord.create(["$discovery"], completed=False)
    expanded_turn = replace(turn, discovery_event_ids=other_turn.source_event_ids)
    terminal_recorded = asyncio.Event()
    release_response = asyncio.Event()
    assert store.try_claim_turn(turn) is True
    assert store.try_claim_turn(other_turn) is True

    async def finish_response() -> None:
        store.record_turn(expanded_turn)
        terminal_recorded.set()
        await release_response.wait()

    controller = SimpleNamespace(deps=SimpleNamespace(turn_store=store))
    response_task = asyncio.create_task(_run_claimed_response(controller, turn, finish_response()))
    await terminal_recorded.wait()

    assert store.try_claim_turn(turn) is False
    release_response.set()
    await response_task
    assert store.try_claim_turn(turn) is True
    assert store.try_claim_turn(other_turn) is False


def test_prepare_redaction_removes_causal_run_suffix(tmp_path: Path) -> None:
    """Redacting a source must delete later output that may depend on that run."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[
            RunOutput(session_id=target.session_id, metadata={"matrix_event_id": "$user_msg"}),
            RunOutput(session_id=target.session_id, metadata={"matrix_event_id": "$other"}),
        ],
    )
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    store.record_turn(_owned_turn_record(target))

    should_suppress = _prepare_redaction(store, target)

    assert should_suppress is False
    assert storage.upserted_session is session
    assert session.runs == []


def test_edit_then_redaction_never_replays_the_old_causal_suffix(tmp_path: Path) -> None:
    """Editing a source must remove later output before its replacement is persisted."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$source")
    scope = HistoryScope(kind="agent", scope_id="agent")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[
            RunOutput(
                session_id=target.session_id,
                metadata={
                    constants.MATRIX_EVENT_ID_METADATA_KEY: "$source",
                    constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY: ["$source"],
                    constants.MATRIX_SEEN_EVENT_IDS_METADATA_KEY: ["$source"],
                },
            ),
            RunOutput(
                session_id=target.session_id,
                metadata={
                    constants.MATRIX_EVENT_ID_METADATA_KEY: "$dependent",
                    constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY: ["$dependent"],
                    constants.MATRIX_SEEN_EVENT_IDS_METADATA_KEY: ["$dependent"],
                },
            ),
        ],
    )
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    source_record = TurnRecord.create(
        ["$source"],
        response_event_id="$reply",
        response_owner="agent",
        requester_id="@user:example.org",
        history_scope=scope,
        conversation_target=target,
    )
    store.record_turn(source_record)

    store.remove_stale_runs_for_edit(
        turn_record=source_record,
        requester_user_id="@user:example.org",
    )
    assert session.runs == []

    session.runs = [
        RunOutput(
            session_id=target.session_id,
            metadata={
                constants.MATRIX_EVENT_ID_METADATA_KEY: "$source",
                constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY: ["$source"],
                constants.MATRIX_SEEN_EVENT_IDS_METADATA_KEY: ["$source"],
            },
        ),
    ]
    should_suppress = _prepare_redaction(store, target, redacted_event_id="$source")

    assert should_suppress is False
    assert session.runs == []


def test_prepare_redaction_removes_runs_that_consumed_the_source(tmp_path: Path) -> None:
    """A later run that consumed redacted context must not remain eligible for replay."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[
            RunOutput(
                session_id=target.session_id,
                metadata={
                    constants.MATRIX_EVENT_ID_METADATA_KEY: "$later",
                    constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY: ["$later"],
                    constants.MATRIX_SEEN_EVENT_IDS_METADATA_KEY: ["$user_msg", "$later"],
                },
            ),
        ],
    )
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    store.record_turn(_owned_turn_record(target))

    should_suppress = _prepare_redaction(store, target)

    assert should_suppress is False
    assert session.runs == []


def test_prepare_redaction_removes_source_from_every_recorded_history_scope(tmp_path: Path) -> None:
    """Later ad-hoc responses must not retain a source consumed outside its original scope."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    agent_scope = HistoryScope(kind="agent", scope_id="agent")
    team_scope = HistoryScope(kind="team", scope_id="team_private")
    agent_session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[RunOutput(session_id=target.session_id, metadata={"matrix_event_id": "$user_msg"})],
    )
    team_session = TeamSession(
        session_id=target.session_id,
        team_id=team_scope.scope_id,
        runs=[
            TeamRunOutput(
                session_id=target.session_id,
                team_id=team_scope.scope_id,
                metadata={
                    constants.MATRIX_EVENT_ID_METADATA_KEY: "$later",
                    constants.MATRIX_SEEN_EVENT_IDS_METADATA_KEY: ["$user_msg", "$later"],
                },
            ),
        ],
    )
    storages = {
        agent_scope.key: _FakeAgentStorage(agent_session),
        team_scope.key: _FakeAgentStorage(team_session),
    }
    state_writer = MagicMock()
    state_writer.create_storage.side_effect = lambda _identity, *, scope: storages[scope.key]
    state_writer.session_type_for_scope.side_effect = lambda scope: (
        SessionType.TEAM if scope.kind == "team" else SessionType.AGENT
    )
    state_writer.history_scope.return_value = agent_scope
    store = TurnStore(
        TurnStoreDeps(
            agent_name="agent",
            tracking_base_path=tmp_path,
            state_writer=state_writer,
            resolver=MagicMock(),
            tool_runtime=MagicMock(),
        ),
    )
    store.record_turn(_owned_turn_record(target))
    store.record_turn(
        TurnRecord.create(
            ["$later"],
            response_event_id="$later-reply",
            response_owner="agent",
            requester_id="@user:example.org",
            history_scope=team_scope,
            conversation_target=target,
        ),
    )

    should_suppress = _prepare_redaction(store, target)

    assert should_suppress is False
    assert agent_session.runs == []
    assert team_session.runs == []


@pytest.mark.parametrize("source_owner", [None, "other"])
def test_prepare_redaction_cleans_later_owned_scopes_across_requesters(
    tmp_path: Path,
    *,
    source_owner: str | None,
) -> None:
    """An unowned source can still contaminate a later private response scope."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    default_scope = HistoryScope(kind="agent", scope_id="agent")
    team_scope = HistoryScope(kind="team", scope_id="team_private")
    team_session = TeamSession(
        session_id=target.session_id,
        team_id=team_scope.scope_id,
        runs=[
            TeamRunOutput(
                session_id=target.session_id,
                team_id=team_scope.scope_id,
                metadata={
                    constants.MATRIX_EVENT_ID_METADATA_KEY: "$later",
                    constants.MATRIX_SEEN_EVENT_IDS_METADATA_KEY: ["$user_msg", "$later"],
                },
            ),
        ],
    )
    storages = {
        (default_scope.key, "@source:example.org"): _FakeAgentStorage(None),
        (team_scope.key, "@later:example.org"): _FakeAgentStorage(team_session),
    }
    state_writer = MagicMock()
    state_writer.history_scope.return_value = default_scope
    state_writer.create_storage.side_effect = lambda identity, *, scope: storages[(scope.key, identity)]
    state_writer.session_type_for_scope.side_effect = lambda scope: (
        SessionType.TEAM if scope.kind == "team" else SessionType.AGENT
    )
    tool_runtime = MagicMock()
    tool_runtime.build_execution_identity.side_effect = lambda *, target, user_id: user_id  # noqa: ARG005
    store = TurnStore(
        TurnStoreDeps(
            agent_name="agent",
            tracking_base_path=tmp_path,
            state_writer=state_writer,
            resolver=MagicMock(),
            tool_runtime=tool_runtime,
        ),
    )
    store.record_turn(
        replace(
            _owned_turn_record(target),
            response_owner=source_owner,
            requester_id="@source:example.org",
        ),
    )
    store.record_turn(
        TurnRecord.create(
            ["$later"],
            response_event_id="$later-reply",
            response_owner="agent",
            requester_id="@later:example.org",
            history_scope=team_scope,
            conversation_target=target,
        ),
    )

    should_suppress = _prepare_redaction(store, target)

    assert should_suppress is False
    assert team_session.runs == []
    assert {call.kwargs["user_id"] for call in tool_runtime.build_execution_identity.call_args_list} == {
        "@source:example.org",
        "@later:example.org",
    }


def test_tombstone_gains_cleanup_context_when_the_source_turn_registers(tmp_path: Path) -> None:
    """A redaction race should become cleanup work only when its source turn registers."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[RunOutput(session_id=target.session_id, metadata={"matrix_event_id": "$user_msg"})],
    )
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    marked = store.mark_source_redacted("$user_msg")
    assert marked is not None
    assert marked.conversation_target is None
    assert marked.pending_redaction_cleanup_event_ids == ()
    pending = store.record_pending_turn(
        TurnRecord.create(
            ["$user_msg"],
            completed=False,
            requester_id="@user:example.org",
            response_owner="agent",
            history_scope=HistoryScope(kind="agent", scope_id="agent"),
            conversation_target=target,
        ),
    )
    assert pending is not None
    assert pending.pending_redaction_cleanup_event_ids == ("$user_msg",)

    should_suppress = store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$user_msg",),
    )

    assert should_suppress is True
    assert session.runs == []
    cleaned = store.get_turn_record("$user_msg")
    assert cleaned is not None
    assert cleaned.pending_redaction_cleanup_event_ids == ()


def test_prepare_redaction_invalidates_compacted_replay(tmp_path: Path) -> None:
    """Redaction must remove content already folded into the durable summary."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    scope = HistoryScope(kind="agent", scope_id="agent")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[
            RunOutput(
                session_id=target.session_id,
                metadata={constants.MATRIX_EVENT_ID_METADATA_KEY: "$post-compaction"},
            ),
        ],
        summary=SessionSummary(summary="The user disclosed REDACTED_SECRET."),
    )
    update_scope_seen_event_ids(session, scope, ["$user_msg", "$older"])
    write_scope_state(
        session,
        scope,
        HistoryScopeState(last_summary_model="summary-model", compacted_run_ids=("$compacted-run",)),
    )
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    store.record_turn(_owned_turn_record(target))

    should_suppress = _prepare_redaction(store, target)

    assert should_suppress is False
    assert storage.upserted_session is session
    assert session.runs == []
    assert session.summary is None
    assert read_scope_seen_event_ids(session, scope) == set()
    assert read_scope_state(session, scope) == HistoryScopeState(compacted_run_ids=("$compacted-run",))


def test_redaction_before_response_registration_tombstones_pending_coalesced_turn(tmp_path: Path) -> None:
    """A source redacted before response startup must suppress its later pending batch."""
    store = _store(tmp_path)
    target = MessageTarget.resolve("!room:example.org", "$thread", "$second")
    team_scope = HistoryScope(kind="team", scope_id="team_private")

    store.mark_source_redacted("$first")
    pending = store.record_pending_turn(
        TurnRecord.create(
            ["$first", "$second"],
            source_event_prompts={"$first": "REDACTED_SECRET", "$second": "keep"},
            requester_id="@user:example.org",
            response_owner="agent",
            history_scope=team_scope,
            conversation_target=target,
        ),
    )

    assert pending is not None
    assert pending.completed is False
    assert pending.redacted_source_event_ids == ("$first",)
    assert pending.source_event_prompts == {"$second": "keep"}
    assert pending.history_scope == team_scope
    assert store.is_handled("$first") is True
    assert store.is_handled("$second") is False


def test_redaction_detaches_from_a_pending_coalesced_turn_after_sibling_completion(tmp_path: Path) -> None:
    """A split sibling identity must not veto the remaining source tombstone."""
    store = _store(tmp_path)
    target = MessageTarget.resolve("!room:example.org", "$thread", "$second")
    scope = HistoryScope(kind="agent", scope_id="agent")
    pending = TurnRecord.create(
        ["$first", "$second"],
        completed=False,
        requester_id="@user:example.org",
        response_owner="agent",
        history_scope=scope,
        conversation_target=target,
    )
    store.record_pending_turn(pending)
    store.record_turn(
        TurnRecord.create(
            ["$second"],
            response_event_id="$second-reply",
            requester_id="@user:example.org",
            response_owner="agent",
            history_scope=scope,
            conversation_target=target,
        ),
    )

    marked = store.mark_source_redacted("$first")

    assert marked is not None
    assert marked.source_event_ids == ("$first",)
    assert marked.anchor_event_id == "$first"
    assert marked.redacted_source_event_ids == ("$first",)
    assert marked.pending_redaction_cleanup_event_ids == ("$first",)
    completed_sibling = store.get_turn_record("$second")
    assert completed_sibling is not None
    assert completed_sibling.source_event_ids == ("$second",)
    assert completed_sibling.response_event_id == "$second-reply"


def test_redaction_cleanup_clears_after_pending_coalesced_turn_splits(tmp_path: Path) -> None:
    """A completed sibling must not leave an old alias cleanup pending forever."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$second")
    scope = HistoryScope(kind="agent", scope_id="agent")
    storage = _FakeAgentStorage(None)
    store = _store_with_storage(tmp_path, storage)
    pending = TurnRecord.create(
        ["$first", "$second"],
        completed=False,
        requester_id="@user:example.org",
        response_owner="agent",
        history_scope=scope,
        conversation_target=target,
    )
    store.record_pending_turn(pending)
    store.mark_source_redacted("$first")
    store.record_turn(
        TurnRecord.create(
            ["$second"],
            response_event_id="$second-reply",
            requester_id="@user:example.org",
            response_owner="agent",
            history_scope=scope,
            conversation_target=target,
        ),
    )

    should_suppress = store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$second",),
    )

    assert should_suppress is False
    cleaned = store.get_turn_record("$first")
    assert cleaned is not None
    assert cleaned.source_event_ids == ("$first",)
    assert cleaned.redacted_source_event_ids == ("$first",)
    assert cleaned.pending_redaction_cleanup_event_ids == ()
    completed_sibling = store.get_turn_record("$second")
    assert completed_sibling is not None
    assert completed_sibling.source_event_ids == ("$second",)
    assert completed_sibling.response_event_id == "$second-reply"


def test_redaction_cleanup_keeps_context_after_colliding_alias_projection(tmp_path: Path) -> None:
    """Projecting a redacted physical source must retain the context needed to sanitize it."""
    relay_event_id = "$relay"
    human_event_id = "$human"
    requester_user_id = "@alice:example.org"
    target = MessageTarget.resolve("!room:example.org", "$thread", human_event_id)
    scope = HistoryScope(kind="agent", scope_id="agent")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[
            RunOutput(
                session_id=target.session_id,
                metadata={constants.MATRIX_EVENT_ID_METADATA_KEY: human_event_id},
            ),
        ],
        summary=SessionSummary(summary="contains REDACTED_SECRET"),
    )
    update_scope_seen_event_ids(session, scope, [human_event_id])
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    store.record_pending_turn(
        TurnRecord.create(
            [relay_event_id, human_event_id],
            completed=False,
            source_event_metadata={
                relay_event_id: SourceEventMetadata(
                    sender="@bob:example.org",
                    discovery_event_id=human_event_id,
                ),
                human_event_id: SourceEventMetadata(sender=requester_user_id),
            },
            requester_id=requester_user_id,
            response_owner="agent",
            history_scope=scope,
            conversation_target=target,
        ),
    )
    store.record_turn(
        TurnRecord.create(
            [relay_event_id],
            response_event_id="$relay-reply",
            requester_id="@bob:example.org",
        ),
    )

    projected = store.mark_source_redacted(human_event_id)

    assert projected is not None
    assert projected.source_event_ids == (human_event_id,)
    assert projected.source_event_metadata == {}
    assert projected.requester_id == requester_user_id
    assert projected.requester_id_for_source(human_event_id) is None
    assert projected.pending_redaction_cleanup_event_ids == (human_event_id,)

    should_suppress = store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$later",),
    )

    assert should_suppress is False
    assert storage.upserted_session is session
    assert session.summary is None
    assert read_scope_seen_event_ids(session, scope) == set()
    cleaned = store.get_turn_record(human_event_id)
    assert cleaned is not None
    assert cleaned.pending_redaction_cleanup_event_ids == ()


def test_active_ad_hoc_team_redaction_uses_pending_response_scope(tmp_path: Path) -> None:
    """Post-lock cleanup must retain the exact team scope recorded before generation."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    scope = HistoryScope(kind="team", scope_id="team_private")
    session = TeamSession(
        session_id=target.session_id,
        team_id=scope.scope_id,
        runs=[
            TeamRunOutput(
                session_id=target.session_id,
                team_id=scope.scope_id,
                metadata={constants.MATRIX_EVENT_ID_METADATA_KEY: "$user_msg"},
            ),
        ],
    )
    storage = MagicMock()
    storage.get_session.return_value = session
    state_writer = MagicMock()
    state_writer.create_storage.return_value = storage
    state_writer.session_type_for_scope.return_value = SessionType.TEAM
    state_writer.history_scope.return_value = HistoryScope(kind="agent", scope_id="agent")
    store = TurnStore(
        TurnStoreDeps(
            agent_name="agent",
            tracking_base_path=tmp_path,
            state_writer=state_writer,
            resolver=MagicMock(),
            tool_runtime=MagicMock(),
        ),
    )
    response_record = TurnRecord.create(
        ["$user_msg"],
        requester_id="@user:example.org",
        response_owner="agent",
        history_scope=scope,
        conversation_target=target,
    )
    store.record_pending_turn(response_record)
    store.mark_source_redacted("$user_msg")
    store.record_turn(replace(response_record, response_event_id="$reply"))

    should_suppress = store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$later",),
    )

    assert should_suppress is False
    assert session.runs == []
    state_writer.create_storage.assert_called_with(ANY, scope=scope)


def test_redaction_sanitizes_coalesced_ledger_prompt_and_metadata(tmp_path: Path) -> None:
    """Sibling edit regeneration must not recover a redacted coalesced prompt."""
    store = _store(tmp_path)
    target = MessageTarget.resolve("!room:example.org", "$thread", "$second")
    store.record_turn(
        TurnRecord.create(
            ["$first", "$second"],
            source_event_prompts={"$first": "REDACTED_SECRET", "$second": "keep"},
            source_event_metadata={
                "$first": SourceEventMetadata(sender="@first:example.org"),
                "$second": SourceEventMetadata(sender="@second:example.org"),
            },
            requester_id="@user:example.org",
            response_owner="agent",
            history_scope=HistoryScope(kind="agent", scope_id="agent"),
            conversation_target=target,
        ),
    )

    sanitized = store.mark_source_redacted("$first")

    assert sanitized is not None
    assert sanitized.redacted_source_event_ids == ("$first",)
    assert sanitized.source_event_prompts == {"$second": "keep"}
    assert sanitized.source_event_metadata == {
        "$second": SourceEventMetadata(sender="@second:example.org"),
    }
    assert store.get_turn_record("$second") == sanitized


@pytest.mark.parametrize("terminal", [False, True])
def test_turn_merge_preserves_redacted_discovery_alias(tmp_path: Path, *, terminal: bool) -> None:
    """Backfilled aliases must retain their tombstone and cleanup intent across turn merges."""
    store = _store(tmp_path)
    existing = TurnRecord.create(
        ["$question"],
        discovery_event_ids=["$selection"],
        redacted_source_event_ids=["$selection"],
        pending_redaction_cleanup_event_ids=["$selection"],
        completed=False,
    )
    store.record_pending_turn(existing)
    incoming = TurnRecord.create(
        ["$question"],
        response_event_id="$response" if terminal else None,
        completed=terminal,
    )

    if terminal:
        store.record_turn(incoming)
    else:
        store.record_pending_turn(incoming)

    merged = store.get_turn_record("$selection")
    assert merged is not None
    assert merged.discovery_event_ids == ("$selection",)
    assert merged.redacted_source_event_ids == ("$selection",)
    assert merged.pending_redaction_cleanup_event_ids == ("$selection",)


def test_multi_bot_redaction_only_queues_cleanup_for_the_bot_with_context(tmp_path: Path) -> None:
    """Bots that never handled a source must not accumulate or probe cleanup work."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    scope = HistoryScope(kind="agent", scope_id="agent")
    owner_session = AgentSession(
        session_id=target.session_id,
        agent_id="owner",
        runs=[],
        summary=SessionSummary(summary="contains REDACTED_SECRET"),
    )
    update_scope_seen_event_ids(owner_session, scope, ["$user_msg"])
    owner_store = _store_with_storage(
        tmp_path,
        _FakeAgentStorage(owner_session),
        agent_name="owner",
    )
    owner_store.record_turn(
        replace(
            _owned_turn_record(target),
            response_owner="owner",
        ),
    )
    unrelated_session = AgentSession(
        session_id=target.session_id,
        agent_id="unrelated",
        runs=[],
        summary=SessionSummary(summary="unrelated"),
    )
    unrelated_store = _store_with_storage(
        tmp_path,
        _FakeAgentStorage(unrelated_session),
        agent_name="unrelated",
    )

    owner_marked = owner_store.mark_source_redacted("$user_msg")
    unrelated_marked = unrelated_store.mark_source_redacted("$user_msg")

    assert owner_marked is not None
    assert owner_marked.pending_redaction_cleanup_event_ids == ("$user_msg",)
    assert unrelated_marked is not None
    assert unrelated_marked.redacted_source_event_ids == ("$user_msg",)
    assert unrelated_marked.pending_redaction_cleanup_event_ids == ()

    owner_store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$later",),
    )
    should_suppress = unrelated_store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$later",),
    )

    assert should_suppress is False
    assert owner_session.summary is None
    assert read_scope_seen_event_ids(owner_session, scope) == set()
    assert unrelated_session.summary is not None
    unrelated_store.deps.state_writer.create_storage.assert_not_called()


def test_turn_store_constructs_private_ledger_from_tracking_base_path(tmp_path: Path) -> None:
    """TurnStore should own its private ledger and persist through the tracking base path."""
    store = _store(tmp_path)

    store.record_turn(TurnRecord.create(["$event"], response_event_id="$response"))

    reloaded_store = _store(tmp_path)

    assert reloaded_store.is_handled("$event")
    turn_record = reloaded_store.get_turn_record("$event")
    assert turn_record is not None
    assert turn_record.response_event_id == "$response"


def test_redaction_tombstone_persists_across_ledger_reload(tmp_path: Path) -> None:
    """A crash after cache mutation must not lose the source-redaction barrier."""
    store = _store(tmp_path)
    target = MessageTarget.resolve("!room:example.org", "$thread", "$event")
    store.record_turn(
        TurnRecord.create(
            ["$event", "$sibling"],
            source_event_prompts={"$event": "REDACTED_SECRET", "$sibling": "keep"},
            requester_id="@user:example.org",
            response_owner="agent",
            history_scope=HistoryScope(kind="agent", scope_id="agent"),
            conversation_target=target,
        ),
    )

    store.mark_source_redacted("$event")
    _reset_handled_turn_ledger_runtime()
    reloaded_store = _store(tmp_path)

    reloaded_record = reloaded_store.get_turn_record("$event")
    assert reloaded_record is not None
    assert reloaded_record.redacted_source_event_ids == ("$event",)
    assert reloaded_record.source_event_prompts == {"$sibling": "keep"}
    assert reloaded_store.is_handled("$event") is True
    assert reloaded_store.is_handled("$sibling") is True


def test_redaction_barrier_ignores_unrelated_prior_persist_failure(tmp_path: Path) -> None:
    """An older failed write must not prevent the redaction tombstone from becoming durable."""
    store = _store(tmp_path)
    real_persist = store._ledger._persist_records
    unrelated_failed = threading.Event()
    failed_once = False

    def persist_with_unrelated_failure(turn_records: tuple[TurnRecord, ...]) -> None:
        nonlocal failed_once
        if not failed_once and any("$unrelated" in record.indexed_event_ids for record in turn_records):
            failed_once = True
            unrelated_failed.set()
            message = "unrelated persist failed"
            raise OSError(message)
        real_persist(turn_records)

    with patch.object(store._ledger, "_persist_records", side_effect=persist_with_unrelated_failure):
        store.record_visible_echo("$unrelated", "$echo")
        assert unrelated_failed.wait(timeout=5)
        marked = store.mark_source_redacted("$redacted")

        assert marked is not None
        store._ledger.flush()

    _reset_handled_turn_ledger_runtime()
    durable_record = _store(tmp_path).get_turn_record("$redacted")
    assert durable_record is not None
    assert durable_record.redacted_source_event_ids == ("$redacted",)
    assert durable_record.pending_redaction_cleanup_event_ids == ()


def test_warm_preserves_lazy_cleanup_until_next_response(tmp_path: Path) -> None:
    """A restart must retain replay cleanup for the conversation's next response."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[RunOutput(session_id=target.session_id, metadata={"matrix_event_id": "$user_msg"})],
    )
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    store.record_turn(_owned_turn_record(target))
    marked = store.mark_source_redacted("$user_msg")

    assert marked is not None
    assert marked.pending_redaction_cleanup_event_ids == ("$user_msg",)
    _reset_handled_turn_ledger_runtime()
    restarted_store = _store_with_storage(tmp_path, storage)

    restarted_store.warm()

    assert len(session.runs or []) == 1
    restarted_record = restarted_store.get_turn_record("$user_msg")
    assert restarted_record is not None
    assert restarted_record.redacted_source_event_ids == ("$user_msg",)
    assert restarted_record.pending_redaction_cleanup_event_ids == ("$user_msg",)
    assert (
        restarted_store._prepare_response_for_redactions(
            target=target,
            source_event_ids=("$later",),
        )
        is False
    )
    assert session.runs == []
    cleaned_record = restarted_store.get_turn_record("$user_msg")
    assert cleaned_record is not None
    assert cleaned_record.pending_redaction_cleanup_event_ids == ()


def test_locked_response_preparation_sanitizes_and_acknowledges_history_cleanup(tmp_path: Path) -> None:
    """The under-lock gate removes replay and acknowledges the completed work."""
    target = MessageTarget.resolve("!room:example.org", "$thread", "$user_msg")
    session = AgentSession(
        session_id=target.session_id,
        agent_id="agent",
        runs=[RunOutput(session_id=target.session_id, metadata={"matrix_event_id": "$user_msg"})],
    )
    storage = _FakeAgentStorage(session)
    store = _store_with_storage(tmp_path, storage)
    store.record_turn(_owned_turn_record(target))
    store.mark_source_redacted("$user_msg")

    should_suppress = store._prepare_response_for_redactions(
        target=target,
        source_event_ids=("$later",),
    )

    assert should_suppress is False
    assert session.runs == []
    record = store.get_turn_record("$user_msg")
    assert record is not None
    assert record.pending_redaction_cleanup_event_ids == ()


def test_turn_record_codec_projects_and_parses_one_versioned_run_schema() -> None:
    """The same codec should own both run projection and recovery parsing."""
    history_scope = HistoryScope(kind="agent", scope_id="agent")
    target = MessageTarget.resolve("!room:example.org", "$thread", "$anchor")
    turn_record = TurnRecord.create(
        ["$first", "$anchor"],
        discovery_event_ids=["$selection"],
        redacted_source_event_ids=["$first"],
        response_event_id="$response",
        source_event_prompts={"$first": "first", "$anchor": "anchor"},
        source_event_metadata={
            "$first": SourceEventMetadata(sender="@alice:example.org", timestamp_ms=1_774_019_700_000),
        },
        response_owner="agent",
        requester_id="@user:example.org",
        correlation_id="corr-1",
        history_scope=history_scope,
        conversation_target=target,
    )

    metadata = TurnRecordCodec.to_run_metadata(turn_record)
    metadata.update(
        {
            constants.MATRIX_EVENT_ID_METADATA_KEY: "$anchor",
            constants.MATRIX_RESPONSE_EVENT_ID_METADATA_KEY: "$response",
            "requester_id": "@user:example.org",
            "correlation_id": "corr-1",
        },
    )
    parsed = TurnRecordCodec.from_run_metadata(metadata)

    assert metadata[constants.MATRIX_TURN_SCHEMA_VERSION_METADATA_KEY] == TurnRecordCodec.schema_version()
    assert metadata[constants.MATRIX_TURN_DISCOVERY_EVENT_IDS_METADATA_KEY] == ["$selection"]
    assert metadata[constants.MATRIX_TURN_REDACTED_SOURCE_EVENT_IDS_METADATA_KEY] == ["$first"]
    assert parsed == turn_record


def test_turn_record_codec_preserves_physical_source_ownership_when_alias_id_collides() -> None:
    """A physical source must outrank another source's discovery alias with the same ID."""
    relay_event_id = "$relay"
    human_event_id = "$human"
    turn_record = TurnRecord.create(
        [relay_event_id, human_event_id],
        source_event_prompts={
            relay_event_id: "routed prompt",
            human_event_id: "physical prompt",
        },
        source_event_metadata={
            relay_event_id: SourceEventMetadata(
                sender="@bob:example.org",
                discovery_event_id=human_event_id,
            ),
            human_event_id: SourceEventMetadata(sender="@alice:example.org"),
        },
        requester_id="@bob:example.org",
    )

    run_metadata = TurnRecordCodec.to_run_metadata(turn_record)
    run_metadata[constants.MATRIX_EVENT_ID_METADATA_KEY] = human_event_id
    recovered = TurnRecordCodec.from_run_metadata(run_metadata)

    assert recovered is not None
    assert recovered.prompt_source_event_id(human_event_id) == human_event_id
    assert recovered.requester_id_for_source(human_event_id) == "@alice:example.org"
    assert recovered.requester_id_for_source(relay_event_id) == "@bob:example.org"


def test_physical_source_membership_outranks_alias_when_metadata_is_partial() -> None:
    """A missing physical metadata row must fail closed instead of resolving through a relay alias."""
    relay_event_id = "$relay"
    human_event_id = "$human"
    turn_record = TurnRecord.create(
        [relay_event_id, human_event_id],
        source_event_metadata={
            relay_event_id: SourceEventMetadata(
                sender="@bob:example.org",
                discovery_event_id=human_event_id,
            ),
        },
        requester_id="@bob:example.org",
    )

    assert turn_record.prompt_source_event_id(human_event_id) == human_event_id
    assert turn_record.requester_id_for_source(human_event_id) is None


def test_redacted_physical_source_does_not_tombstone_colliding_relay_alias() -> None:
    """Redacting a physical source must retain the sibling relay and its prompt."""
    relay_event_id = "$relay"
    human_event_id = "$human"
    turn_record = TurnRecord.create(
        [relay_event_id, human_event_id],
        redacted_source_event_ids=[human_event_id],
        source_event_prompts={
            relay_event_id: "routed prompt",
            human_event_id: "physical prompt",
        },
        source_event_metadata={
            relay_event_id: SourceEventMetadata(
                sender="@bob:example.org",
                discovery_event_id=human_event_id,
            ),
            human_event_id: SourceEventMetadata(sender="@alice:example.org"),
        },
        requester_id="@bob:example.org",
    )

    assert turn_record.prompt_source_event_id(human_event_id) == human_event_id
    assert turn_record.replay_source_event_ids == (relay_event_id,)
    assert turn_record.source_event_prompts == {relay_event_id: "routed prompt"}


def test_turn_record_codecs_preserve_explicit_unknown_source_ownership() -> None:
    """An explicit empty source map must survive persistence and disable singleton fallback."""
    event_id = "$source"
    turn_record = TurnRecord.create(
        [event_id],
        source_event_metadata={},
        requester_id="@stale:example.org",
    )

    ledger_recovered = TurnRecordCodec._from_ledger_record(
        event_id,
        TurnRecordCodec._to_ledger_record(turn_record),
    )
    run_metadata = TurnRecordCodec.to_run_metadata(turn_record)
    run_metadata[constants.MATRIX_EVENT_ID_METADATA_KEY] = event_id
    run_recovered = TurnRecordCodec.from_run_metadata(run_metadata)

    assert turn_record.source_event_metadata == {}
    assert ledger_recovered is not None
    assert ledger_recovered.source_event_metadata == {}
    assert run_recovered is not None
    assert run_recovered.source_event_metadata == {}
    assert run_recovered.requester_id_for_source(event_id) is None


def test_build_run_metadata_normalizes_discovery_aliases(tmp_path: Path) -> None:
    """Additional discovery IDs should share canonical source-ID normalization."""
    store = _store(tmp_path)
    turn_record = TurnRecord.create(["$first", "$anchor"])

    metadata = store.build_run_metadata(
        turn_record,
        additional_discovery_event_ids=("", "$first", "$selection", "$selection"),
    )

    assert metadata == {
        constants.MATRIX_TURN_SCHEMA_VERSION_METADATA_KEY: TurnRecordCodec.schema_version(),
        constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY: ["$first", "$anchor"],
        constants.MATRIX_TURN_DISCOVERY_EVENT_IDS_METADATA_KEY: ["$selection"],
    }


def test_discovery_alias_recovery_repairs_anchor_and_alias_rows(tmp_path: Path) -> None:
    """Missing-ledger recovery should index one non-coalesced turn by its anchor and discovery alias."""
    metadata = TurnRecordCodec.to_run_metadata(
        TurnRecord.create(
            ["$question"],
            response_owner="agent",
        ),
    )
    metadata[constants.MATRIX_TURN_DISCOVERY_EVENT_IDS_METADATA_KEY] = ["$selection"]
    metadata[constants.MATRIX_EVENT_ID_METADATA_KEY] = "$question"
    metadata[constants.MATRIX_RESPONSE_EVENT_ID_METADATA_KEY] = "$response"
    recovery_record = TurnRecordCodec.from_run_metadata(metadata)

    assert recovery_record is not None
    assert recovery_record.source_event_ids == ("$question",)
    assert recovery_record.discovery_event_ids == ("$selection",)
    assert not recovery_record.is_coalesced

    for lookup_event_id in ("$question", "$selection"):
        store = _store(tmp_path / lookup_event_id.removeprefix("$"))
        loaded = _load_with_recovery(
            store,
            original_event_id=lookup_event_id,
            recovery_record=recovery_record,
        )

        assert loaded is not None
        assert loaded.source_event_ids == ("$question",)
        assert loaded.discovery_event_ids == ("$selection",)
        for indexed_event_id in ("$question", "$selection"):
            repaired = store.get_turn_record(indexed_event_id)
            assert repaired is not None
            assert repaired.source_event_ids == ("$question",)
            assert repaired.discovery_event_ids == ("$selection",)
            assert store.is_handled(indexed_event_id)


def test_recovery_does_not_replace_a_conflicting_completed_identity(tmp_path: Path) -> None:
    """Repair missing aliases without overwriting another completed source turn."""
    store = _store(tmp_path)
    store.record_turn(TurnRecord.create(["$selection"], response_event_id="$selection-response"))
    recovery_record = TurnRecord.create(
        ["$question"],
        discovery_event_ids=["$selection"],
        response_event_id="$question-response",
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$question",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.source_event_ids == ("$question",)
    assert loaded.discovery_event_ids == ()
    assert loaded.indexed_event_ids == ("$question",)
    assert store.get_turn_record("$question") == loaded
    selection_record = store.get_turn_record("$selection")
    assert selection_record is not None
    assert selection_record.source_event_ids == ("$selection",)
    assert selection_record.response_event_id == "$selection-response"

    store._ledger.flush()
    _reset_handled_turn_ledger_runtime()
    reloaded_store = _store(tmp_path)
    assert reloaded_store.get_turn_record("$question") == loaded
    assert reloaded_store.get_turn_record("$selection") == selection_record


def test_newer_delivered_run_recovers_mutable_facts_after_crash(tmp_path: Path) -> None:
    """A delivered run newer than the ledger should repair the edit crash window."""
    store = _store(tmp_path)
    ledger_record = TurnRecord.create(
        ["$first", "$anchor"],
        response_event_id="$old-response",
        source_event_prompts={"$first": "old first", "$anchor": "old anchor"},
        source_event_revisions={
            "$first": (10, "$old-edit"),
        },
        visible_echo_event_id="$echo",
        timestamp=10,
    )
    store._ledger.record_handled_turn(ledger_record)
    recovery_record = TurnRecord.create(
        ["$first", "$anchor"],
        response_event_id="$new-response",
        source_event_prompts={"$first": "edited first", "$anchor": "old anchor"},
        source_event_revisions={
            "$first": (20, "$new-edit"),
        },
        response_owner="agent",
        timestamp=20,
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$first",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.source_event_ids == ledger_record.source_event_ids
    assert loaded.anchor_event_id == ledger_record.anchor_event_id
    assert loaded.response_event_id == "$new-response"
    assert loaded.source_event_prompts == {"$first": "edited first", "$anchor": "old anchor"}
    assert loaded.source_event_revisions == {
        "$first": (20, "$new-edit"),
    }
    assert loaded.visible_echo_event_id == "$echo"
    assert loaded.response_owner == "agent"
    assert loaded.timestamp == 20


def test_recovery_preserves_newer_ledger_only_sibling_edit(tmp_path: Path) -> None:
    """Recovery must merge edit facts per source instead of replacing the whole map."""
    store = _store(tmp_path)
    ledger_record = TurnRecord.create(
        ["$first", "$anchor"],
        response_event_id="$old-response",
        source_event_prompts={"$first": "old first", "$anchor": "suppressed anchor"},
        source_event_revisions={"$anchor": (30, "$anchor-edit")},
        timestamp=10,
    )
    store._ledger.record_handled_turn(ledger_record)
    recovery_record = TurnRecord.create(
        ["$first", "$anchor"],
        response_event_id="$new-response",
        source_event_prompts={"$first": "edited first", "$anchor": "old anchor"},
        source_event_revisions={"$first": (20, "$first-edit")},
        timestamp=20,
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$first",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.source_event_prompts == {
        "$first": "edited first",
        "$anchor": "suppressed anchor",
    }
    assert loaded.source_event_revisions == {
        "$first": (20, "$first-edit"),
        "$anchor": (30, "$anchor-edit"),
    }
    assert loaded.response_event_id == "$new-response"


def test_recovery_preserves_newer_routed_alias_prompt(tmp_path: Path) -> None:
    """A newer human-alias revision must carry its owned relay prompt through recovery."""
    store = _store(tmp_path)
    source_metadata = {
        "$relay": SourceEventMetadata(sender="@user:example.org", discovery_event_id="$human"),
        "$anchor": SourceEventMetadata(sender="@user:example.org"),
    }
    store._ledger.record_handled_turn(
        TurnRecord.create(
            ["$relay", "$anchor"],
            discovery_event_ids=["$human"],
            response_event_id="$old-response",
            source_event_prompts={"$relay": "new relay", "$anchor": "anchor"},
            source_event_revisions={"$human": (30, "$new-edit")},
            source_event_metadata=source_metadata,
            timestamp=10,
        ),
    )
    recovery_record = TurnRecord.create(
        ["$relay", "$anchor"],
        discovery_event_ids=["$human"],
        response_event_id="$new-response",
        source_event_prompts={"$relay": "stale relay", "$anchor": "anchor"},
        source_event_revisions={"$human": (20, "$stale-edit")},
        source_event_metadata=source_metadata,
        timestamp=20,
    )

    loaded = _load_with_recovery(store, original_event_id="$human", recovery_record=recovery_record)

    assert loaded is not None
    assert loaded.source_event_prompts == {"$relay": "new relay", "$anchor": "anchor"}
    assert loaded.source_event_revisions == {"$human": (30, "$new-edit")}


def test_recovery_without_prompts_preserves_durable_prompt_map(tmp_path: Path) -> None:
    """A delivered recovery lacking prompt metadata cannot erase durable coalesced bodies."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(
            ["$first", "$anchor"],
            response_event_id="$old-response",
            source_event_prompts={"$first": "first", "$anchor": "anchor"},
            timestamp=10,
        ),
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$first",
        recovery_record=TurnRecord.create(
            ["$first", "$anchor"],
            response_event_id="$new-response",
            timestamp=20,
        ),
    )

    assert loaded is not None
    assert loaded.source_event_prompts == {"$first": "first", "$anchor": "anchor"}


def test_recovery_preserves_explicit_unknown_source_ownership(tmp_path: Path) -> None:
    """A newer explicit unknown-ownership marker must not inherit stale ledger attribution."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(
            ["$event"],
            response_event_id="$old-response",
            source_event_metadata={
                "$event": SourceEventMetadata(sender="@stale:example.org"),
            },
            timestamp=10,
        ),
    )
    recovery_record = TurnRecord.create(
        ["$event"],
        response_event_id="$new-response",
        source_event_metadata={},
        requester_id="@current:example.org",
        timestamp=20,
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$event",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.response_event_id == "$new-response"
    assert loaded.source_event_metadata == {}
    assert loaded.requester_id_for_source("$event") is None


def test_routed_alias_redaction_marks_owning_relay_under_lock(tmp_path: Path) -> None:
    """Under-lock redaction checks must recognize a physical relay tombstoned by its alias."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(
            ["$relay", "$anchor"],
            discovery_event_ids=["$human"],
            response_event_id="$response",
            source_event_prompts={"$relay": "secret", "$anchor": "keep"},
            source_event_metadata={
                "$relay": SourceEventMetadata(sender="@user:example.org", discovery_event_id="$human"),
                "$anchor": SourceEventMetadata(sender="@user:example.org"),
            },
        ),
    )

    marked = store.mark_source_redacted("$human")

    assert marked is not None
    assert marked.source_event_prompts == {"$anchor": "keep"}
    assert store._any_source_redacted(("$relay",)) is True


def test_same_second_delivered_run_repairs_fractional_ledger_timestamp(tmp_path: Path) -> None:
    """Second-resolution run times should still repair a later run from the same second."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(["$event"], response_event_id="$old-response", timestamp=10.9),
    )
    recovery_record = TurnRecord.create(["$event"], response_event_id="$new-response", timestamp=10)

    loaded = _load_with_recovery(
        store,
        original_event_id="$event",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.response_event_id == "$new-response"
    assert loaded.timestamp > 10.9


def test_repeated_delivered_run_recovery_keeps_ledger_version_stable(tmp_path: Path) -> None:
    """Idempotent recovery should not rewrite the ledger with synthetic timestamp drift."""
    store = _store(tmp_path)
    ledger_record = TurnRecord.create(
        ["$event"],
        response_event_id="$response",
        response_owner="agent",
        timestamp=10,
    )
    store._ledger.record_handled_turn(ledger_record)
    recovery_record = TurnRecord.create(
        ["$event"],
        response_event_id="$response",
        response_owner="agent",
        timestamp=20,
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$event",
        recovery_record=recovery_record,
    )

    assert loaded == ledger_record
    assert store.get_turn_record("$event") == ledger_record


def test_newer_interrupted_run_keeps_delivered_ledger_outcome(tmp_path: Path) -> None:
    """A newer run without Matrix delivery must not replace a visible response."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(["$event"], response_event_id="$response", timestamp=10),
    )
    recovery_record = TurnRecord.create(["$event"], completed=False, timestamp=20)

    loaded = _load_with_recovery(
        store,
        original_event_id="$event",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.response_event_id == "$response"
    assert loaded.completed
    assert loaded.timestamp == 10


def test_interrupted_recovery_does_not_mix_prompt_and_revision(tmp_path: Path) -> None:
    """An unfinished run cannot pair its edit revision with the delivered ledger prompt."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(
            ["$event"],
            response_event_id="$response",
            source_event_prompts={"$event": "base prompt"},
            timestamp=10,
        ),
    )
    recovery_record = TurnRecord.create(
        ["$event"],
        completed=False,
        source_event_prompts={"$event": "edited prompt"},
        source_event_revisions={"$event": (20, "$edit")},
        timestamp=20,
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$event",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.source_event_prompts == {"$event": "base prompt"}
    assert loaded.source_event_revisions is None
    assert loaded.response_event_id == "$response"


def test_recovery_does_not_adopt_revision_without_its_prompt(tmp_path: Path) -> None:
    """A ledger edit revision is unusable unless its matching durable prompt survived."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(
            ["$event"],
            response_event_id="$old-response",
            source_event_revisions={"$event": (20, "$new-edit")},
            timestamp=10,
        ),
    )
    recovery_record = TurnRecord.create(
        ["$event"],
        response_event_id="$new-response",
        source_event_prompts={"$event": "old prompt"},
        source_event_revisions={"$event": (10, "$old-edit")},
        timestamp=20,
    )

    loaded = _load_with_recovery(store, original_event_id="$event", recovery_record=recovery_record)

    assert loaded is not None
    assert loaded.source_event_prompts == {"$event": "old prompt"}
    assert loaded.source_event_revisions == {"$event": (10, "$old-edit")}


def test_terminal_write_refreshes_ledger_precedence_timestamp(tmp_path: Path) -> None:
    """A successful terminal write should become newer than its recovered input."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(["$event"], response_event_id="$old-response", timestamp=1),
    )

    store.record_turn(TurnRecord.create(["$event"], response_event_id="$new-response", timestamp=1))

    updated = store.get_turn_record("$event")
    assert updated is not None
    assert updated.response_event_id == "$new-response"
    assert updated.timestamp > 1


def test_terminal_turn_can_replace_a_provisional_source_identity(tmp_path: Path) -> None:
    """A partial visible echo may join the canonical coalesced turn that completes it."""
    store = _store(tmp_path)
    store.record_visible_echo("$second", "$echo")

    store.record_turn(TurnRecord.create(["$first", "$second"], response_event_id="$response"))

    first_record = store.get_turn_record("$first")
    second_record = store.get_turn_record("$second")
    assert first_record is not None
    assert first_record == second_record
    assert first_record.source_event_ids == ("$first", "$second")
    assert first_record.visible_echo_event_id == "$echo"


def test_visible_echo_is_finalized_only_after_replacement_acknowledgement(tmp_path: Path) -> None:
    """A posted placeholder should not become a terminal router outcome before its edit succeeds."""
    store = _store(tmp_path)
    store.record_visible_echo("$event", "$echo")

    assert store.finalized_visible_echo_for_sources(("$event",)) is None

    store.record_finalized_visible_echo("$event", "$echo", is_fallback=False)

    record = store.get_turn_record("$event")
    assert record is not None
    assert not record.completed
    assert record.response_event_id == "$echo"
    assert store.finalized_visible_echo_for_sources(("$event",)) == "$echo"
    finalized = store.finalized_visible_echo("$event")
    assert finalized is not None
    assert finalized.event_id == "$echo"
    assert finalized.is_fallback is False


def test_visible_echo_finalization_requires_tracked_placeholder(tmp_path: Path) -> None:
    """An acknowledgement alone must not create a finalized visible echo."""
    store = _store(tmp_path)

    store.record_finalized_visible_echo("$event", "$echo", is_fallback=False)

    assert store.get_turn_record("$event") is None


def test_visible_echo_finalization_cannot_overwrite_terminal_outcome(tmp_path: Path) -> None:
    """A late edit acknowledgement should preserve a concurrently completed turn."""
    store = _store(tmp_path)
    store.record_visible_echo("$event", "$echo")
    store.record_turn(TurnRecord.create(["$event"], response_event_id="$response"))

    store.record_finalized_visible_echo("$event", "$echo", is_fallback=False)

    record = store.get_turn_record("$event")
    assert record is not None
    assert record.completed
    assert record.response_event_id == "$response"
    finalized = store.finalized_visible_echo("$event")
    assert finalized is not None
    assert finalized.event_id == "$echo"
    assert finalized.is_fallback is False


def test_finalized_visible_echo_keeps_transcript_over_fallback(tmp_path: Path) -> None:
    """Transcript replacement should upgrade fallback and reject later downgrade."""
    store = _store(tmp_path)
    store.record_visible_echo("$event", "$echo")

    store.record_finalized_visible_echo("$event", "$echo", is_fallback=True)
    fallback = store.finalized_visible_echo("$event")
    assert fallback is not None
    assert fallback.is_fallback is True

    store.record_finalized_visible_echo("$event", "$echo", is_fallback=False)
    transcript = store.finalized_visible_echo("$event")
    assert transcript is not None
    assert transcript.is_fallback is False

    store.record_finalized_visible_echo("$event", "$echo", is_fallback=True)
    assert store.finalized_visible_echo("$event") == transcript

    store._ledger.flush()
    _reset_handled_turn_ledger_runtime()
    assert _store(tmp_path).finalized_visible_echo("$event") == transcript


def test_terminal_turn_rejects_conflicting_completed_canonical_source(tmp_path: Path) -> None:
    """A completed source cannot be reassigned into a different canonical turn."""
    store = _store(tmp_path)
    store.record_turn(TurnRecord.create(["$first"], response_event_id="$first-response"))

    store.record_turn(TurnRecord.create(["$first", "$second"], response_event_id="$other-response"))

    first_record = store.get_turn_record("$first")
    assert first_record is not None
    assert first_record.source_event_ids == ("$first",)
    assert first_record.response_event_id == "$first-response"
    assert store.get_turn_record("$second") is None


def test_run_metadata_without_current_schema_version_is_not_recovery_data() -> None:
    """Stale pre-user run metadata should not create an implicit migration path."""
    assert (
        TurnRecordCodec.from_run_metadata(
            {
                constants.MATRIX_EVENT_ID_METADATA_KEY: "$event",
                constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY: ["$event"],
            },
        )
        is None
    )


def test_run_metadata_with_empty_normalized_sources_falls_back_to_anchor() -> None:
    """Current metadata should never decode into an eventless canonical record."""
    parsed = TurnRecordCodec.from_run_metadata(
        {
            constants.MATRIX_TURN_SCHEMA_VERSION_METADATA_KEY: TurnRecordCodec.schema_version(),
            constants.MATRIX_EVENT_ID_METADATA_KEY: "$anchor",
            constants.MATRIX_SOURCE_EVENT_IDS_METADATA_KEY: ["", None, 42],
        },
    )

    assert parsed is not None
    assert parsed.anchor_event_id == "$anchor"
    assert parsed.source_event_ids == ("$anchor",)


def test_undelivered_run_repairs_as_incomplete_and_remains_retryable(tmp_path: Path) -> None:
    """A persisted run without Matrix response linkage must not become a handled turn."""
    store = _store(tmp_path)
    metadata = TurnRecordCodec.to_run_metadata(
        TurnRecord.create(["$event"], response_owner="agent"),
    )
    metadata[constants.MATRIX_EVENT_ID_METADATA_KEY] = "$event"
    recovery_record = TurnRecordCodec.from_run_metadata(metadata)

    assert recovery_record is not None
    loaded = _load_with_recovery(
        store,
        original_event_id="$event",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert not loaded.completed
    repaired = store.get_turn_record("$event")
    assert repaired is not None
    assert not repaired.completed
    assert repaired.response_owner == "agent"
    assert not store.is_handled("$event")


def test_load_turn_uses_ledger_identity_and_outcome_then_backfills_missing_context(tmp_path: Path) -> None:
    """Ledger facts should win field-by-field while absent optional context comes from run metadata."""
    store = _store(tmp_path)
    ledger_record = TurnRecord.create(
        ["$first", "$anchor"],
        response_event_id="$ledger-response",
        source_event_prompts={"$first": "ledger first", "$anchor": "ledger anchor"},
        requester_id="@ledger-user:example.org",
    )
    store.record_turn(ledger_record)
    persisted_ledger_record = store.get_turn_record("$first")
    assert persisted_ledger_record is not None
    recovery_target = MessageTarget.resolve("!room:example.org", None, "$anchor")
    recovery_record = TurnRecord.create(
        ["$run-only", "$anchor"],
        response_event_id="$run-response",
        source_event_prompts={"$run-only": "run", "$anchor": "run anchor"},
        response_owner="agent",
        requester_id="@run-user:example.org",
        history_scope=HistoryScope(kind="agent", scope_id="agent"),
        conversation_target=recovery_target,
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$first",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.source_event_ids == ("$first", "$anchor")
    assert loaded.anchor_event_id == "$anchor"
    assert loaded.response_event_id == "$ledger-response"
    assert loaded.source_event_prompts == {"$first": "ledger first", "$anchor": "ledger anchor"}
    assert loaded.requester_id == "@ledger-user:example.org"
    assert loaded.response_owner == "agent"
    assert loaded.history_scope == HistoryScope(kind="agent", scope_id="agent")
    assert loaded.conversation_target == recovery_target
    assert loaded.timestamp > persisted_ledger_record.timestamp
    repaired = store.get_turn_record("$first")
    assert repaired == loaded


def test_load_turn_repairs_missing_ledger_row_from_run_metadata(tmp_path: Path) -> None:
    """Run metadata should recover and immediately backfill an absent ledger row."""
    store = _store(tmp_path)
    recovery_record = TurnRecord.create(
        ["$event"],
        response_event_id="$response",
        response_owner="agent",
    )

    loaded = _load_with_recovery(
        store,
        original_event_id="$event",
        recovery_record=recovery_record,
    )

    assert loaded is not None
    assert loaded.timestamp > 0
    assert replace(loaded, timestamp=0.0) == recovery_record
    repaired = store.get_turn_record("$event")
    assert repaired is not None
    assert repaired.response_event_id == "$response"
    assert repaired.response_owner == "agent"


def test_record_turn_preserves_existing_optional_facts_at_the_owner_boundary(tmp_path: Path) -> None:
    """TurnStore, rather than the physical ledger, should merge repeated writes."""
    store = _store(tmp_path)
    store.record_turn(
        TurnRecord.create(
            ["$event"],
            response_event_id="$first-response",
            requester_id="@user:example.org",
            correlation_id="corr-1",
        ),
    )

    store.record_turn(TurnRecord(source_event_ids=("$event",), response_event_id="$second-response"))

    record = store.get_turn_record("$event")
    assert record is not None
    assert record.response_event_id == "$second-response"
    assert record.requester_id == "@user:example.org"
    assert record.correlation_id == "corr-1"


def test_visible_echo_cannot_overwrite_concurrent_terminal_outcome(tmp_path: Path) -> None:
    """A delayed visible-echo update must preserve a terminal write racing behind it."""
    store = _store(tmp_path)
    terminal_record = TurnRecord.create(["$event"], response_event_id="$response")
    echo_record_built = threading.Event()
    release_echo_record = threading.Event()
    terminal_started = threading.Event()
    terminal_finished = threading.Event()
    create_turn_record = TurnRecord.create

    def blocking_create(
        source_event_ids: list[str],
        *,
        completed: bool = True,
        **kwargs: object,
    ) -> TurnRecord:
        turn_record = create_turn_record(source_event_ids, completed=completed, **kwargs)
        if not completed:
            echo_record_built.set()
            assert release_echo_record.wait(timeout=2)
        return turn_record

    def record_visible_echo() -> None:
        store.record_visible_echo("$event", "$echo")

    def record_terminal_outcome() -> None:
        terminal_started.set()
        store.record_turn(terminal_record)
        terminal_finished.set()

    with patch.object(TurnRecord, "create", side_effect=blocking_create):
        echo_thread = threading.Thread(target=record_visible_echo)
        echo_thread.start()
        assert echo_record_built.wait(timeout=2)

        terminal_thread = threading.Thread(target=record_terminal_outcome)
        terminal_thread.start()
        assert terminal_started.wait(timeout=2)
        assert not terminal_finished.wait(timeout=0.1)

        release_echo_record.set()
        echo_thread.join(timeout=2)
        terminal_thread.join(timeout=2)

    assert not echo_thread.is_alive()
    assert not terminal_thread.is_alive()
    record = store.get_turn_record("$event")
    assert record is not None
    assert record.completed
    assert record.response_event_id == "$response"
    assert record.visible_echo_event_id == "$echo"


def test_record_responded_turn_rejects_empty_response_event_id(tmp_path: Path) -> None:
    """The durable response boundary should reject a noncanonical empty event ID."""
    store = _store(tmp_path)
    noncanonical_record = TurnRecord(source_event_ids=("$source",), response_event_id="")

    with pytest.raises(RuntimeError, match="requires a visible Matrix response event ID"):
        store.record_responded_turn(noncanonical_record)


@pytest.mark.parametrize("recovery_response_event_id", [None, "$stale-response"])
def test_recovery_cannot_overwrite_concurrent_terminal_outcome(
    tmp_path: Path,
    recovery_response_event_id: str | None,
) -> None:
    """Slow incomplete or delivered recovery must preserve a concurrent terminal write."""
    store = _store(tmp_path)
    store._ledger.record_handled_turn(
        TurnRecord.create(["$event"], response_event_id="$old-response", timestamp=9),
    )
    recovery_started = threading.Event()
    release_recovery = threading.Event()
    load_finished = threading.Event()
    loaded_record: list[TurnRecord | None] = []
    recovery_record = TurnRecord.create(
        ["$event"],
        response_event_id=recovery_response_event_id,
        completed=recovery_response_event_id is not None,
        response_owner="agent",
        timestamp=10,
    )

    def load_recovery(_request: object) -> TurnRecord:
        recovery_started.set()
        assert release_recovery.wait(timeout=2)
        return recovery_record

    def load_turn() -> None:
        loaded_record.append(
            store.load_turn(
                room=MagicMock(room_id="!room:example.org"),
                thread_id=None,
                original_event_id="$event",
                requester_user_id="@user:example.org",
            ),
        )
        load_finished.set()

    with patch.object(store, "_load_persisted_turn_record", side_effect=load_recovery):
        load_thread = threading.Thread(target=load_turn)
        load_thread.start()
        assert recovery_started.wait(timeout=2)

        with patch("mindroom.handled_turns.time.time", return_value=10.9):
            store.record_turn(TurnRecord.create(["$event"], response_event_id="$response"))
        release_recovery.set()
        assert load_finished.wait(timeout=2)
        load_thread.join(timeout=2)

    assert not load_thread.is_alive()
    assert len(loaded_record) == 1
    assert loaded_record[0] is not None
    assert loaded_record[0].completed
    assert loaded_record[0].response_event_id == "$response"
    assert loaded_record[0].response_owner == "agent"
    assert loaded_record[0].timestamp > 10.9
    record = store.get_turn_record("$event")
    assert record == loaded_record[0]


def test_only_turn_store_imports_handled_turn_ledger_in_production() -> None:
    """HandledTurnLedger imports should stay isolated to TurnStore in production code."""
    src_root = Path(__file__).resolve().parents[1] / "src" / "mindroom"
    offenders: list[str] = []

    for path in src_root.rglob("*.py"):
        if path.name in {"turn_store.py", "handled_turns.py"}:
            continue
        module = ast.parse(path.read_text(), filename=str(path))
        for node in ast.walk(module):
            if not isinstance(node, ast.ImportFrom):
                continue
            if node.module != "mindroom.handled_turns":
                continue
            if any(alias.name == "HandledTurnLedger" for alias in node.names):
                offenders.append(path.relative_to(src_root).as_posix())
                break

    assert offenders == []


def test_agent_bot_does_not_expose_removed_handled_turn_ledger_shim(tmp_path: Path) -> None:
    """AgentBot instances should route handled-turn state only through TurnStore."""
    config = bind_runtime_paths(Config(), test_runtime_paths(tmp_path))
    bot = AgentBot(
        agent_user=AgentMatrixUser(
            agent_name="agent",
            user_id="@mindroom_agent:localhost",
            display_name="Agent",
            password=TEST_PASSWORD,
        ),
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
    )

    removed_attr = "_handled" + "_turn_ledger"
    assert removed_attr not in AgentBot.__dict__
    assert not hasattr(bot, removed_attr)
    assert removed_attr not in vars(bot)


def test_router_turn_replay_uses_persisted_ledger_across_two_restarts(tmp_path: Path) -> None:
    """Router relay turns have durable ledger state but no Agno run storage."""

    def router_store() -> TurnStore:
        config = bind_runtime_paths(Config(), test_runtime_paths(tmp_path))
        return TurnStore(
            TurnStoreDeps(
                agent_name="router",
                tracking_base_path=tmp_path / "tracking",
                state_writer=ConversationStateWriter(
                    ConversationStateWriterDeps(
                        runtime=SimpleNamespace(config=config),
                        logger=MagicMock(),
                        runtime_paths=runtime_paths_for(config),
                        agent_name="router",
                    ),
                ),
                resolver=MagicMock(),
                tool_runtime=MagicMock(),
            ),
        )

    target = MessageTarget.resolve("!room:localhost", "$thread", "$source")
    expected = TurnRecord.create(
        ["$source"],
        response_event_id="$relay",
        response_owner="router",
        requester_id="@user:localhost",
        conversation_target=target,
    )
    router_store().record_turn(expected)

    for _restart in range(2):
        _reset_handled_turn_ledger_runtime()
        restarted = router_store()
        with patch.object(
            restarted,
            "_load_persisted_turn_record",
            side_effect=AssertionError("ledger-only entity attempted run recovery"),
        ):
            loaded = restarted.load_turn(
                room=MagicMock(room_id="!room:localhost"),
                thread_id="$thread",
                original_event_id="$source",
                requester_user_id="@user:localhost",
            )

        assert loaded is not None
        assert replace(loaded, timestamp=0.0) == replace(expected, timestamp=0.0)
        assert loaded.timestamp > 0


def test_no_test_references_removed_bot_handled_turn_ledger_shim() -> None:
    """Tests should route all handled-turn access through TurnStore."""
    tests_root = Path(__file__).resolve().parent
    needle = "._handled" + "_turn_ledger"
    offenders = [
        path.relative_to(tests_root).as_posix() for path in tests_root.rglob("*.py") if needle in path.read_text()
    ]

    assert offenders == []
