turnstone/tests/test_persistence_reconciliation.py

"""Adversarial coverage for accepted-row persistence reconciliation."""

from __future__ import annotations

import threading
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, patch

import pytest

from tests._session_helpers import RecordingUI, make_result, make_session
from tests.test_session_manager import _make_manager
from turnstone.core.session import ConversationPersistenceError, EffectStatus
from turnstone.core.storage import ConversationCommitConflictError
from turnstone.core.trajectory import Role
from turnstone.core.workstream import WorkstreamState

if TYPE_CHECKING:
    from collections.abc import Callable


def _journal(
    session: Any,
    persist: Callable[[], int],
    *,
    commit_key: str = "commit-a",
) -> Any:
    with session._history_handoff_lock:
        return session._journal_conversation_row_locked(
            commit_key=commit_key,
            message={"role": "assistant", "content": "accepted"},
            persist=persist,
            event_id=7,
        )


def test_transient_reconciliation_is_single_attempt_due_gated_and_capped() -> None:
    session = make_session()
    clock = [100.0]
    outcomes = iter([0, 0, 41])
    persist = MagicMock(side_effect=lambda: next(outcomes))
    entry = _journal(session, persist)
    assert session.conversation_persistence_status()["state"] == "pending"

    with (
        patch("turnstone.core.session.time.monotonic", side_effect=lambda: clock[0]),
        patch.object(
            session,
            "_conversation_persistence_retry_delay",
            side_effect=[2.0, 4.0],
        ),
    ):
        with pytest.raises(ConversationPersistenceError):
            session._persist_pending_conversation_commit(entry)

        assert persist.call_count == 1
        assert session.conversation_persistence_status()["state"] == "retrying"
        assert session._conversation_persistence_next_retry_at == 102.0

        clock[0] = 101.999
        assert session.reconcile_unresolved_persistence_if_due(now=clock[0]) is False
        with pytest.raises(ConversationPersistenceError):
            session._reconcile_pending_conversation_commits()
        assert persist.call_count == 1

        clock[0] = 102.0
        assert session.reconcile_unresolved_persistence_if_due(now=clock[0]) is True
        assert persist.call_count == 2
        assert session._conversation_persistence_next_retry_at == 106.0

        clock[0] = 106.0
        assert session.reconcile_unresolved_persistence_if_due(now=clock[0]) is True

    assert persist.call_count == 3
    assert session.has_unresolved_conversation_persistence() is False
    assert session.conversation_persistence_status() == {
        "state": "healthy",
        "pending_rows": 0,
        "attempts": 0,
        "first_failure_at": None,
        "last_failure_at": None,
        "next_retry_at": None,
    }
    with patch("turnstone.core.session.random.uniform", return_value=60.0) as jitter:
        assert session._conversation_persistence_retry_delay(10_000) == 60.0
    assert jitter.call_args.args == (30.0, 60.0)


def test_repeated_failure_dedupes_repair_fanout_until_recovery() -> None:
    session = make_session()
    session.ui.on_history_resync = MagicMock()
    session.ui.on_persistence_state_changed = MagicMock()
    persist = MagicMock(side_effect=[0, 0, 9])
    entry = _journal(session, persist)

    with patch.object(session, "_conversation_persistence_retry_delay", return_value=0.0):
        with pytest.raises(ConversationPersistenceError):
            session._persist_pending_conversation_commit(entry)
        assert session.reconcile_unresolved_persistence_if_due(now=float("inf")) is True
        assert session.reconcile_unresolved_persistence_if_due(now=float("inf")) is True

    session.ui.on_history_resync.assert_called_once_with("conversation_persistence_unresolved")
    assert session.ui.on_persistence_state_changed.call_count == 2
    assert session.conversation_persistence_status()["state"] == "healthy"


def test_concurrent_due_sweeps_cannot_duplicate_one_retry_attempt() -> None:
    session = make_session()
    retry_entered = threading.Event()
    release_retry = threading.Event()
    calls = 0

    def _persist() -> int:
        nonlocal calls
        calls += 1
        if calls == 1:
            return 0
        retry_entered.set()
        assert release_retry.wait(5)
        return 0

    entry = _journal(session, _persist)
    delays = iter([0.0, 10.0])
    with (
        patch("turnstone.core.session.time.monotonic", return_value=100.0),
        patch.object(
            session,
            "_conversation_persistence_retry_delay",
            side_effect=lambda _attempt: next(delays),
        ),
    ):
        with pytest.raises(ConversationPersistenceError):
            session._persist_pending_conversation_commit(entry)

        results: list[bool] = []
        sweeps = [
            threading.Thread(
                target=lambda: results.append(
                    session.reconcile_unresolved_persistence_if_due(now=100.0)
                ),
                daemon=True,
            )
            for _ in range(2)
        ]
        sweeps[0].start()
        assert retry_entered.wait(5)
        sweeps[1].start()
        release_retry.set()
        for sweep in sweeps:
            sweep.join(5)

    assert all(not sweep.is_alive() for sweep in sweeps)
    assert calls == 2
    assert sorted(results) == [False, True]
    assert session._conversation_persistence_next_retry_at == 110.0


def test_commit_conflict_is_chained_permanent_and_not_history_acked() -> None:
    session = make_session()
    conflict = ConversationCommitConflictError("immutable mismatch")
    persist = MagicMock(side_effect=conflict)
    entry = _journal(session, persist)

    with pytest.raises(ConversationPersistenceError) as raised:
        session._persist_pending_conversation_commit(entry)

    assert raised.value.__cause__ is conflict
    assert "immutable durable commit conflict" in str(raised.value)
    assert persist.call_count == 1
    status = session.conversation_persistence_status()
    assert status["state"] == "conflict"
    assert status["next_retry_at"] is None
    assert set(status) == {
        "state",
        "pending_rows",
        "attempts",
        "first_failure_at",
        "last_failure_at",
        "next_retry_at",
    }

    assert session.reconcile_unresolved_persistence_if_due(now=float("inf")) is False
    with pytest.raises(ConversationPersistenceError) as retried:
        session._reconcile_pending_conversation_commits(_force_retry=True)
    assert retried.value is raised.value
    assert persist.call_count == 1

    rows, _token = session.capture_history_handoff(
        lambda _overscan: [
            {
                "role": "assistant",
                "content": "different durable value",
                "_commit_key": entry.commit_key,
            }
        ]
    )
    assert session.has_unresolved_conversation_persistence() is True
    assert session.conversation_persistence_status()["state"] == "conflict"
    assert rows[0]["content"] == "accepted"


def test_history_lost_ack_reconciliation_clears_all_retry_metadata() -> None:
    session = make_session()
    session.ui.on_persistence_state_changed = MagicMock()
    entry = _journal(session, MagicMock(return_value=0))
    with pytest.raises(ConversationPersistenceError):
        session._persist_pending_conversation_commit(entry)

    assert session.conversation_persistence_status()["state"] == "retrying"
    session.capture_history_handoff(
        lambda _overscan: [
            {
                "role": "assistant",
                "content": "accepted",
                "_commit_key": entry.commit_key,
            }
        ]
    )

    assert session.conversation_persistence_status() == {
        "state": "healthy",
        "pending_rows": 0,
        "attempts": 0,
        "first_failure_at": None,
        "last_failure_at": None,
        "next_retry_at": None,
    }
    assert session.ui.on_persistence_state_changed.call_count == 2


def test_already_acked_race_has_no_fabricated_row_id() -> None:
    session = make_session()
    persist_entered = threading.Event()
    release_persist = threading.Event()

    def _persist() -> int:
        persist_entered.set()
        assert release_persist.wait(5)
        return 42

    persist = MagicMock(side_effect=_persist)
    entry = _journal(session, persist)
    results: list[object] = []

    first = threading.Thread(
        target=lambda: results.append(session._persist_pending_conversation_commit(entry)),
        daemon=True,
    )
    second = threading.Thread(
        target=lambda: results.append(session._persist_pending_conversation_commit(entry)),
        daemon=True,
    )
    first.start()
    assert persist_entered.wait(5)
    second.start()
    release_persist.set()
    first.join(5)
    second.join(5)

    assert not first.is_alive() and not second.is_alive()
    assert persist.call_count == 1
    assert results == [None, None]


def test_terminal_barrier_blocks_a_late_due_sweep() -> None:
    session = make_session()
    persist = MagicMock(return_value=0)
    entry = _journal(session, persist)
    with pytest.raises(ConversationPersistenceError):
        session._persist_pending_conversation_commit(entry)
    assert persist.call_count == 1

    session.shutdown_publication_and_drain_durability()
    assert session.reconcile_unresolved_persistence_if_due(now=float("inf")) is False
    assert persist.call_count == 1


def test_terminal_race_is_a_benign_cancelled_maintenance_pass() -> None:
    session = make_session()
    entry = _journal(session, MagicMock(return_value=0))
    with pytest.raises(ConversationPersistenceError):
        session._persist_pending_conversation_commit(entry)

    from turnstone.core.session import GenerationCancelled

    with patch.object(
        session,
        "_reconcile_pending_conversation_commits",
        side_effect=GenerationCancelled(),
    ):
        assert session.reconcile_unresolved_persistence_if_due(now=float("inf")) is False


def test_soft_close_forces_transient_retry_but_never_retries_conflict() -> None:
    transient = make_session()
    transient_persist = MagicMock(side_effect=[0, 17])
    transient_entry = _journal(transient, transient_persist)
    with (
        patch.object(
            transient,
            "_conversation_persistence_retry_delay",
            return_value=60.0,
        ),
        pytest.raises(ConversationPersistenceError),
    ):
        transient._persist_pending_conversation_commit(transient_entry)

    assert transient.prepare_soft_close() is True
    assert transient_persist.call_count == 2
    assert transient.has_unresolved_conversation_persistence() is False

    conflicted = make_session()
    conflict = ConversationCommitConflictError("immutable mismatch")
    conflict_persist = MagicMock(side_effect=conflict)
    conflict_entry = _journal(conflicted, conflict_persist)
    with pytest.raises(ConversationPersistenceError):
        conflicted._persist_pending_conversation_commit(conflict_entry)

    assert conflicted.prepare_soft_close() is False
    assert conflicted._publication_shutdown is False
    assert conflict_persist.call_count == 1


def test_assistant_persistence_recovery_completes_unstarted_tool_prefix(
    tmp_db: Any,
) -> None:
    """Recovery cannot expose an assistant tool call without its TOOL row."""
    from turnstone.core.memory import register_workstream

    manager, _adapter, _storage = _make_manager(max_active=1)
    ws = manager.create(user_id="owner", ws_id="tool-prefix-recovery")
    session = make_session(ws_id=ws.id, user_id="owner", ui=RecordingUI())
    session._title_generated = True
    ws.session = session
    register_workstream(ws.id, user_id="owner")
    tool_calls = [
        {
            "id": call_id,
            "type": "function",
            "function": {"name": tool_name, "arguments": "{}"},
        }
        for call_id, tool_name in (
            ("call-never-started-a", "write_file"),
            ("call-never-started-b", "notify"),
        )
    ]
    assistant_attempts = 0
    saved_roles: list[tuple[str, str | None]] = []

    def _save_message(
        _ws_id: str,
        role: str,
        _content: str,
        *_args: Any,
        **kwargs: Any,
    ) -> int:
        nonlocal assistant_attempts
        saved_roles.append((role, kwargs.get("tool_call_id")))
        if role == "assistant" and kwargs.get("tool_calls"):
            assistant_attempts += 1
            if assistant_attempts == 1:
                return 0
        return len(saved_roles) + 10

    stream_calls = 0

    def _stream(_generation: int) -> Any:
        nonlocal stream_calls
        stream_calls += 1
        if stream_calls == 1:
            return make_result(tool_calls=tool_calls)
        # The next provider call sees a complete assistant/tool prefix before
        # the newly admitted user row. This is the structural provider seam
        # that the old recovery path violated.
        roles = [turn.role for turn in session.messages]
        assert roles == [
            Role.USER,
            Role.ASSISTANT,
            Role.TOOL,
            Role.TOOL,
            Role.USER,
        ]
        assistant = session.messages[1]
        assert [call.id for call in assistant.tool_calls] == [
            "call-never-started-a",
            "call-never-started-b",
        ]
        assert [turn.tool_call_id for turn in session.messages[2:4]] == [
            "call-never-started-a",
            "call-never-started-b",
        ]
        return make_result("continued safely")

    execute = MagicMock()
    with (
        patch("turnstone.core.session.save_message", side_effect=_save_message) as save,
        patch.object(session, "_stream_response", side_effect=_stream),
        patch.object(session, "_execute_tools", execute),
        patch.object(session, "_visible_memory_count", return_value=0),
        patch.object(session, "_print_status_line"),
    ):
        with pytest.raises(ConversationPersistenceError):
            session.send("use the tool", acting_user_id="owner")

        execute.assert_not_called()
        assert saved_roles == [("user", None), ("assistant", None)]
        assert session.conversation_persistence_status()["state"] == "retrying"
        assert session.conversation_persistence_status()["pending_rows"] == 3
        assert [turn.role for turn in session.messages] == [
            Role.USER,
            Role.ASSISTANT,
            Role.TOOL,
            Role.TOOL,
        ]
        cancelled = session.messages[-2:]
        assert [turn.tool_call_id for turn in cancelled] == [
            "call-never-started-a",
            "call-never-started-b",
        ]
        assert all(turn.effect_status is EffectStatus.NONE for turn in cancelled)
        assert all("no side effects" in turn.text.lower() for turn in cancelled)

        manager.set_state(ws.id, WorkstreamState.ERROR)
        retry_at = session._conversation_persistence_next_retry_at
        assert retry_at is not None
        assert manager.reconcile_unresolved_persistence(now=retry_at - 0.001) == []
        assert save.call_count == 2

        assert manager.reconcile_unresolved_persistence(now=float("inf")) == [ws.id]
        assert saved_roles == [
            ("user", None),
            ("assistant", None),
            ("assistant", None),
            ("tool", "call-never-started-a"),
            ("tool", "call-never-started-b"),
        ]
        assert session.conversation_persistence_status()["state"] == "healthy"
        assert ws.state is WorkstreamState.IDLE

        session.send("continue", acting_user_id="owner")

    execute.assert_not_called()
    assert stream_calls == 2
    assert session.has_unresolved_conversation_persistence() is False


def test_assistant_commit_conflict_completes_live_tool_prefix_but_never_retries(
    tmp_db: Any,
) -> None:
    """A permanent assistant conflict is structurally complete but fail-stop."""
    session = make_session(user_id="owner", ui=RecordingUI())
    session._title_generated = True
    conflict = ConversationCommitConflictError("immutable mismatch")
    conversation_writes: list[str] = []

    def _save_message(
        _ws_id: str,
        role: str,
        _content: str,
        *_args: Any,
        **kwargs: Any,
    ) -> int:
        conversation_writes.append(role)
        if role == "assistant" and kwargs.get("tool_calls"):
            raise conflict
        if role == "tool":
            raise AssertionError("conflicted tool row must never be written")
        return len(conversation_writes)

    tool_call = {
        "id": "conflicted-call",
        "type": "function",
        "function": {"name": "write_file", "arguments": "{}"},
    }
    execute = MagicMock()
    stream = MagicMock(return_value=make_result(tool_calls=[tool_call]))
    with (
        patch("turnstone.core.session.save_message", side_effect=_save_message),
        patch.object(session, "_stream_response", stream),
        patch.object(session, "_execute_tools", execute),
        patch.object(session, "_visible_memory_count", return_value=0),
        patch.object(session, "_print_status_line"),
    ):
        with pytest.raises(ConversationPersistenceError) as raised:
            session.send("use the tool", acting_user_id="owner")

        assert raised.value.__cause__ is conflict
        execute.assert_not_called()
        assert conversation_writes == ["user", "assistant"]
        assert [turn.role for turn in session.messages] == [
            Role.USER,
            Role.ASSISTANT,
            Role.TOOL,
        ]
        assert session.messages[-1].effect_status is EffectStatus.NONE
        assert session.conversation_persistence_status()["state"] == "conflict"
        assert session.conversation_persistence_status()["pending_rows"] == 2

        assert session.reconcile_unresolved_persistence_if_due(now=float("inf")) is False
        with pytest.raises(ConversationPersistenceError):
            session.send("must remain blocked", acting_user_id="owner")

    assert conversation_writes == ["user", "assistant"]
    execute.assert_not_called()
    stream.assert_called_once()


def test_terminal_barrier_waits_for_admitted_sweep_then_prevents_another() -> None:
    session = make_session()
    retry_entered = threading.Event()
    release_retry = threading.Event()
    calls = 0

    def _persist() -> int:
        nonlocal calls
        calls += 1
        if calls == 1:
            return 0
        retry_entered.set()
        assert release_retry.wait(5)
        return 23

    entry = _journal(session, _persist)
    with pytest.raises(ConversationPersistenceError):
        session._persist_pending_conversation_commit(entry)

    repair = threading.Thread(
        target=lambda: session.reconcile_unresolved_persistence_if_due(now=float("inf")),
        daemon=True,
    )
    repair.start()
    assert retry_entered.wait(5)

    terminal = threading.Thread(
        target=session.shutdown_publication_and_drain_durability,
        daemon=True,
    )
    terminal.start()
    terminal.join(0.05)
    assert terminal.is_alive(), "terminal delete barrier passed an in-flight repair"

    release_retry.set()
    repair.join(5)
    terminal.join(5)
    assert not repair.is_alive() and not terminal.is_alive()
    assert calls == 2
    assert session.reconcile_unresolved_persistence_if_due(now=float("inf")) is False
    assert calls == 2


class _RecoverableSession:
    def __init__(self, manager: Any) -> None:
        self._manager = manager
        self.unresolved = True
        self.reconcile_calls = 0
        self.manager_lock_was_free = False
        self.cancelled = False
        self.closed = False

    def has_unresolved_conversation_persistence(self) -> bool:
        return self.unresolved

    def reconcile_unresolved_persistence_if_due(self, *, now: float) -> bool:
        del now
        self.reconcile_calls += 1
        acquired = threading.Event()

        def _probe_manager_lock() -> None:
            with self._manager._lock:
                acquired.set()

        probe = threading.Thread(target=_probe_manager_lock, daemon=True)
        probe.start()
        self.manager_lock_was_free = acquired.wait(1.0)
        probe.join(1.0)
        self.unresolved = False
        return True

    def cancel(self) -> None:
        self.cancelled = True

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


def test_manager_sweep_runs_outside_lock_and_capacity_retries_once() -> None:
    manager, _adapter, _storage = _make_manager(max_active=1)
    incumbent = manager.create(user_id="owner", ws_id="incumbent")
    recoverable = _RecoverableSession(manager)
    incumbent.session = recoverable

    replacement = manager.create(user_id="owner", ws_id="replacement")

    assert replacement.id == "replacement"
    assert recoverable.reconcile_calls == 1
    assert recoverable.manager_lock_was_free is True
    assert recoverable.cancelled is True
    assert recoverable.closed is True


def test_manager_repairs_exact_persistence_error_to_idle_and_frees_capacity(
    tmp_db: Any,
) -> None:
    from turnstone.core.memory import register_workstream

    manager, _adapter, _storage = _make_manager(max_active=1)
    ws = manager.create(user_id="owner", ws_id="real-persistence-error")
    session = make_session(ws_id=ws.id)
    session.ui.on_state_change = MagicMock()
    ws.session = session
    register_workstream(ws.id, user_id="owner")
    persist = MagicMock(side_effect=[0, 31])
    entry = _journal(session, persist)

    with pytest.raises(ConversationPersistenceError) as raised:
        session._persist_pending_conversation_commit(entry)
    session._record_fatal_error(raised.value)
    manager.set_state(ws.id, WorkstreamState.ERROR)
    assert session.conversation_persistence_fatal_revision() is not None

    # A still-running recovery send owns the state and suppresses maintenance.
    with ws._lock:
        ws._worker_running = True
    assert manager.reconcile_unresolved_persistence(now=float("inf")) == []
    assert persist.call_count == 1
    assert ws.state is WorkstreamState.ERROR
    with ws._lock:
        ws._worker_running = False

    assert manager.reconcile_unresolved_persistence(now=float("inf")) == [ws.id]
    assert persist.call_count == 2
    assert ws.state is WorkstreamState.IDLE
    assert session.conversation_persistence_fatal_revision() is None

    replacement = manager.create(user_id="owner", ws_id="replacement-after-repair")
    assert replacement.id == "replacement-after-repair"


def test_manager_does_not_clear_an_error_with_newer_fatal_provenance(tmp_db: Any) -> None:
    from turnstone.core.memory import register_workstream

    manager, _adapter, _storage = _make_manager(max_active=1)
    ws = manager.create(user_id="owner", ws_id="newer-fatal")
    session = make_session(ws_id=ws.id)
    session.ui.on_state_change = MagicMock()
    ws.session = session
    register_workstream(ws.id, user_id="owner")
    persist = MagicMock(side_effect=[0, 37])
    entry = _journal(session, persist)

    with pytest.raises(ConversationPersistenceError) as raised:
        session._persist_pending_conversation_commit(entry)
    session._record_fatal_error(raised.value)
    session._record_fatal_error(RuntimeError("newer independent failure"))
    manager.set_state(ws.id, WorkstreamState.ERROR)

    assert session.conversation_persistence_fatal_revision() is None
    assert manager.reconcile_unresolved_persistence(now=float("inf")) == [ws.id]
    assert session.has_unresolved_conversation_persistence() is False
    assert ws.state is WorkstreamState.ERROR


def test_manager_retires_persistence_error_after_history_proves_lost_ack(
    tmp_db: Any,
) -> None:
    from turnstone.core.memory import register_workstream

    manager, _adapter, _storage = _make_manager(max_active=1)
    ws = manager.create(user_id="owner", ws_id="history-repaired-error")
    session = make_session(ws_id=ws.id)
    session.ui.on_state_change = MagicMock()
    ws.session = session
    register_workstream(ws.id, user_id="owner")
    entry = _journal(session, MagicMock(return_value=0))

    with pytest.raises(ConversationPersistenceError) as raised:
        session._persist_pending_conversation_commit(entry)
    # The repair event can make another browser fetch history before the
    # failing worker reaches its fatal-record tail. Prove that this ordering
    # still retains persistence provenance even though history clears the
    # live journal error first.
    session.capture_history_handoff(
        lambda _overscan: [
            {
                "role": "assistant",
                "content": "accepted",
                "_commit_key": entry.commit_key,
            }
        ]
    )
    assert session.has_unresolved_conversation_persistence() is False

    session._record_fatal_error(raised.value)
    manager.set_state(ws.id, WorkstreamState.ERROR)
    assert ws.state is WorkstreamState.ERROR

    assert manager.reconcile_unresolved_persistence(now=float("inf")) == [ws.id]
    assert ws.state is WorkstreamState.IDLE
    assert session.conversation_persistence_fatal_revision() is None


class _MaintenanceManager:
    def __init__(self, stop: threading.Event) -> None:
        self.stop = stop
        self.reconcile_calls = 0
        self.reap_calls = 0
        self.close_calls = 0

    def reconcile_unresolved_persistence(self) -> list[str]:
        self.reconcile_calls += 1
        if self.reconcile_calls >= 2:
            self.stop.set()
        return []

    def reap_stale_creating_reservations(self, _max_age_seconds: float) -> list[str]:
        self.reap_calls += 1
        return []

    def close_idle(self, _max_age_seconds: float) -> list[str]:
        self.close_calls += 1
        return []

    def subscribe_to_state(self, _callback: Any) -> None:
        raise AssertionError("idle-disabled maintenance must not subscribe")

    def unsubscribe_from_state(self, _callback: Any) -> None:
        raise AssertionError("idle-disabled maintenance must not unsubscribe")


def test_server_maintenance_reconciles_when_idle_eviction_is_disabled() -> None:
    from turnstone import server

    stop = threading.Event()
    manager = _MaintenanceManager(stop)
    with patch.object(server, "PERSISTENCE_RECONCILE_INTERVAL_SECONDS", 0.01):
        server._idle_cleanup_thread(
            manager,  # type: ignore[arg-type]
            0.0,
            MagicMock(),
            stop=stop,
        )

    assert manager.reconcile_calls == 2
    assert manager.close_calls == 0
    assert manager.reap_calls == 1


def test_coordinator_maintenance_reconciles_when_idle_eviction_is_disabled() -> None:
    from turnstone.console.server import _coord_idle_cleanup_thread
    from turnstone.core import session_manager

    stop = threading.Event()
    manager = _MaintenanceManager(stop)
    with patch.object(session_manager, "PERSISTENCE_RECONCILE_INTERVAL_SECONDS", 0.01):
        _coord_idle_cleanup_thread(
            manager,  # type: ignore[arg-type]
            0.0,
            stop,
        )

    assert manager.reconcile_calls == 2
    assert manager.close_calls == 0
    assert manager.reap_calls == 1


def test_capacity_reconcile_forces_a_definite_probe(tmp_db: Any) -> None:
    """``_reserve_and_install``'s last-chance repair runs ONCE, so it must
    not skip a session whose locks are momentarily held.

    The maintenance sweep probes without blocking and re-probes a second
    later; this caller has no second pass — and the sessions likeliest to
    be contended are exactly the ones whose unresolved journals emptied
    its candidate list, so skipping them turns a repairable capacity
    stall into ``All N slots are active``."""
    manager, _adapter, _storage = _make_manager(max_active=1)
    session = make_session()
    ws = manager.create(user_id="u1", ws_id="cap-ws")
    ws.session = session
    ws.state = WorkstreamState.IDLE

    probes: list[str] = []
    session.has_unresolved_conversation_persistence = (  # type: ignore[method-assign]
        lambda: probes.append("blocking") or False
    )
    session.has_unresolved_conversation_persistence_nowait = (  # type: ignore[attr-defined]
        lambda: probes.append("nowait") or None
    )

    # The maintenance shape skips a contended session outright.
    manager.reconcile_unresolved_persistence()
    assert probes == ["nowait"]

    # The one-shot capacity shape forces an answer instead.
    probes.clear()
    manager.reconcile_unresolved_persistence(blocking=True)
    assert probes == ["blocking"]