deer-flow/backend/tests/test_sandbox_orphan_reconciliation.py

"""Tests for sandbox container orphan reconciliation on startup.

Covers:
- SandboxBackend.list_running() default behavior
- LocalContainerBackend.list_running() with mocked docker commands
- _parse_docker_timestamp() / _extract_host_port() helpers
- AioSandboxProvider._reconcile_orphans() decision logic
- SIGHUP signal handler registration
"""

import importlib
import json
import signal
import threading
import time
from datetime import UTC, datetime
from unittest.mock import MagicMock, patch

import pytest

from deerflow.community.aio_sandbox.aio_sandbox_provider import SandboxBeingDestroyedError
from deerflow.community.aio_sandbox.ownership import compute_lease_ttl
from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo

# ── SandboxBackend.list_running() default ────────────────────────────────────


def test_backend_list_running_default_returns_empty():
    """Base SandboxBackend.list_running() returns empty list (backward compat for RemoteSandboxBackend)."""
    from deerflow.community.aio_sandbox.backend import SandboxBackend

    class StubBackend(SandboxBackend):
        def create(self, thread_id, sandbox_id, extra_mounts=None, *, user_id=None):
            del thread_id, sandbox_id, extra_mounts, user_id
            pass

        def destroy(self, info):
            pass

        def is_alive(self, info):
            return False

        def discover(self, sandbox_id):
            return None

    backend = StubBackend()
    assert backend.list_running() == []


# ── Helpers ──────────────────────────────────────────────────────────────────


def _make_local_backend():
    """Create a LocalContainerBackend with minimal config."""
    from deerflow.community.aio_sandbox.local_backend import LocalContainerBackend

    return LocalContainerBackend(
        image="test-image:latest",
        base_port=8080,
        container_prefix="deer-flow-sandbox",
        config_mounts=[],
        environment={},
    )


def _make_inspect_entry(name: str, created: str, host_port: str | None = None) -> dict:
    """Build a minimal docker inspect JSON entry matching the real schema."""
    ports: dict = {}
    if host_port is not None:
        ports["8080/tcp"] = [{"HostIp": "0.0.0.0", "HostPort": host_port}]
    return {
        "Name": f"/{name}",  # docker inspect prefixes names with "/"
        "Created": created,
        "NetworkSettings": {"Ports": ports},
    }


def _mock_ps_and_inspect(monkeypatch, ps_output: str, inspect_payload: list | None):
    """Patch subprocess.run to serve fixed ps + inspect responses."""
    import subprocess

    def mock_run(cmd, **kwargs):
        result = MagicMock()
        if len(cmd) >= 2 and cmd[1] == "ps":
            result.returncode = 0
            result.stdout = ps_output
            result.stderr = ""
            return result
        if len(cmd) >= 2 and cmd[1] == "inspect":
            if inspect_payload is None:
                result.returncode = 1
                result.stdout = ""
                result.stderr = "inspect failed"
                return result
            result.returncode = 0
            result.stdout = json.dumps(inspect_payload)
            result.stderr = ""
            return result
        result.returncode = 1
        result.stdout = ""
        result.stderr = "unexpected command"
        return result

    monkeypatch.setattr(subprocess, "run", mock_run)


# ── LocalContainerBackend.list_running() ─────────────────────────────────────


def test_list_running_returns_containers(monkeypatch):
    """list_running should enumerate containers via docker ps and batch-inspect them."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")

    _mock_ps_and_inspect(
        monkeypatch,
        ps_output="deer-flow-sandbox-abc12345\ndeer-flow-sandbox-def67890\n",
        inspect_payload=[
            _make_inspect_entry("deer-flow-sandbox-abc12345", "2026-04-08T01:22:50.000000000Z", "8081"),
            _make_inspect_entry("deer-flow-sandbox-def67890", "2026-04-08T02:22:50.000000000Z", "8082"),
        ],
    )

    infos = backend.list_running()

    assert len(infos) == 2
    ids = {info.sandbox_id for info in infos}
    assert ids == {"abc12345", "def67890"}
    urls = {info.sandbox_url for info in infos}
    assert "http://localhost:8081" in urls
    assert "http://localhost:8082" in urls


def test_list_running_empty_when_no_containers(monkeypatch):
    """list_running should return empty list when docker ps returns nothing."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")
    _mock_ps_and_inspect(monkeypatch, ps_output="", inspect_payload=[])

    assert backend.list_running() == []


def test_list_running_skips_non_matching_names(monkeypatch):
    """list_running should skip containers whose names don't match the prefix pattern."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")

    _mock_ps_and_inspect(
        monkeypatch,
        ps_output="deer-flow-sandbox-abc12345\nsome-other-container\n",
        inspect_payload=[
            _make_inspect_entry("deer-flow-sandbox-abc12345", "2026-04-08T01:22:50Z", "8081"),
        ],
    )

    infos = backend.list_running()
    assert len(infos) == 1
    assert infos[0].sandbox_id == "abc12345"


def test_list_running_includes_containers_without_port(monkeypatch):
    """Containers without a port mapping should still be listed (with empty URL)."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")

    _mock_ps_and_inspect(
        monkeypatch,
        ps_output="deer-flow-sandbox-abc12345\n",
        inspect_payload=[
            _make_inspect_entry("deer-flow-sandbox-abc12345", "2026-04-08T01:22:50Z", host_port=None),
        ],
    )

    infos = backend.list_running()
    assert len(infos) == 1
    assert infos[0].sandbox_id == "abc12345"
    assert infos[0].sandbox_url == ""


def test_list_running_handles_docker_failure(monkeypatch):
    """list_running should return empty list when docker ps fails."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")

    import subprocess

    def mock_run(cmd, **kwargs):
        result = MagicMock()
        result.returncode = 1
        result.stdout = ""
        result.stderr = "daemon not running"
        return result

    monkeypatch.setattr(subprocess, "run", mock_run)

    assert backend.list_running() == []


def test_list_running_handles_inspect_failure(monkeypatch):
    """list_running should return empty list when batch inspect fails."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")

    _mock_ps_and_inspect(
        monkeypatch,
        ps_output="deer-flow-sandbox-abc12345\n",
        inspect_payload=None,  # Signals inspect failure
    )

    assert backend.list_running() == []


def test_list_running_handles_malformed_inspect_json(monkeypatch):
    """list_running should return empty list when docker inspect emits invalid JSON."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")

    import subprocess

    def mock_run(cmd, **kwargs):
        result = MagicMock()
        if len(cmd) >= 2 and cmd[1] == "ps":
            result.returncode = 0
            result.stdout = "deer-flow-sandbox-abc12345\n"
            result.stderr = ""
        else:
            result.returncode = 0
            result.stdout = "this is not json"
            result.stderr = ""
        return result

    monkeypatch.setattr(subprocess, "run", mock_run)

    assert backend.list_running() == []


def test_list_running_uses_single_batch_inspect_call(monkeypatch):
    """list_running should issue exactly ONE docker inspect call regardless of container count."""
    backend = _make_local_backend()
    monkeypatch.setattr(backend, "_runtime", "docker")

    inspect_call_count = {"count": 0}

    import subprocess

    def mock_run(cmd, **kwargs):
        result = MagicMock()
        if len(cmd) >= 2 and cmd[1] == "ps":
            result.returncode = 0
            result.stdout = "deer-flow-sandbox-a\ndeer-flow-sandbox-b\ndeer-flow-sandbox-c\n"
            result.stderr = ""
            return result
        if len(cmd) >= 2 and cmd[1] == "inspect":
            inspect_call_count["count"] += 1
            # Expect all three names passed in a single call
            assert cmd[2:] == ["deer-flow-sandbox-a", "deer-flow-sandbox-b", "deer-flow-sandbox-c"]
            result.returncode = 0
            result.stdout = json.dumps(
                [
                    _make_inspect_entry("deer-flow-sandbox-a", "2026-04-08T01:22:50Z", "8081"),
                    _make_inspect_entry("deer-flow-sandbox-b", "2026-04-08T01:22:50Z", "8082"),
                    _make_inspect_entry("deer-flow-sandbox-c", "2026-04-08T01:22:50Z", "8083"),
                ]
            )
            result.stderr = ""
            return result
        result.returncode = 1
        result.stdout = ""
        return result

    monkeypatch.setattr(subprocess, "run", mock_run)

    infos = backend.list_running()
    assert len(infos) == 3
    assert inspect_call_count["count"] == 1  # ← The core performance assertion


# ── _parse_docker_timestamp() ────────────────────────────────────────────────


def test_parse_docker_timestamp_with_nanoseconds():
    """Should correctly parse Docker's ISO 8601 timestamp with nanoseconds."""
    from deerflow.community.aio_sandbox.local_backend import _parse_docker_timestamp

    ts = _parse_docker_timestamp("2026-04-08T01:22:50.123456789Z")
    assert ts > 0
    expected = datetime(2026, 4, 8, 1, 22, 50, tzinfo=UTC).timestamp()
    assert abs(ts - expected) < 1.0


def test_parse_docker_timestamp_without_fractional_seconds():
    """Should parse plain ISO 8601 timestamps without fractional seconds."""
    from deerflow.community.aio_sandbox.local_backend import _parse_docker_timestamp

    ts = _parse_docker_timestamp("2026-04-08T01:22:50Z")
    expected = datetime(2026, 4, 8, 1, 22, 50, tzinfo=UTC).timestamp()
    assert abs(ts - expected) < 1.0


def test_parse_docker_timestamp_empty_returns_zero():
    from deerflow.community.aio_sandbox.local_backend import _parse_docker_timestamp

    assert _parse_docker_timestamp("") == 0.0
    assert _parse_docker_timestamp("not a timestamp") == 0.0


# ── _extract_host_port() ─────────────────────────────────────────────────────


def test_extract_host_port_returns_mapped_port():
    from deerflow.community.aio_sandbox.local_backend import _extract_host_port

    entry = {"NetworkSettings": {"Ports": {"8080/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8081"}]}}}
    assert _extract_host_port(entry, 8080) == 8081


def test_extract_host_port_returns_none_when_unmapped():
    from deerflow.community.aio_sandbox.local_backend import _extract_host_port

    entry = {"NetworkSettings": {"Ports": {}}}
    assert _extract_host_port(entry, 8080) is None


def test_extract_host_port_handles_missing_fields():
    from deerflow.community.aio_sandbox.local_backend import _extract_host_port

    assert _extract_host_port({}, 8080) is None
    assert _extract_host_port({"NetworkSettings": None}, 8080) is None


# ── AioSandboxProvider._reconcile_orphans() ──────────────────────────────────


def _make_shared_ownership_store(**kwargs):
    """A store two provider instances can share.

    Sharing one store object between two providers is how these tests model two
    gateway instances pointed at one Redis: the provider only ever sees the
    ``SandboxOwnershipStore`` ABC, so the ownership behaviour exercised here is
    backend-agnostic. The redis backend's own semantics are pinned separately in
    ``test_sandbox_ownership_store.py``.
    """
    from deerflow.community.aio_sandbox.ownership.memory import MemoryOwnershipStore

    kwargs.setdefault("ttl_seconds", 600)
    return MemoryOwnershipStore(owner_id="__shared__", **kwargs)


class _ScopedOwnershipStore:
    """View of a shared store as seen by one instance (rebinds ``owner_id``).

    Rebinding is a test-only trick to model two instances against one store, and
    it is only sound while one thread at a time is inside the shared store. The
    heartbeat-hold tests deliberately run a main-thread ``take`` as worker-b
    concurrently with a heartbeat-thread ``claim`` as worker-a, so the rebind has
    to be serialized with the call it applies to: otherwise a ``claim`` can
    execute under worker-b's id, read worker-a's ``del:`` lease as a peer's,
    return ``False``, and kill the heartbeat — after which the marker lapses and
    the next ``take`` succeeds spuriously. The GIL makes that window small, not
    absent, so the tests flake rather than fail.
    """

    # Serializes owner_id rebind + call across every view of one shared store.
    # A class-level lock is deliberate: each worker gets its own view object, and
    # the thing being guarded is the single shared store they both mutate.
    _rebind_lock = threading.Lock()

    def __init__(self, shared, owner_id: str):
        self._shared = shared
        self._owner_id = owner_id

    @property
    def owner_id(self) -> str:
        return self._owner_id

    @property
    def supports_cross_process(self) -> bool:
        return True

    def _as_me(self, fn, *args):
        with self._rebind_lock:
            previous = self._shared._owner_id
            self._shared._owner_id = self._owner_id
            try:
                return fn(*args)
            finally:
                self._shared._owner_id = previous

    def take(self, sandbox_id):
        return self._as_me(self._shared.take, sandbox_id)

    def claim(self, sandbox_id, *, for_destroy: bool = False):
        return self._as_me(lambda sid: self._shared.claim(sid, for_destroy=for_destroy), sandbox_id)

    def renew(self, sandbox_id):
        return self._as_me(self._shared.renew, sandbox_id)

    def release(self, sandbox_id):
        return self._as_me(self._shared.release, sandbox_id)

    def owner(self, sandbox_id):
        return self._shared.owner(sandbox_id)

    def close(self):
        pass


def _make_provider_for_reconciliation(tmp_path=None, *, worker_id: str = "worker-test", store=None):
    """Build a minimal AioSandboxProvider without triggering __init__ side effects.

    WARNING: This helper intentionally bypasses ``__init__`` via ``__new__`` so
    tests don't depend on Docker or touch the real idle-checker/renewal threads.
    The downside is that this helper is tightly coupled to the set of attributes
    set up in ``AioSandboxProvider.__init__``.  If ``__init__`` gains a new
    attribute that ``_reconcile_orphans`` (or other methods under test) reads,
    this helper must be updated in lockstep — otherwise tests will fail with a
    confusing ``AttributeError`` instead of a meaningful assertion failure.

    Pass a shared *store* (see ``_make_shared_ownership_store``) to two providers
    to model two gateway instances coordinating through one ownership backend.
    ``tmp_path`` is accepted and ignored: ownership no longer lives on disk.
    """
    from deerflow.config.sandbox_config import SandboxOwnershipConfig

    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)
    provider._lock = threading.Lock()
    provider._sandboxes = {}
    provider._sandbox_infos = {}
    provider._thread_sandboxes = {}
    provider._thread_locks = {}
    provider._last_activity = {}
    provider._warm_pool = {}
    provider._active_sandbox_identity = {}
    provider._warm_pool_identity = {}
    provider._unowned_since = {}
    provider._local_teardown = set()
    provider._acquire_epoch = {}
    provider._acquire_epoch_counter = 0
    provider._acquire_inflight = {}
    provider._shutdown_called = False
    provider._idle_checker_stop = threading.Event()
    provider._idle_checker_thread = None
    provider._renewal_stop = threading.Event()
    provider._renewal_thread = None
    provider._config = {
        "idle_timeout": 600,
        "replicas": 3,
    }
    provider._backend = MagicMock()
    provider._owner_id = worker_id
    provider._ownership_config = SandboxOwnershipConfig()
    if store is None:
        from deerflow.community.aio_sandbox.ownership.memory import MemoryOwnershipStore

        provider._ownership = MemoryOwnershipStore(owner_id=worker_id, ttl_seconds=600)
    else:
        provider._ownership = _ScopedOwnershipStore(store, worker_id)
    return provider


def test_reconcile_adopts_old_containers_into_warm_pool(tmp_path):
    """Lease-free containers are adopted into warm pool — idle checker handles cleanup."""
    provider = _make_provider_for_reconciliation(tmp_path)
    now = time.time()

    old_info = SandboxInfo(
        sandbox_id="old12345",
        sandbox_url="http://localhost:8081",
        container_name="deer-flow-sandbox-old12345",
        created_at=now - 1200,  # 20 minutes old, > 600s idle_timeout
    )
    provider._backend.list_running.return_value = [old_info]

    provider._reconcile_orphans()

    # Should NOT destroy directly — let idle checker handle it
    provider._backend.destroy.assert_not_called()
    assert "old12345" in provider._warm_pool


def test_reconcile_adopts_young_containers(tmp_path):
    """Young lease-free containers are adopted into warm pool for potential reuse."""
    provider = _make_provider_for_reconciliation(tmp_path)
    now = time.time()

    young_info = SandboxInfo(
        sandbox_id="young123",
        sandbox_url="http://localhost:8082",
        container_name="deer-flow-sandbox-young123",
        created_at=now - 60,  # 1 minute old, < 600s idle_timeout
    )
    provider._backend.list_running.return_value = [young_info]

    provider._reconcile_orphans()

    provider._backend.destroy.assert_not_called()
    assert "young123" in provider._warm_pool
    adopted_info, release_ts = provider._warm_pool["young123"]
    assert adopted_info.sandbox_id == "young123"


def test_reconcile_mixed_containers_all_adopted(tmp_path):
    """All lease-free containers (old and young) are adopted into warm pool."""
    provider = _make_provider_for_reconciliation(tmp_path)
    now = time.time()

    old_info = SandboxInfo(
        sandbox_id="old_one",
        sandbox_url="http://localhost:8081",
        container_name="deer-flow-sandbox-old_one",
        created_at=now - 1200,
    )
    young_info = SandboxInfo(
        sandbox_id="young_one",
        sandbox_url="http://localhost:8082",
        container_name="deer-flow-sandbox-young_one",
        created_at=now - 60,
    )
    provider._backend.list_running.return_value = [old_info, young_info]

    provider._reconcile_orphans()

    provider._backend.destroy.assert_not_called()
    assert "old_one" in provider._warm_pool
    assert "young_one" in provider._warm_pool


def test_reconcile_skips_already_tracked_containers(tmp_path):
    """Containers already in _sandboxes or _warm_pool should be skipped."""
    provider = _make_provider_for_reconciliation(tmp_path)
    now = time.time()

    existing_info = SandboxInfo(
        sandbox_id="existing1",
        sandbox_url="http://localhost:8081",
        container_name="deer-flow-sandbox-existing1",
        created_at=now - 1200,
    )
    # Pre-populate _sandboxes to simulate already-tracked container
    provider._sandboxes["existing1"] = MagicMock()
    provider._backend.list_running.return_value = [existing_info]

    provider._reconcile_orphans()

    provider._backend.destroy.assert_not_called()
    # The pre-populated sandbox should NOT be moved into warm pool
    assert "existing1" not in provider._warm_pool


def test_reconcile_handles_backend_failure(tmp_path):
    """Reconciliation should not crash if backend.list_running() fails."""
    provider = _make_provider_for_reconciliation(tmp_path)
    provider._backend.list_running.side_effect = RuntimeError("docker not available")

    # Should not raise
    provider._reconcile_orphans()

    assert provider._warm_pool == {}


def test_reconcile_no_running_containers(tmp_path):
    """Reconciliation with no running containers is a no-op."""
    provider = _make_provider_for_reconciliation(tmp_path)
    provider._backend.list_running.return_value = []

    provider._reconcile_orphans()

    provider._backend.destroy.assert_not_called()
    assert provider._warm_pool == {}


def test_reconcile_skips_container_owned_by_peer():
    """#4206: do not adopt a container another instance still owns."""
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    now = time.time()
    info = SandboxInfo(
        sandbox_id="shared01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-shared01",
        created_at=now - 50,
    )
    worker_a._publish_ownership("shared01")
    worker_b._backend.list_running.return_value = [info]

    worker_b._reconcile_orphans()

    assert "shared01" not in worker_b._warm_pool
    worker_b._backend.destroy.assert_not_called()
    # The lease is still A's — B's failed claim must not have stolen it.
    assert shared.owner("shared01") == "worker-a"


def test_idle_reap_does_not_destroy_peer_owned_warm_entry():
    """#4206: idle reaper must not stop a container another instance owns."""
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    worker_b._config["idle_timeout"] = 60
    now = time.time()
    info = SandboxInfo(
        sandbox_id="a99c8444",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-a99c8444",
        created_at=now - 50,
    )
    # Simulate the bad old path: B already has it in warm (or adopted wrongly).
    worker_b._warm_pool["a99c8444"] = (info, now - 61)
    worker_a._publish_ownership("a99c8444")

    worker_b._reap_expired_warm(idle_timeout=60)

    worker_b._backend.destroy.assert_not_called()


def test_multi_worker_release_then_peer_reconcile_cannot_kill():
    """#4206 issue-log path: A release→warm; B reconcile+reap must not destroy."""
    shared = _make_shared_ownership_store()
    destroyed: list[str] = []
    running: dict[str, SandboxInfo] = {}

    def list_running():
        return list(running.values())

    def destroy(info: SandboxInfo):
        destroyed.append(info.sandbox_id)
        running.pop(info.sandbox_id, None)

    backend = MagicMock()
    backend.list_running.side_effect = list_running
    backend.destroy.side_effect = destroy

    sid = "a99c8444"
    info = SandboxInfo(
        sandbox_id=sid,
        sandbox_url="http://localhost:8080",
        container_name=f"deer-flow-sandbox-{sid}",
        created_at=time.time() - 50,
    )
    running[sid] = info

    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_a._backend = backend
    worker_a._config["idle_timeout"] = 60
    # A released to warm and holds the lease.
    worker_a._warm_pool[sid] = (info, time.time())
    worker_a._publish_ownership(sid)

    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    worker_b._backend = backend
    worker_b._config["idle_timeout"] = 60
    worker_b._reconcile_orphans()
    assert sid not in worker_b._warm_pool

    # Even if B somehow had it warm, reap must refuse.
    worker_b._warm_pool[sid] = (info, time.time() - 61)
    worker_b._reap_expired_warm(idle_timeout=60)
    assert sid not in destroyed
    assert sid in running
    assert sid in worker_a._warm_pool


def test_expired_lease_lets_peer_adopt_crashed_owner_container():
    """The crash path still works: once a dead owner's lease lapses, adopt it.

    The counterpart to the tests above — ownership must not become a permanent
    leak when the owning instance dies without releasing. Adoption is delayed by
    the recovery grace, but a dead owner never republishes, so it still happens.
    """
    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    shared = _make_shared_ownership_store(ttl_seconds=0.05)
    dead = _make_provider_for_reconciliation(worker_id="worker-dead", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="crashed1",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-crashed1",
        created_at=time.time() - 50,
    )
    dead._publish_ownership("crashed1")
    worker_b._backend.list_running.return_value = [info]

    # Owner "crashes": stops renewing. Its lease lapses in the store.
    time.sleep(0.1)

    now = time.time()
    with patch.object(aio_mod.time, "time", return_value=now):
        worker_b._reconcile_orphans()
    assert "crashed1" not in worker_b._warm_pool, "adopted a lapsed lease without waiting out the recovery grace"

    # The dead owner never republishes, so the grace runs out and B adopts.
    with patch.object(aio_mod.time, "time", return_value=now + compute_lease_ttl(worker_b._ownership_config) + 1):
        worker_b._reconcile_orphans()

    assert "crashed1" in worker_b._warm_pool
    assert shared.owner("crashed1") == "worker-b"


# ── Ownership store rework (#4206): fail-closed publish, renewal independence ──


def test_acquire_fails_closed_when_ownership_cannot_be_published():
    """Establishment is fail-closed: never hand out a sandbox we could not own.

    The provider used to swallow the lease-write error and return the sandbox id
    on the next line, so a store outage silently disabled the only cross-instance
    exclusion while the sandbox was handed out as usable — peers then saw an
    unowned live container and reaped it.
    """
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError

    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    worker._ownership = MagicMock()
    worker._ownership.take.side_effect = OwnershipBackendError("store down")

    info = SandboxInfo(
        sandbox_id="new001",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-new001",
        created_at=time.time(),
    )

    with pytest.raises(OwnershipBackendError):
        worker._register_created_sandbox("t1", "new001", info, user_id="u1")

    # The just-created container must not be leaked as an unowned orphan.
    worker._backend.destroy.assert_called_once_with(info)
    assert "new001" not in worker._sandboxes


def test_reuse_fails_closed_when_ownership_cannot_be_published():
    """Same fail-closed rule on the in-process reuse path."""
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError

    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    info = SandboxInfo(
        sandbox_id="sb1",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-sb1",
        created_at=time.time(),
    )
    worker._sandboxes["sb1"] = MagicMock()
    worker._sandbox_infos["sb1"] = info
    worker._thread_sandboxes[("u1", "t1")] = "sb1"
    worker._check_tracked_sandbox_alive = MagicMock(return_value=True)
    worker._ownership = MagicMock()
    worker._ownership.take.side_effect = OwnershipBackendError("store down")

    with pytest.raises(OwnershipBackendError):
        worker._reuse_in_process_sandbox("t1", user_id="u1")


def test_destroy_fails_closed_when_ownership_unknown():
    """A store that cannot answer must not be read as 'container is free'."""
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError

    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    worker._ownership = MagicMock()
    worker._ownership.claim.side_effect = OwnershipBackendError("store down")

    info = SandboxInfo(
        sandbox_id="unknown1",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-unknown1",
        created_at=time.time() - 50,
    )

    worker._destroy_warm_entry("unknown1", info, reason="idle_timeout", still_reapable=lambda: True)

    worker._backend.destroy.assert_not_called()


def test_reconcile_fails_closed_when_ownership_unknown():
    """A store outage must not turn every peer container into an adoptable orphan."""
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError

    worker = _make_provider_for_reconciliation(worker_id="worker-b")
    worker._ownership = MagicMock()
    # Configure what the adoption grace reads, or it short-circuits before the
    # claim: a bare MagicMock answers `owner()` with a truthy mock, which reads
    # as "peer-owned" and defers — so the assertion below would pass without the
    # fail-closed branch ever running (it did exactly that until this was fixed).
    worker._ownership.supports_cross_process = True
    worker._ownership.owner.return_value = None
    worker._ownership.claim.side_effect = OwnershipBackendError("store down")
    info = SandboxInfo(
        sandbox_id="unknown2",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-unknown2",
        created_at=time.time() - 50,
    )
    worker._backend.list_running.return_value = [info]

    # Unowned for a full grace, so the container is adoptable and the only thing
    # left standing between it and the warm pool is the claim.
    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    now = time.time()
    with patch.object(aio_mod.time, "time", return_value=now):
        worker._reconcile_orphans()
    with patch.object(aio_mod.time, "time", return_value=now + compute_lease_ttl(worker._ownership_config) + 1):
        worker._reconcile_orphans()

    assert worker._ownership.claim.called, "the fail-closed claim branch was never reached; this test guards nothing"
    assert "unknown2" not in worker._warm_pool
    worker._backend.destroy.assert_not_called()


@pytest.mark.parametrize("idle_timeout", [0, 600])
def test_init_always_starts_lease_renewal(monkeypatch, idle_timeout):
    """Renewal liveness must not ride on the idle checker's switch.

    ``_renew_active_leases`` used to have exactly one caller — ``_cleanup_idle_resources``
    — and ``__init__`` only starts the idle checker when ``idle_timeout > 0``.
    ``idle_timeout: 0`` is a supported config (``config.example.yaml`` documents it
    as "keep warm VMs until shutdown"), so on that config nothing ever refreshed a
    lease and #4206 returned one TTL later.

    This drives ``__init__`` on purpose: the defect is in *who starts renewal*, so
    a test that calls ``_start_lease_renewal()`` directly passes on the broken code
    and guards nothing.
    """
    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")

    started: list[str] = []
    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_load_config", lambda self: {"idle_timeout": idle_timeout, "replicas": 3, "ownership": None})
    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_create_backend", lambda self: MagicMock())
    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_reconcile_orphans", lambda self: None)
    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_register_signal_handlers", lambda self: None)
    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_start_lease_renewal", lambda self: started.append("renewal"))
    monkeypatch.setattr(aio_mod.AioSandboxProvider, "_start_idle_checker", lambda self: started.append("idle"))
    monkeypatch.setattr(aio_mod.atexit, "register", lambda *a, **k: None)

    aio_mod.AioSandboxProvider()

    assert "renewal" in started, f"renewal must start at idle_timeout={idle_timeout}; ownership liveness cannot depend on the idle reaper"
    assert ("idle" in started) is (idle_timeout > 0)


def test_renewal_keeps_the_sandbox_when_the_store_cannot_answer():
    """The one deliberate exception to fail-closed, and it had no test.

    Everywhere else an unanswerable store means "not ours". Renewal is the
    opposite on purpose: `_refresh_ownership` returns True on an
    `OwnershipBackendError` because unknown is not lost, and the TTL still bounds
    how long a genuinely dead owner holds the lease. Invert it and a Redis outage
    makes every instance drop every active and warm sandbox at once — the same
    fleet-wide eviction the LAPSED/LOST split exists to prevent, which is pinned
    only for the flushed-store path, never for a raising one.
    """
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError

    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    worker._ownership = MagicMock()
    worker._ownership.renew.side_effect = OwnershipBackendError("store down")
    info = SandboxInfo(
        sandbox_id="live02",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-live02",
        created_at=time.time(),
    )
    sandbox = MagicMock()
    worker._sandboxes["live02"] = sandbox
    worker._sandbox_infos["live02"] = info
    worker._thread_sandboxes[("u1", "t1")] = "live02"
    worker._warm_pool["warm02"] = (info, time.time())

    worker._renew_owned_leases()

    assert worker._ownership.renew.called, "renewal never reached the store; this test guards nothing"
    assert "live02" in worker._sandboxes, "a store outage evicted a live sandbox nobody had taken"
    assert ("u1", "t1") in worker._thread_sandboxes
    assert "warm02" in worker._warm_pool, "a store outage dropped a warm entry nobody had taken"
    sandbox.close.assert_not_called()
    worker._backend.destroy.assert_not_called()


def test_renewal_keeps_the_sandbox_when_lapsed_reclaim_cannot_answer():
    """A store outage after LAPSED is still unknown, not proof of a peer.

    ``renew()`` and the follow-up ``claim()`` are separate store round trips. A
    key can be observed absent and the store can then become unreachable before
    the claim. Renewal must keep the sandbox and retry rather than route the
    claim through the ordinary fail-closed reap helper and evict a live entry.
    """
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError, RenewOutcome

    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    worker._ownership = MagicMock()
    worker._ownership.renew.return_value = RenewOutcome.LAPSED
    worker._ownership.claim.side_effect = OwnershipBackendError("store down after lapsed renew")
    info = _info("live03")
    sandbox = MagicMock()
    worker._sandboxes["live03"] = sandbox
    worker._sandbox_infos["live03"] = info
    worker._thread_sandboxes[("u1", "t1")] = "live03"

    worker._renew_owned_leases()

    worker._ownership.claim.assert_called_once_with("live03")
    assert "live03" in worker._sandboxes, "a failed LAPSED re-claim evicted a live sandbox nobody had taken"
    assert worker._thread_sandboxes[("u1", "t1")] == "live03"
    sandbox.close.assert_not_called()
    worker._backend.destroy.assert_not_called()


def test_load_config_carries_the_stream_bridge_section():
    """Hop 1 of the "no extra config for multi-instance" promise.

    The redis inference reads `app_config.stream_bridge`, so `_load_config` has to
    carry it. Nothing pinned this: the only test that drives `__init__`
    monkeypatches `_load_config` wholesale and omits the key entirely, so deleting
    it here left every test green while every config.yaml-native multi-instance
    deployment silently fell back to `memory` — #4206 reopened on exactly the
    deployments the inference exists for.
    """
    from deerflow.config.stream_bridge_config import StreamBridgeConfig

    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)

    bridge = StreamBridgeConfig(type="redis", redis_url="redis://bridge:6379/0")
    app_config = MagicMock()
    app_config.stream_bridge = bridge
    app_config.sandbox = MagicMock(ownership=None, image=None, port=None, container_prefix=None, idle_timeout=600, replicas=3, mounts=[], environment={})

    with patch.object(aio_mod, "get_app_config", return_value=app_config):
        loaded = provider._load_config()

    assert loaded["stream_bridge"] is bridge, "_load_config dropped the stream_bridge section the redis inference reads"


def test_init_infers_redis_ownership_from_a_redis_stream_bridge():
    """Hop 2: `__init__` must actually feed the bridge into the resolver.

    Drives the real `__init__` against a real `AppConfig`-shaped object rather
    than stubbing `_load_config`, because the defect would be in the wiring
    between them — the same reason `test_init_always_starts_lease_renewal` drives
    `__init__` instead of calling `_start_lease_renewal` directly.
    """
    from deerflow.config.stream_bridge_config import StreamBridgeConfig

    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")

    app_config = MagicMock()
    app_config.stream_bridge = StreamBridgeConfig(type="redis", redis_url="redis://bridge:6379/0")
    # No sandbox.ownership section at all: the deployment never configured one.
    app_config.sandbox = MagicMock(ownership=None, image=None, port=None, container_prefix=None, idle_timeout=600, replicas=3, mounts=[], environment={})

    built: list = []

    def fake_store(config, *, owner_id=None):
        built.append(config)
        store = MagicMock()
        store.supports_cross_process = True
        return store

    with (
        patch.object(aio_mod, "get_app_config", return_value=app_config),
        patch.object(aio_mod, "make_sandbox_ownership_store", side_effect=fake_store),
        patch.object(aio_mod.AioSandboxProvider, "_create_backend", lambda self: MagicMock()),
        patch.object(aio_mod.AioSandboxProvider, "_reconcile_orphans", lambda self: None),
        patch.object(aio_mod.AioSandboxProvider, "_register_signal_handlers", lambda self: None),
        patch.object(aio_mod.AioSandboxProvider, "_start_lease_renewal", lambda self: None),
        patch.object(aio_mod.AioSandboxProvider, "_start_idle_checker", lambda self: None),
        patch.object(aio_mod.atexit, "register", lambda *a, **k: None),
    ):
        aio_mod.AioSandboxProvider()

    assert len(built) == 1
    assert built[0].type == "redis", "a redis stream bridge did not infer a redis ownership store; multi-instance deployments silently fall back to memory"
    assert built[0].redis_url == "redis://bridge:6379/0"


def test_renewal_loop_refreshes_owned_leases():
    """The renewal thread actually renews (the loop body, not just its wiring)."""
    from deerflow.config.sandbox_config import SandboxOwnershipConfig

    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    worker._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.05)
    worker._sandboxes["sb1"] = MagicMock()
    worker._publish_ownership("sb1")

    renewed: list[str] = []
    real_renew = worker._ownership.renew

    def counting_renew(sandbox_id):
        renewed.append(sandbox_id)
        return real_renew(sandbox_id)

    worker._ownership.renew = counting_renew

    worker._start_lease_renewal()
    try:
        deadline = time.time() + 3
        while not renewed and time.time() < deadline:
            time.sleep(0.02)
    finally:
        worker._stop_lease_renewal()

    assert renewed == ["sb1"] or renewed[0] == "sb1"


def test_renewal_covers_warm_entries_not_just_active():
    """A warm container is still ours; letting its lease lapse invites adoption."""
    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    info = SandboxInfo(
        sandbox_id="warm01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-warm01",
        created_at=time.time(),
    )
    worker._sandboxes["active01"] = MagicMock()
    worker._warm_pool["warm01"] = (info, time.time())
    worker._publish_ownership("active01")
    worker._publish_ownership("warm01")

    renewed: list[str] = []
    worker._ownership = MagicMock()
    worker._ownership.renew.side_effect = lambda sid: renewed.append(sid) or True

    worker._renew_owned_leases()

    assert set(renewed) == {"active01", "warm01"}


def test_renewal_does_not_forget_a_warm_entry_mid_teardown():
    """Renewal must not pop the warm entry retained by an in-flight stop.

    A warm teardown deliberately leaves the entry visible until the backend stop
    succeeds. Its ``del:us`` marker makes ordinary ``renew()`` report LOST, but
    that is local teardown state rather than a peer takeover. If the stop then
    fails, the warm entry must still be present for retry or reclaim.
    """
    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    info = SandboxInfo(
        sandbox_id="warm-stop",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-warm-stop",
        created_at=time.time(),
    )
    worker._warm_pool["warm-stop"] = (info, time.time())
    worker._publish_ownership("warm-stop")

    stop_started = threading.Event()
    let_stop_fail = threading.Event()

    def failing_stop(_entry):
        stop_started.set()
        assert let_stop_fail.wait(timeout=5), "the test never released the backend stop"
        raise RuntimeError("stop failed")

    worker._backend.destroy.side_effect = failing_stop
    result = {}
    reaper = threading.Thread(
        target=lambda: result.update(
            destroyed=worker._destroy_warm_entry(
                "warm-stop",
                info,
                reason="idle_timeout",
                still_reapable=lambda: True,
            )
        ),
        daemon=True,
    )
    reaper.start()
    try:
        assert stop_started.wait(timeout=5), "the warm teardown never reached the backend stop"
        assert "warm-stop" in worker._local_teardown, "precondition: the local teardown reservation is not held"

        worker._renew_owned_leases()

        assert "warm-stop" in worker._warm_pool, "renewal forgot a warm entry while this instance was stopping it"
    finally:
        let_stop_fail.set()
        reaper.join(timeout=5)

    assert not reaper.is_alive(), "the failed warm teardown did not finish"
    assert result.get("destroyed") is False
    assert "warm-stop" in worker._warm_pool, "a failed stop lost the warm entry it promised to retain"


def test_lost_lease_drops_sandbox_without_destroying_container():
    """Losing the lease means the container is someone else's — drop it, don't kill it.

    Destroying here would be the exact cross-instance kill the store exists to
    prevent, just triggered from the renewal path instead of the reaper. Only our
    host-side handle goes away, and it must be closed rather than leaked (#2872).
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="moved01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-moved01",
        created_at=time.time(),
    )
    sandbox = MagicMock()
    worker_a._sandboxes["moved01"] = sandbox
    worker_a._sandbox_infos["moved01"] = info
    worker_a._thread_sandboxes[("u1", "t1")] = "moved01"
    worker_a._last_activity["moved01"] = time.time()
    worker_a._publish_ownership("moved01")

    # The thread's next turn routes to B, which takes over ownership.
    worker_b._publish_ownership("moved01")

    worker_a._renew_owned_leases()

    assert "moved01" not in worker_a._sandboxes
    assert "moved01" not in worker_a._sandbox_infos
    assert ("u1", "t1") not in worker_a._thread_sandboxes
    worker_a._backend.destroy.assert_not_called()
    sandbox.close.assert_called_once()
    assert shared.owner("moved01") == "worker-b"


def test_ownership_rollback_on_create_closes_the_client_it_drops():
    """The rollback destroys the container; its host-side client must not leak (#2872)."""
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError

    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    worker._ownership = MagicMock()
    worker._ownership.take.side_effect = OwnershipBackendError("store down")
    info = SandboxInfo(
        sandbox_id="new002",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-new002",
        created_at=time.time(),
    )

    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    created: list[MagicMock] = []

    def fake_aio_sandbox(**kwargs):
        sandbox = MagicMock()
        created.append(sandbox)
        return sandbox

    with patch.object(aio_mod, "AioSandbox", side_effect=fake_aio_sandbox):
        with pytest.raises(OwnershipBackendError):
            worker._register_created_sandbox("t1", "new002", info, user_id="u1")

    worker._backend.destroy.assert_called_once_with(info)
    assert created and created[0].close.call_count == 1


def test_acquire_takes_over_ownership_so_a_thread_can_move_instances():
    """A thread's next turn can land on another instance; it must not be stranded.

    Ownership answers "who reaps this", not "who may use it". A conditional claim
    here would refuse while the previous instance's lease was still live and break
    every load-balanced follow-up turn.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="thread01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-thread01",
        created_at=time.time(),
    )
    worker_a._publish_ownership("thread01")
    assert shared.owner("thread01") == "worker-a"

    # B serves the thread's next turn and discovers the existing container.
    assert worker_b._register_discovered_sandbox("t1", info, user_id="u1") == "thread01"
    assert shared.owner("thread01") == "worker-b"


def test_store_losing_all_state_does_not_evict_live_sandboxes():
    """A Redis restart must not drop every in-flight sandbox fleet-wide.

    `renew()` returns falsy for two very different situations: a peer took the
    lease, and the lease is simply absent. Treating them the same meant that when
    Redis restarted without persistence — every key gone, nobody holding
    anything — each instance evicted every sandbox it was actively serving.
    A lapsed lease must be re-established instead.
    """
    shared = _make_shared_ownership_store()
    worker = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    info = SandboxInfo(
        sandbox_id="live01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-live01",
        created_at=time.time(),
    )
    sandbox = MagicMock()
    worker._sandboxes["live01"] = sandbox
    worker._sandbox_infos["live01"] = info
    worker._thread_sandboxes[("u1", "t1")] = "live01"
    worker._publish_ownership("live01")

    # The store loses everything, as a Redis restart without persistence does.
    shared._leases.clear()

    worker._renew_owned_leases()

    assert "live01" in worker._sandboxes, "a store restart evicted a live sandbox nobody had taken"
    assert ("u1", "t1") in worker._thread_sandboxes
    sandbox.close.assert_not_called()
    worker._backend.destroy.assert_not_called()
    # And it is ours again, so peers still cannot reap it.
    assert shared.owner("live01") == "worker-a"


def test_peer_reconcile_after_state_loss_does_not_steal_a_live_container():
    """The other half of the store-restart case: a peer must not adopt first.

    ``_refresh_ownership`` already refuses to read an absent lease as
    abandonment. Reconciliation must not contradict it on the other path: after
    the store loses every key, each live owner is still serving its containers
    and simply has not reached its next renewal tick. An instance reconciling in
    that window sees no lease and would adopt every one of them; the real owner's
    next renewal then reports LOST and it drops a sandbox mid-turn, which the
    adopter later idle-destroys — #4206 through the back door.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    info = SandboxInfo(
        sandbox_id="live01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-live01",
        created_at=time.time(),
    )
    sandbox = MagicMock()
    worker_a._sandboxes["live01"] = sandbox
    worker_a._sandbox_infos["live01"] = info
    worker_a._thread_sandboxes[("u1", "t1")] = "live01"
    worker_a._publish_ownership("live01")

    # The store loses everything, as a Redis restart without persistence does.
    # Worker A is alive and still serving live01.
    shared._leases.clear()

    # A peer starts up and reconciles before A's renewal tick fires.
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    worker_b._backend.list_running.return_value = [info]
    worker_b._reconcile_orphans()

    assert "live01" not in worker_b._warm_pool, "a peer adopted a container whose owner is still alive and serving it"

    # A's renewal tick finally fires: it must still own and keep the sandbox.
    worker_a._renew_owned_leases()

    assert "live01" in worker_a._sandboxes, "a peer's reconcile evicted a live sandbox after the store lost its state"
    assert ("u1", "t1") in worker_a._thread_sandboxes
    sandbox.close.assert_not_called()
    worker_b._backend.destroy.assert_not_called()
    assert shared.owner("live01") == "worker-a"


def test_adoption_grace_expires_so_a_truly_orphaned_container_is_still_adopted():
    """The grace must delay adoption, not disable it.

    A container that stays unowned across a full lease TTL has no live owner —
    a surviving owner republishes within one renewal interval, which is shorter
    than the TTL by construction. Reconciliation must adopt it then, or a crashed
    instance's containers would leak forever.
    """
    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    shared = _make_shared_ownership_store()
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    ttl = compute_lease_ttl(worker_b._ownership_config)
    info = SandboxInfo(
        sandbox_id="crashed1",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-crashed1",
        created_at=time.time() - 50,
    )
    worker_b._backend.list_running.return_value = [info]

    now = time.time()
    with patch.object(aio_mod.time, "time", return_value=now):
        worker_b._reconcile_orphans()
    assert "crashed1" not in worker_b._warm_pool, "adopted a keyless container without waiting out the recovery grace"

    # Nobody republished the lease across a full TTL: the owner is really gone.
    with patch.object(aio_mod.time, "time", return_value=now + ttl + 1):
        worker_b._reconcile_orphans()

    assert "crashed1" in worker_b._warm_pool, "the grace never expired, so a crashed owner's container would leak forever"
    assert shared.owner("crashed1") == "worker-b"


def test_adoption_grace_restarts_when_a_live_owner_republishes():
    """A republished lease must reset the grace, not just pause it.

    Reset and pause only diverge on a **second** lapse. Pause leaves the original
    timestamp behind, so the next time the lease drops the grace is already spent
    and the adopter takes a live container with no wait at all. Stopping at "A
    republished, B defers" would prove nothing — a paused timer defers there too,
    because the container simply reads as owned. So the second lapse is the whole
    test; without it this passes with the reset deleted.
    """
    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    ttl = compute_lease_ttl(worker_b._ownership_config)
    info = SandboxInfo(
        sandbox_id="live01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-live01",
        created_at=time.time(),
    )
    worker_b._backend.list_running.return_value = [info]

    now = time.time()
    # B starts its grace on a container that currently looks unowned.
    with patch.object(aio_mod.time, "time", return_value=now):
        worker_b._reconcile_orphans()

    # A republishes mid-grace (its renewal tick re-establishing a lapsed lease).
    worker_a._publish_ownership("live01")

    with patch.object(aio_mod.time, "time", return_value=now + ttl + 1):
        worker_b._reconcile_orphans()

    assert "live01" not in worker_b._warm_pool, "a stale grace expired over a lease a live owner had already republished"
    assert shared.owner("live01") == "worker-a"

    # The republish must have cleared B's timer, not merely paused it. A second
    # blip drops the key again: B has to serve a *fresh* full grace, which A's
    # next renewal tick will beat. A paused timer would still hold the original
    # start, so B would adopt A's live container instantly, with no grace at all.
    assert "live01" not in worker_b._unowned_since, "the republish left a stale grace timer behind"

    shared._leases.clear()
    with patch.object(aio_mod.time, "time", return_value=now + ttl + 2):
        worker_b._reconcile_orphans()

    assert "live01" not in worker_b._warm_pool, "a grace timer left over from before the republish expired instantly on the next lapse"


def test_acquire_refuses_a_container_a_peer_is_destroying():
    """#4206's last window: `take()` must not overrun a destroyer's claim.

    Sequence: B's reaper claims X for destroy and starts the (slow) container
    stop; a turn for X's thread routes to A. An unconditional takeover would hand
    A a sandbox that B's stop is about to kill mid-turn.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="dying01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-dying01",
        created_at=time.time(),
    )

    # B decides to reap it and marks the teardown, then its stop is in flight.
    assert worker_b._claim_ownership("dying01", for_destroy=True) is True

    # A's acquire must refuse rather than hand out a doomed container.
    with pytest.raises(SandboxBeingDestroyedError):
        worker_a._register_discovered_sandbox("t1", info, user_id="u1")

    assert "dying01" not in worker_a._sandboxes


def test_teardown_marker_is_held_for_a_stop_that_outlives_the_lease_ttl():
    """The `del:` state must not expire out from under an in-flight stop.

    `test_acquire_refuses_a_container_a_peer_is_destroying` above proves the
    marker refuses a takeover — but never lets it expire. `claim(for_destroy)`
    writes it with the ordinary lease TTL and nothing refreshes it: `renew()`
    only extends `own:` and reports a teardown as LOST, and the destroy paths
    drop the sandbox from the maps the renewal loop iterates. So a container
    stop that outlives the TTL let the marker lapse, a peer's `take()` then
    succeeded against the still-running container, and the stop landed on the
    turn that had just been handed it — the very window `del:` exists to close,
    reopened by its own expiry. The `flock` this replaced could not expire; a
    lease can, so it has to be held on purpose.
    """
    from deerflow.config.sandbox_config import SandboxOwnershipConfig

    lease_ttl = 0.15
    shared = _make_shared_ownership_store(ttl_seconds=lease_ttl)
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    # A legal config: the schema bounds only renewal > 0 and multiplier >= 2.
    worker_a._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.05, ttl_multiplier=3.0)
    info = SandboxInfo(
        sandbox_id="doomed1",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-doomed1",
        created_at=time.time(),
    )

    stop_entered = threading.Event()
    release_stop = threading.Event()

    def slow_destroy(entry):
        stop_entered.set()
        release_stop.wait(timeout=5)

    worker_a._backend.destroy = MagicMock(side_effect=slow_destroy)
    worker_a._warm_pool["doomed1"] = (info, time.time())

    reaper = threading.Thread(
        target=lambda: worker_a._destroy_warm_entry("doomed1", info, reason="idle_timeout", still_reapable=lambda: True),
        daemon=True,
    )
    reaper.start()
    try:
        assert stop_entered.wait(timeout=5), "the reaper never reached the backend stop"

        # Across a span several times the lease TTL, a turn for this thread must
        # keep being refused — the container is still being stopped.
        deadline = time.time() + lease_ttl * 4
        while time.time() < deadline:
            assert not worker_b._ownership.take("doomed1"), "a peer took a container whose stop was still in flight"
            time.sleep(0.02)
    finally:
        release_stop.set()
        reaper.join(timeout=5)

    # Once the stop returns the marker is dropped, so the thread can cold-start.
    assert shared.owner("doomed1") is None, "the teardown marker outlived the stop that justified it"
    assert worker_b._ownership.take("doomed1") is True


def test_unhealthy_drop_holds_the_teardown_marker_for_its_stop():
    """The third `del:`-marked stop path needs the same hold as the other two.

    `_drop_unhealthy_sandbox` claims for destroy and then blocks on the backend
    stop exactly like `_destroy_warm_entry` and `destroy()`. Its sibling test
    `test_unhealthy_sandbox_owned_by_peer_is_not_destroyed` pins the *gate* — a
    peer-owned container is not stopped — but never lets the marker **expire**
    during an in-flight stop, which is the same blind spot that hid this window
    on the other two paths. It untracks before claiming, so `_renew_owned_leases`
    cannot see the id either: nothing refreshes the marker unless the stop holds
    it.
    """
    from deerflow.config.sandbox_config import SandboxOwnershipConfig

    lease_ttl = 0.15
    shared = _make_shared_ownership_store(ttl_seconds=lease_ttl)
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    worker_a._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.05, ttl_multiplier=3.0)
    info = SandboxInfo(
        sandbox_id="sick01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-sick01",
        created_at=time.time(),
    )

    stop_entered = threading.Event()
    release_stop = threading.Event()

    def slow_destroy(entry):
        stop_entered.set()
        release_stop.wait(timeout=5)

    worker_a._backend.destroy = MagicMock(side_effect=slow_destroy)
    worker_a._sandboxes["sick01"] = MagicMock()
    worker_a._sandbox_infos["sick01"] = info

    dropper = threading.Thread(
        target=lambda: worker_a._drop_unhealthy_sandbox("sick01", "health check failed"),
        daemon=True,
    )
    dropper.start()
    try:
        assert stop_entered.wait(timeout=5), "the drop never reached the backend stop"

        deadline = time.time() + lease_ttl * 4
        while time.time() < deadline:
            assert not worker_b._ownership.take("sick01"), "a peer took a container whose unhealthy-drop stop was still in flight"
            time.sleep(0.02)
    finally:
        release_stop.set()
        dropper.join(timeout=5)

    assert shared.owner("sick01") is None, "the teardown marker outlived the stop that justified it"


def test_destroy_holds_the_teardown_marker_for_its_stop():
    """The third of the three `del:`-marked stops, and the one with no test.

    `_destroy_warm_entry` and `_drop_unhealthy_sandbox` each have a held-marker
    test; `destroy()` is wrapped but nothing pins the wrap, so deleting it goes
    unnoticed. "Every path does X" claims keep leaving exactly one sibling
    untested — this is that sibling.
    """
    from deerflow.config.sandbox_config import SandboxOwnershipConfig

    lease_ttl = 0.15
    shared = _make_shared_ownership_store(ttl_seconds=lease_ttl)
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    worker_a._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.05, ttl_multiplier=3.0)
    info = SandboxInfo(
        sandbox_id="doomed3",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-doomed3",
        created_at=time.time(),
    )

    stop_entered = threading.Event()
    release_stop = threading.Event()

    def slow_destroy(entry):
        stop_entered.set()
        release_stop.wait(timeout=5)

    worker_a._backend.destroy = MagicMock(side_effect=slow_destroy)
    worker_a._sandboxes["doomed3"] = MagicMock()
    worker_a._sandbox_infos["doomed3"] = info

    destroyer = threading.Thread(target=lambda: worker_a.destroy("doomed3"), daemon=True)
    destroyer.start()
    try:
        assert stop_entered.wait(timeout=5), "destroy never reached the backend stop"

        deadline = time.time() + lease_ttl * 4
        while time.time() < deadline:
            assert not worker_b._ownership.take("doomed3"), "a peer took a container whose destroy() stop was still in flight"
            time.sleep(0.02)
    finally:
        release_stop.set()
        destroyer.join(timeout=5)

    assert shared.owner("doomed3") is None, "the teardown marker outlived the stop that justified it"


def test_destroy_releases_the_teardown_marker_when_the_stop_fails():
    """A failed stop must not strand the id under a `del:` marker.

    `_destroy_warm_entry` releases on both outcomes and says why: the container is
    still up, so leaving it marked would block its thread from ever re-acquiring
    it. `destroy()` had no such guard — a raising backend propagated straight past
    the release, and `take()` stays refused against `del:`, so the thread could not
    acquire until the TTL lapsed. Fails safe rather than fatal (nobody stops a live
    container), but the three stop paths must agree on this.

    The error still propagates: `shutdown()` logs per sandbox off it, so releasing
    must not turn into swallowing.
    """
    shared = _make_shared_ownership_store()
    worker = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    info = SandboxInfo(
        sandbox_id="boom01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-boom01",
        created_at=time.time(),
    )
    worker._sandboxes["boom01"] = MagicMock()
    worker._sandbox_infos["boom01"] = info
    worker._backend.destroy = MagicMock(side_effect=RuntimeError("docker daemon is unreachable"))

    with pytest.raises(RuntimeError, match="docker daemon is unreachable"):
        worker.destroy("boom01")

    assert shared.owner("boom01") is None, "a failed stop left the id stranded under a teardown marker"
    # The container may well still be running, so its thread must be able to take
    # it back rather than wait out the TTL.
    assert worker._ownership.take("boom01") is True


class _BlockHeartbeatClaim:
    """Store view whose heartbeat *refresh* claim blocks until released.

    ``claim(for_destroy=True)`` is issued twice per teardown: once by the
    caller's gate (``_claim_ownership``) and then repeatedly by the heartbeat.
    This lets the gate through and blocks from the second onward, so a refresh
    can be held in flight while the context manager exits — the interleaving the
    release-ordering fix has to survive.
    """

    def __init__(self, inner):
        self._inner = inner
        self._for_destroy_claims = 0
        self._lock = threading.Lock()
        self.heartbeat_blocked = threading.Event()
        self.unblock = threading.Event()
        self.refresh_completed = threading.Event()

    @property
    def owner_id(self):
        return self._inner.owner_id

    @property
    def supports_cross_process(self):
        return self._inner.supports_cross_process

    def take(self, sandbox_id):
        return self._inner.take(sandbox_id)

    def renew(self, sandbox_id):
        return self._inner.renew(sandbox_id)

    def release(self, sandbox_id):
        return self._inner.release(sandbox_id)

    def owner(self, sandbox_id):
        return self._inner.owner(sandbox_id)

    def close(self):
        pass

    def claim(self, sandbox_id, *, for_destroy: bool = False):
        if for_destroy:
            with self._lock:
                self._for_destroy_claims += 1
                nth = self._for_destroy_claims
            if nth >= 2:  # the heartbeat refresh, not the caller's gate claim
                self.heartbeat_blocked.set()
                self.unblock.wait(timeout=10)
                result = self._inner.claim(sandbox_id, for_destroy=for_destroy)
                # The refresh has now rewritten `del:`. Signalling only after it
                # lands lets the test settle on the final state instead of racing
                # the transient window between a caller-side release and this write.
                self.refresh_completed.set()
                return result
        return self._inner.claim(sandbox_id, for_destroy=for_destroy)


def test_teardown_join_budget_covers_refresh_and_final_release():
    """A normally timing-out refresh plus release must fit before deferral.

    Redis bounds each store operation at five seconds. Context exit can catch
    the heartbeat in a refresh and must then wait for its final release, so the
    join budget needs to exceed both sequential operation bounds rather than
    only one of them.
    """
    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    store_operation_timeout_seconds = 5.0

    assert aio_mod.AioSandboxProvider._TEARDOWN_JOIN_TIMEOUT_SECONDS > 2 * store_operation_timeout_seconds


def test_teardown_release_waits_for_the_heartbeat_to_exit():
    """A refresh still in flight when the stop finishes must not resurrect `del:`.

    The `del:` marker's final release is the heartbeat's own last act, not the
    caller's. If the caller cleared it, a refresh `claim` still blocked in the
    store (redis has no infinitely-patient call, but it can be mid-round-trip)
    would land *after* that release and rewrite `del:` on a container whose stop
    already completed — refusing a fresh `take()` (or rolling back a fresh
    create) until the TTL. Owning the release inside the heartbeat sequences it
    strictly after the last refresh. (fancyboi999, PR #4221)
    """
    from deerflow.config.sandbox_config import SandboxOwnershipConfig

    # Long store TTL so nothing lapses via TTL during the test — the only thing
    # that may clear the marker is a real release.
    shared = _make_shared_ownership_store(ttl_seconds=30)
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    worker_a._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.02, ttl_multiplier=2.0)
    # Give up on the join while the heartbeat is still blocked, so the context
    # returns with a refresh genuinely in flight.
    worker_a._TEARDOWN_JOIN_TIMEOUT_SECONDS = 0.1
    blocking = _BlockHeartbeatClaim(worker_a._ownership)
    worker_a._ownership = blocking

    info = SandboxInfo(
        sandbox_id="defer1",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-defer1",
        created_at=time.time(),
    )

    def slow_destroy(entry):
        # Return only once the heartbeat has entered (and blocked in) a refresh,
        # so the context exit runs its bounded join against an in-flight claim.
        assert blocking.heartbeat_blocked.wait(timeout=5), "the heartbeat never issued a refresh claim"

    worker_a._backend.destroy = MagicMock(side_effect=slow_destroy)
    worker_a._warm_pool["defer1"] = (info, time.time())

    reaper = threading.Thread(
        target=lambda: worker_a._destroy_warm_entry("defer1", info, reason="idle_timeout", still_reapable=lambda: True),
        daemon=True,
    )
    reaper.start()
    # The join times out (heartbeat blocked), so the destroy path returns while
    # the refresh is still in flight — exactly where the old caller-side release ran.
    reaper.join(timeout=10)
    assert not reaper.is_alive(), "the destroy path never returned while the heartbeat was blocked"

    # Let the in-flight refresh complete: it rewrites `del:`. Wait until it has
    # actually landed, so the assertion reads the settled state rather than the
    # transient window between a caller-side release and this write. Only a
    # heartbeat-owned release, sequenced after the refresh, can leave the id clean.
    blocking.unblock.set()
    assert blocking.refresh_completed.wait(timeout=5), "the in-flight refresh never completed"

    deadline = time.time() + 5
    while time.time() < deadline and shared.owner("defer1") is not None:
        time.sleep(0.02)
    assert shared.owner("defer1") is None, "a refresh that landed after the release stranded the id under a `del:` marker"
    assert worker_b._ownership.take("defer1") is True


def test_evict_keeps_the_warm_entry_when_the_claim_is_refused():
    """Replica eviction must not pop before it knows the container is going away.

    The sibling of `test_refused_idle_destroy_keeps_the_warm_entry`, which pins
    the same rule for the idle path. Popping first on a refused claim loses the
    container: still running, owned by a peer, and no longer in any of our maps —
    so nothing here would ever reap or reclaim it.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="peer01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-peer01",
        created_at=time.time(),
    )
    worker_a._warm_pool["peer01"] = (info, time.time() - 5)
    # A peer owns it, so A's eviction claim is refused.
    worker_b._publish_ownership("peer01")

    assert worker_a._evict_oldest_warm() is None

    worker_a._backend.destroy.assert_not_called()
    assert "peer01" in worker_a._warm_pool, "evicting popped a container it was refused permission to destroy"
    assert shared.owner("peer01") == "worker-b"


def test_reclaim_drops_a_container_a_peer_is_destroying():
    """The warm-pool half of the acquire-side teardown refusal.

    `test_cached_sandbox_being_destroyed_is_dropped_not_reused` pins the
    in-process reuse path and `test_acquire_refuses_a_container_a_peer_is_destroying`
    the discover path; reclaim is the third and had no test. It must not raise
    (the caller falls through to a cold start) and must not leave the doomed
    container in the warm pool.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="dying02",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-dying02",
        created_at=time.time(),
    )
    worker_a._warm_pool["dying02"] = (info, time.time())
    worker_a._check_tracked_sandbox_alive = MagicMock(return_value=True)

    # B's reaper marks the teardown; its stop is in flight.
    assert worker_b._claim_ownership("dying02", for_destroy=True) is True

    reclaimed = worker_a._reclaim_warm_pool_sandbox("t1", "dying02", user_id="u1")

    assert reclaimed is None, "reclaimed a container a peer is tearing down"
    assert "dying02" not in worker_a._warm_pool
    assert "dying02" not in worker_a._sandboxes
    worker_a._backend.destroy.assert_not_called()


def test_created_sandbox_is_rolled_back_when_a_peer_is_destroying_its_id():
    """Rollback must cover a teardown marker, not just a store outage.

    `test_ownership_rollback_on_create_closes_the_client_it_drops` drives this
    path with `OwnershipBackendError` only. The comment says the teardown case is
    reachable too — a peer that died mid-stop leaves a `del:` marker until its
    TTL lapses — and without rollback the container we just started is leaked as
    an adoptable orphan.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="fresh01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-fresh01",
        created_at=time.time(),
    )
    # A peer's teardown marker is still on this id when we finish creating.
    assert worker_b._claim_ownership("fresh01", for_destroy=True) is True

    with pytest.raises(SandboxBeingDestroyedError):
        worker_a._register_created_sandbox("t1", "fresh01", info, user_id="u1")

    worker_a._backend.destroy.assert_called_once_with(info)
    assert "fresh01" not in worker_a._sandboxes, "a container we could not own was handed out anyway"


def test_shutdown_does_not_stop_a_peers_warm_container():
    """Shutdown is a reap path and must be gated like every other one.

    Nothing drove `shutdown()` with a non-empty warm pool, so a loop that called
    `_backend.destroy` directly — skipping the ownership claim — would go
    unnoticed. On a multi-instance gateway that is #4206 on the shutdown path:
    our exit stops a container a live peer is serving.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    mine = SandboxInfo(sandbox_id="mine01", sandbox_url="http://localhost:8080", container_name="c-mine01", created_at=time.time())
    theirs = SandboxInfo(sandbox_id="peer02", sandbox_url="http://localhost:8081", container_name="c-peer02", created_at=time.time())

    worker_a._warm_pool["mine01"] = (mine, time.time())
    worker_a._publish_ownership("mine01")
    worker_a._warm_pool["peer02"] = (theirs, time.time())
    worker_b._publish_ownership("peer02")  # a live peer owns this one

    worker_a.shutdown()

    destroyed = {call.args[0].sandbox_id for call in worker_a._backend.destroy.call_args_list}
    assert "mine01" in destroyed, "shutdown left our own warm container running"
    assert "peer02" not in destroyed, "shutdown stopped a container a live peer owns"
    assert shared.owner("peer02") == "worker-b"


def test_teardown_heartbeat_stops_when_the_stop_returns():
    """A finite TTL must survive the fix, or a crashed destroyer leaks forever.

    The heartbeat is what holds the exclusion, so it has to die with the stop:
    if it outlived the destroy the marker would be refreshed indefinitely and no
    peer could ever adopt or recreate the container.
    """
    from deerflow.config.sandbox_config import SandboxOwnershipConfig

    shared = _make_shared_ownership_store(ttl_seconds=0.15)
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_a._ownership_config = SandboxOwnershipConfig(renewal_interval_seconds=0.05, ttl_multiplier=3.0)
    info = SandboxInfo(
        sandbox_id="doomed2",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-doomed2",
        created_at=time.time(),
    )
    worker_a._warm_pool["doomed2"] = (info, time.time())

    assert worker_a._destroy_warm_entry("doomed2", info, reason="idle_timeout", still_reapable=lambda: True) is True

    # Named rather than counted: threading.active_count() is global and other
    # tests' idle-checker/renewal threads make it noise, so a count comparison
    # here passes straight through a leak.
    assert [t for t in threading.enumerate() if t.name == "sandbox-teardown-lease"] == [], "a teardown heartbeat thread outlived its stop"

    # And nothing keeps refreshing the marker past its TTL.
    time.sleep(0.3)
    assert shared.owner("doomed2") is None


def test_cached_sandbox_being_destroyed_is_dropped_not_reused():
    """The same window on the warm/in-process reuse path falls through cleanly."""
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="dying02",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-dying02",
        created_at=time.time(),
    )
    sandbox = MagicMock()
    worker_a._sandboxes["dying02"] = sandbox
    worker_a._sandbox_infos["dying02"] = info
    worker_a._thread_sandboxes[("u1", "t1")] = "dying02"
    worker_a._check_tracked_sandbox_alive = MagicMock(return_value=True)

    worker_b._claim_ownership("dying02", for_destroy=True)

    # Returns None (not the id, and not an exception) so acquire cold-starts.
    assert worker_a._reuse_in_process_sandbox("t1", user_id="u1") is None
    assert "dying02" not in worker_a._sandboxes
    worker_a._backend.destroy.assert_not_called()


def test_destroy_claims_before_untracking():
    """A refused claim must not lose the container from every map.

    Untracking first meant a peer-owned container was dropped from `_sandboxes`
    and `_warm_pool` and then not destroyed — still running, and now invisible to
    the instance that had been tracking it.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="peer01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-peer01",
        created_at=time.time(),
    )
    sandbox = MagicMock()
    worker_a._sandboxes["peer01"] = sandbox
    worker_a._sandbox_infos["peer01"] = info
    worker_b._publish_ownership("peer01")

    worker_a.destroy("peer01")

    worker_a._backend.destroy.assert_not_called()
    assert "peer01" in worker_a._sandboxes, "untracked a container it was refused permission to destroy"
    sandbox.close.assert_not_called()


def test_refused_idle_destroy_keeps_the_warm_entry():
    """Popping before deciding loses the container: running, tracked by nobody."""
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    worker_a._config["idle_timeout"] = 60
    info = SandboxInfo(
        sandbox_id="warmpeer",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-warmpeer",
        created_at=time.time(),
    )
    worker_a._warm_pool["warmpeer"] = (info, time.time() - 999)
    worker_b._publish_ownership("warmpeer")

    worker_a._reap_expired_warm(idle_timeout=60)

    worker_a._backend.destroy.assert_not_called()
    assert "warmpeer" in worker_a._warm_pool, "dropped a warm entry it did not actually destroy"


def test_unhealthy_sandbox_owned_by_peer_is_not_destroyed():
    """The one reap path that used to skip the ownership gate entirely."""
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = SandboxInfo(
        sandbox_id="sick01",
        sandbox_url="http://localhost:8080",
        container_name="deer-flow-sandbox-sick01",
        created_at=time.time(),
    )
    worker_a._sandboxes["sick01"] = MagicMock()
    worker_a._sandbox_infos["sick01"] = info
    worker_b._publish_ownership("sick01")

    worker_a._drop_unhealthy_sandbox("sick01", "failed health check")

    worker_a._backend.destroy.assert_not_called()
    assert shared.owner("sick01") == "worker-b"


def test_get_does_not_touch_ownership_store():
    """get() is a pure in-memory lookup — it must not do store IO.

    ``ensure_sandbox_initialized_async`` calls ``provider.get()`` directly on the
    event loop, so store IO here would be blocking filesystem or network IO on
    the hot path. Ownership is published on acquire and refreshed by the renewal
    thread instead.
    """
    worker = _make_provider_for_reconciliation(worker_id="worker-a")
    sandbox = MagicMock()
    worker._sandboxes["sb1"] = sandbox
    worker._ownership = MagicMock()

    assert worker.get("sb1") is sandbox

    worker._ownership.take.assert_not_called()
    worker._ownership.claim.assert_not_called()
    worker._ownership.renew.assert_not_called()
    worker._ownership.owner.assert_not_called()


def test_reconcile_multiple_containers_all_adopted(tmp_path):
    """Multiple lease-free containers should all be adopted into warm pool."""
    provider = _make_provider_for_reconciliation(tmp_path)
    now = time.time()

    info1 = SandboxInfo(sandbox_id="cont_one", sandbox_url="http://localhost:8081", created_at=now - 1200)
    info2 = SandboxInfo(sandbox_id="cont_two", sandbox_url="http://localhost:8082", created_at=now - 1200)

    provider._backend.list_running.return_value = [info1, info2]

    provider._reconcile_orphans()

    provider._backend.destroy.assert_not_called()
    assert "cont_one" in provider._warm_pool
    assert "cont_two" in provider._warm_pool


def test_reconcile_zero_created_at_adopted():
    """Containers with created_at=0 (unknown age) should still be adopted into warm pool."""
    provider = _make_provider_for_reconciliation()

    info = SandboxInfo(sandbox_id="unknown1", sandbox_url="http://localhost:8081", created_at=0.0)
    provider._backend.list_running.return_value = [info]

    provider._reconcile_orphans()

    provider._backend.destroy.assert_not_called()
    assert "unknown1" in provider._warm_pool


def test_reconcile_idle_timeout_zero_adopts_all():
    """When idle_timeout=0 (disabled), all containers are still adopted into warm pool."""
    provider = _make_provider_for_reconciliation()
    provider._config["idle_timeout"] = 0
    now = time.time()

    old_info = SandboxInfo(sandbox_id="old_one", sandbox_url="http://localhost:8081", created_at=now - 7200)
    young_info = SandboxInfo(sandbox_id="young_one", sandbox_url="http://localhost:8082", created_at=now - 60)
    provider._backend.list_running.return_value = [old_info, young_info]

    provider._reconcile_orphans()

    provider._backend.destroy.assert_not_called()
    assert "old_one" in provider._warm_pool
    assert "young_one" in provider._warm_pool


# ── SIGHUP signal handler ───────────────────────────────────────────────────


def test_sighup_handler_registered():
    """SIGHUP handler should be registered on Unix systems."""
    if not hasattr(signal, "SIGHUP"):
        pytest.skip("SIGHUP not available on this platform")

    provider = _make_provider_for_reconciliation()

    # Save original handlers for ALL signals we'll modify
    original_sighup = signal.getsignal(signal.SIGHUP)
    original_sigterm = signal.getsignal(signal.SIGTERM)
    original_sigint = signal.getsignal(signal.SIGINT)
    try:
        aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
        provider._original_sighup = original_sighup
        provider._original_sigterm = original_sigterm
        provider._original_sigint = original_sigint
        provider.shutdown = MagicMock()

        aio_mod.AioSandboxProvider._register_signal_handlers(provider)

        # Verify SIGHUP handler is no longer the default
        handler = signal.getsignal(signal.SIGHUP)
        assert handler != signal.SIG_DFL, "SIGHUP handler should be registered"
    finally:
        # Restore ALL original handlers to avoid leaking state across tests
        signal.signal(signal.SIGHUP, original_sighup)
        signal.signal(signal.SIGTERM, original_sigterm)
        signal.signal(signal.SIGINT, original_sigint)


# ── Same-process reap vs. promote ────────────────────────────────────────────
#
# The ownership store excludes peers and nothing else: `claim()` and `take()`
# both succeed against our *own* lease by design. So between this instance's
# reaper threads and its own acquire path there is no store-level exclusion at
# all, and every reaper decides outside `_lock` (a store round trip must not be
# held under the lock that guards every acquire).
#
# Each test below blocks its reaper inside the store round trip — where the
# window actually lives — lets a promote path run, then reads the SETTLED state
# after both threads have finished. The assertion is never on a transient.


def _active_sandbox(provider, sandbox_id, info, *, thread_key=("u", "t1")):
    """Track *info* as an active sandbox on *provider*."""
    from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox

    provider._sandboxes[sandbox_id] = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
    provider._sandbox_infos[sandbox_id] = info
    provider._last_activity[sandbox_id] = time.time()
    provider._thread_sandboxes[thread_key] = sandbox_id
    provider._backend.is_alive.return_value = True


def _info(sandbox_id, *, created_at=None):
    return SandboxInfo(
        sandbox_id=sandbox_id,
        sandbox_url="http://localhost:8080",
        container_name=f"deer-flow-sandbox-{sandbox_id}",
        created_at=time.time() if created_at is None else created_at,
    )


class _GateOnClaim:
    """Store view that holds the first ``claim`` until released.

    Models the Redis round trip the reaper makes outside ``_lock`` — the window
    a same-process acquire slips through.
    """

    def __init__(self, inner):
        self._inner = inner
        self.entered = threading.Event()
        # Not named `release`: __getattr__ forwards everything else to the real
        # store, and an attribute by that name would shadow `store.release()`.
        self.let_through = threading.Event()
        self._armed = True

    def __getattr__(self, name):
        return getattr(self._inner, name)

    @property
    def supports_cross_process(self):
        return True

    def claim(self, sandbox_id, *, for_destroy=False):
        if self._armed:
            self._armed = False
            self.entered.set()
            assert self.let_through.wait(timeout=5), "the test never released the gated claim"
        return self._inner.claim(sandbox_id, for_destroy=for_destroy)


def _run_reap_vs_promote(provider, gate, reap, promote):
    """Run *reap* until it blocks in the store, then *promote*; settle and return."""
    reaper = threading.Thread(target=reap, daemon=True)
    reaper.start()
    try:
        assert gate.entered.wait(timeout=5), "the reaper never reached the ownership store"
        handed_out = promote()
    finally:
        gate.let_through.set()
        reaper.join(timeout=5)
    assert not reaper.is_alive(), "the reaper never finished"
    return handed_out


@pytest.mark.parametrize(
    ("reason", "reap_name"),
    [("replica_enforcement", "_evict_oldest_warm"), ("idle_timeout", "_reap_expired_warm")],
)
def test_warm_reaper_does_not_stop_a_container_this_instance_just_reclaimed(reason, reap_name):
    """Both warm reapers defer the pop, so both need the same reservation.

    The deferred pop is deliberate — popping first loses the container on a
    refused claim — but it leaves the entry visible in `_warm_pool` for the whole
    stop, so `_reclaim_warm_pool_sandbox` can promote it and hand it to an agent
    while the stop is in flight. `claim(for_destroy=True)` does not refuse this:
    the lease is already `own:us`, and a claim only refuses *peers*.

    On `main` neither reaper could hit this — `WarmPoolLifecycleMixin` popped the
    entry under the lock before destroying — so this is a regression these
    overrides introduce, not a pre-existing gap.
    """
    provider = _make_provider_for_reconciliation()
    gate = _GateOnClaim(provider._ownership)
    provider._ownership = gate
    info = _info("warm1")
    # Old enough for the idle reaper; the only entry, so also the eviction pick.
    provider._warm_pool["warm1"] = (info, time.time() - 10_000)
    provider._ownership._inner.take("warm1")

    reap = provider._evict_oldest_warm if reap_name == "_evict_oldest_warm" else lambda: provider._reap_expired_warm(600)
    handed_out = _run_reap_vs_promote(
        provider,
        gate,
        reap,
        lambda: provider._reclaim_warm_pool_sandbox("t1", "warm1", user_id="u"),
    )

    assert handed_out is None, f"{reap_name} let a turn reclaim a container it was stopping ({reason})"
    assert "warm1" not in provider._sandboxes, "the reclaimed sandbox was promoted into active tracking"


def test_idle_destroy_does_not_stop_a_container_this_instance_just_reused():
    """The idle checker's "still idle?" re-check must gate the reservation, not precede it.

    `_cleanup_idle_sandboxes` re-verified idleness under the lock and then called
    `destroy()`, which claims ownership *before* untracking — so the re-check and
    the act are separated by a store round trip. A turn that reuses the sandbox
    in that window gets its container stopped mid-turn.

    The window is pre-existing in shape (main re-checks and destroys the same
    way) but this PR widened it from a few instructions to a network round trip
    by adding the ownership claim, so it is in scope here.
    """
    provider = _make_provider_for_reconciliation()
    gate = _GateOnClaim(provider._ownership)
    provider._ownership = gate
    info = _info("idle1")
    _active_sandbox(provider, "idle1", info)
    provider._last_activity["idle1"] = 0.0  # long idle
    provider._ownership._inner.take("idle1")

    handed_out = _run_reap_vs_promote(
        provider,
        gate,
        lambda: provider._cleanup_idle_sandboxes(600),
        lambda: provider._reuse_in_process_sandbox("t1", user_id="u"),
    )

    assert handed_out is None, "a turn reused a sandbox whose container the idle checker was stopping"


def test_reuse_does_not_return_a_sandbox_reserved_during_its_health_check():
    """A teardown that wins during reuse's unlocked health check must stay won.

    Reuse checks the local teardown reservation before calling ``is_alive``, but
    that backend call runs outside ``_lock``.  The idle reaper can reserve the id
    in that gap and then pause in its ownership claim while reuse publishes a
    fresh ``own:`` lease.  Map membership alone is not a safe final check: the
    reaper deliberately keeps the active entry tracked until its claim succeeds.
    """
    provider = _make_provider_for_reconciliation()
    gate = _GateOnClaim(provider._ownership)
    provider._ownership = gate
    info = _info("idlerace")
    _active_sandbox(provider, "idlerace", info)
    provider._last_activity["idlerace"] = 0.0
    provider._ownership._inner.take("idlerace")

    health_started, let_health_finish = threading.Event(), threading.Event()

    def gated_health(_info):
        health_started.set()
        assert let_health_finish.wait(timeout=5), "the test never released the health check"
        return True

    provider._backend.is_alive.side_effect = gated_health
    result = {}
    acquire = threading.Thread(target=lambda: result.update(id=provider._reuse_in_process_sandbox("t1", user_id="u")), daemon=True)
    reaper = threading.Thread(target=lambda: provider._cleanup_idle_sandboxes(600), daemon=True)

    acquire.start()
    try:
        assert health_started.wait(timeout=5), "reuse never reached its unlocked health check"
        reaper.start()
        assert gate.entered.wait(timeout=5), "the idle reaper never reserved the sandbox and reached its ownership claim"
        assert "idlerace" in provider._local_teardown, "precondition: the idle reaper must hold the local reservation"

        let_health_finish.set()
        acquire.join(timeout=5)
        assert not acquire.is_alive(), "reuse never completed"
        handed_out = result.get("id")
    finally:
        let_health_finish.set()
        gate.let_through.set()
        acquire.join(timeout=5)
        reaper.join(timeout=5)

    assert not reaper.is_alive(), "the idle reaper never finished"
    assert handed_out is None, "reuse returned a sandbox whose local teardown reservation had already won"
    provider._backend.destroy.assert_called_once_with(info)
    assert provider.get("idlerace") is None


def test_discovery_does_not_install_a_sandbox_reserved_while_publishing():
    """Discovery must re-check the local reservation after its store round trip."""
    provider = _make_provider_for_reconciliation()
    info = _info("discoverrace")
    published, let_install = threading.Event(), threading.Event()
    real_publish = provider._publish_ownership

    def gated_publish(sandbox_id):
        result = real_publish(sandbox_id)
        published.set()
        assert let_install.wait(timeout=5), "the test never released discovery after publish"
        return result

    provider._publish_ownership = gated_publish
    result = {}

    def register():
        try:
            result["id"] = provider._register_discovered_sandbox("t1", info, user_id="u")
        except Exception as e:
            result["error"] = e

    registrar = threading.Thread(target=register, daemon=True)
    registrar.start()
    try:
        assert published.wait(timeout=5), "discovery never published ownership"
        assert provider._reserve_local_teardown("discoverrace", lambda: True), "the test could not reserve the discovered sandbox"
        let_install.set()
        registrar.join(timeout=5)
    finally:
        let_install.set()
        registrar.join(timeout=5)
        provider._finish_local_teardown("discoverrace")

    assert not registrar.is_alive(), "discovery never completed"
    assert isinstance(result.get("error"), SandboxBeingDestroyedError)
    assert result.get("id") is None
    assert "discoverrace" not in provider._sandboxes
    assert provider.get("discoverrace") is None


def test_unhealthy_drop_refuses_discovery_of_the_container_it_is_stopping():
    """`_drop_unhealthy_sandbox` untracks first, so discovery is its open window.

    Once the maps are cleared, an acquire misses both caches and falls through to
    backend discovery, which finds the still-running container. `take()` only
    refuses a `del:` lease, and this path's claim has not run yet — so without a
    local reservation the acquire is handed a container this instance is about
    to stop.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("sick1")
    _active_sandbox(provider, "sick1", info)
    provider._ownership.take("sick1")
    # Block at the claim, not at the stop: after the claim the `del:` marker is
    # already published and the store refuses the take by itself, so gating any
    # later would pass without the reservation existing at all. The window that
    # needs the reservation is untrack -> claim, where the lease still says
    # `own:us` and `take()` succeeds.
    gate = _GateOnClaim(provider._ownership)
    provider._ownership = gate

    reaper = threading.Thread(target=lambda: provider._drop_unhealthy_sandbox("sick1", "failed health check"), daemon=True)
    reaper.start()
    try:
        assert gate.entered.wait(timeout=5), "the drop never reached the ownership claim"
        assert provider._ownership._inner.owner("sick1") == provider._owner_id, "precondition: the lease must still read as ours, not a teardown"
        with pytest.raises(SandboxBeingDestroyedError):
            provider._register_discovered_sandbox("t1", info, user_id="u")
    finally:
        gate.let_through.set()
        reaper.join(timeout=5)

    assert "sick1" not in provider._sandboxes


def test_renewal_does_not_drop_a_sandbox_this_instance_re_acquired():
    """A `LOST` verdict is stale the moment an acquire takes the lease back.

    `_renew_owned_leases` calls `renew()` outside `_lock`; between that answer
    and `_forget_lost_sandbox`'s pop, this instance's own acquire can `take()`
    the lease back and hand the sandbox to a turn. The reuse path hands out the
    **same** tracked `AioSandbox`, so an object-identity check would not notice —
    the pop then closes a client mid-turn and the agent's tool calls fail until
    the next turn re-discovers.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = _info("moved1")
    _active_sandbox(worker_a, "moved1", info)
    worker_b._ownership.take("moved1")  # peer holds it -> renew() will report LOST

    entered = threading.Event()
    release = threading.Event()
    original_forget = worker_a._forget_lost_sandbox

    def gated_forget(sandbox_id, **kwargs):
        entered.set()
        assert release.wait(timeout=5)
        return original_forget(sandbox_id, **kwargs)

    worker_a._forget_lost_sandbox = gated_forget

    reaper = threading.Thread(target=worker_a._renew_owned_leases, daemon=True)
    reaper.start()
    try:
        assert entered.wait(timeout=5), "renewal never decided the lease was lost"
        assert worker_a._reuse_in_process_sandbox("t1", user_id="u") == "moved1"
    finally:
        release.set()
        reaper.join(timeout=5)

    assert shared.owner("moved1") == "worker-a", "the acquire should have taken the lease back"
    assert "moved1" in worker_a._sandboxes, "renewal dropped a sandbox this instance owns again"


def test_release_does_not_drop_a_warm_entry_this_instance_re_acquired():
    """`release()` refreshes ownership outside the lock and has the same staleness.

    Sibling of the renewal path: the turn ends, `release()` parks the entry warm
    and refreshes its lease, and the thread's *next* turn can reclaim it while
    that round trip is in flight.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = _info("parked1")
    _active_sandbox(worker_a, "parked1", info)
    worker_b._ownership.take("parked1")  # peer holds it -> refresh reports LOST

    entered = threading.Event()
    release_gate = threading.Event()
    original_forget = worker_a._forget_lost_sandbox

    def gated_forget(sandbox_id, **kwargs):
        entered.set()
        assert release_gate.wait(timeout=5)
        return original_forget(sandbox_id, **kwargs)

    worker_a._forget_lost_sandbox = gated_forget

    reaper = threading.Thread(target=lambda: worker_a.release("parked1"), daemon=True)
    reaper.start()
    try:
        assert entered.wait(timeout=5), "release never decided the lease was lost"
        assert worker_a._reclaim_warm_pool_sandbox("t1", "parked1", user_id="u") == "parked1"
    finally:
        release_gate.set()
        reaper.join(timeout=5)

    assert shared.owner("parked1") == "worker-a"
    assert "parked1" in worker_a._sandboxes, "release dropped a warm entry this instance re-acquired"


def test_discovered_sandbox_client_is_closed_when_ownership_publish_fails():
    """The discover path owns a host-side client even though it owns no container.

    "Nothing to roll back" is true of the container — we did not create it — but
    not of the `AioSandbox` HTTP client constructed before the publish. The
    sibling create path already closes it on the same failure.
    """
    from deerflow.community.aio_sandbox.ownership import OwnershipBackendError

    provider = _make_provider_for_reconciliation()
    provider._ownership = MagicMock()
    provider._ownership.take.side_effect = OwnershipBackendError("store down")

    created = []
    aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
    real_sandbox_cls = aio_mod.AioSandbox

    def tracking_sandbox(**kwargs):
        sandbox = real_sandbox_cls(**kwargs)
        sandbox.close = MagicMock(side_effect=sandbox.close)
        created.append(sandbox)
        return sandbox

    with patch.object(aio_mod, "AioSandbox", side_effect=tracking_sandbox):
        with pytest.raises(OwnershipBackendError):
            provider._register_discovered_sandbox("t1", _info("found1"), user_id="u")

    assert len(created) == 1, "the discover path should have constructed exactly one client"
    created[0].close.assert_called_once()
    assert "found1" not in provider._sandboxes


def test_idle_destroy_skips_a_sandbox_re_acquired_after_the_idle_snapshot():
    """The idle predicate must read live activity, not just gate the reservation.

    `_cleanup_idle_sandboxes` snapshots idle candidates under the lock and acts
    on them one at a time; a turn landing in between makes the snapshot wrong.
    The reservation alone does not catch this — nothing is being torn down yet —
    so the "still idle?" check has to travel with it as a live predicate.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("idle2")
    _active_sandbox(provider, "idle2", info)
    provider._last_activity["idle2"] = 0.0
    provider._ownership.take("idle2")

    real_destroy_tracked = provider._destroy_tracked

    def turn_lands_first(sandbox_id, **kwargs):
        # A turn arrives between the idle snapshot and the reservation.
        provider._last_activity[sandbox_id] = time.time()
        return real_destroy_tracked(sandbox_id, **kwargs)

    provider._destroy_tracked = turn_lands_first
    provider._cleanup_idle_sandboxes(600)

    provider._backend.destroy.assert_not_called()
    assert "idle2" in provider._sandboxes, "idle cleanup destroyed a sandbox that was re-acquired after the snapshot"


# ── Guard visibility vs. the transition it guards ────────────────────────────
#
# A guard must become visible no later than the state transition it guards. The
# epoch alone cannot satisfy that for `take()`: the takeover is durable before
# `take()` returns (redis has committed the SET while the reply is in flight),
# and the epoch can only be written after it returns. So the acquire path
# publishes an *intent* mark before the round trip, and the epoch covers the
# other half — "an acquire completed since you decided".


class _TakeCommitsThenPauses:
    """Store view whose ``take`` performs the real write, then pauses.

    Not an artificial injection point: on redis the server has committed the
    takeover while the reply is still travelling back, so `own:us` is externally
    visible before the caller resumes.
    """

    def __init__(self, inner):
        self._inner = inner
        self.committed = threading.Event()
        self.let_return = threading.Event()
        self._armed = True

    def __getattr__(self, name):
        return getattr(self._inner, name)

    @property
    def supports_cross_process(self):
        return True

    def take(self, sandbox_id):
        result = self._inner.take(sandbox_id)
        if self._armed:
            self._armed = False
            self.committed.set()
            assert self.let_return.wait(timeout=5), "the test never released take()"
        return result


def _hold_forget(provider):
    """Park *provider*'s next `_forget_lost_sandbox` and return its two events."""
    ready, release = threading.Event(), threading.Event()
    original = provider._forget_lost_sandbox

    def gated(sandbox_id, **kwargs):
        ready.set()
        assert release.wait(timeout=5), "the test never released the forget"
        return original(sandbox_id, **kwargs)

    provider._forget_lost_sandbox = gated
    return ready, release


def test_renewal_does_not_drop_a_sandbox_whose_takeover_is_mid_flight():
    """The epoch is written after `take()` returns; the takeover is durable before it.

    In that interval the store already says the container is ours while the
    epoch still reads as it did when the renewal decided `LOST`. Without a guard
    established *before* the round trip, the stale forget walks through, drops
    the maps and closes the client — and the acquire path then returns an id the
    provider no longer tracks, so `get()` answers `None` for the rest of the turn.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = _info("midtake")
    _active_sandbox(worker_a, "midtake", info)
    worker_b._ownership.take("midtake")  # peer holds it -> renew() reports LOST

    ready, release = _hold_forget(worker_a)
    renewal = threading.Thread(target=worker_a._renew_owned_leases, daemon=True)
    renewal.start()
    assert ready.wait(timeout=5), "renewal never decided the lease was lost"

    gate = _TakeCommitsThenPauses(worker_a._ownership)
    worker_a._ownership = gate
    result = {}
    acquire = threading.Thread(target=lambda: result.update(id=worker_a._reuse_in_process_sandbox("t1", user_id="u")), daemon=True)
    acquire.start()
    try:
        assert gate.committed.wait(timeout=5), "take() never committed the takeover"
        assert shared.owner("midtake") == "worker-a", "precondition: the takeover must already be durable"
        release.set()
        renewal.join(timeout=5)
    finally:
        gate.let_return.set()
        acquire.join(timeout=5)

    assert result.get("id") == "midtake"
    assert "midtake" in worker_a._sandboxes, "a stale LOST dropped a sandbox whose takeover was already committed"
    assert worker_a.get("midtake") is not None, "acquire returned an id whose get() is None"


def test_reuse_falls_through_when_its_entry_is_dropped_while_publishing():
    """Before the intent mark is set, a `LOST` forget is both current and correct.

    Distinct from the mid-flight window: here nothing is wrong with the forget —
    the peer really does hold the lease and no acquire has taken it back yet, so
    the epoch matches and no intent is registered. The bug is on the other side:
    reuse decided to hand out a tracked entry, and must not return that decision
    once the entry is gone. Falling through re-discovers and builds a fresh
    client; returning the id would hand back a sandbox whose `get()` is `None`.
    """
    shared = _make_shared_ownership_store()
    worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
    worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
    info = _info("prepub")
    _active_sandbox(worker_a, "prepub", info)
    worker_b._ownership.take("prepub")

    ready, release = _hold_forget(worker_a)
    renewal = threading.Thread(target=worker_a._renew_owned_leases, daemon=True)
    renewal.start()
    assert ready.wait(timeout=5), "renewal never decided the lease was lost"

    # Park reuse in the gap between its map re-check and the intent mark.
    at_publish, let_publish = threading.Event(), threading.Event()
    real_publish = worker_a._publish_ownership

    def gated_publish(sandbox_id):
        at_publish.set()
        assert let_publish.wait(timeout=5)
        return real_publish(sandbox_id)

    worker_a._publish_ownership = gated_publish
    result = {}
    acquire = threading.Thread(target=lambda: result.update(id=worker_a._reuse_in_process_sandbox("t1", user_id="u")), daemon=True)
    acquire.start()
    try:
        assert at_publish.wait(timeout=5), "reuse never reached the ownership publish"
        release.set()
        renewal.join(timeout=5)
    finally:
        let_publish.set()
        acquire.join(timeout=5)

    assert result.get("id") is None, "reuse returned an id it no longer tracks"


def test_reconcile_does_not_adopt_a_container_this_instance_is_tearing_down():
    """Adoption is a promote, so it needs the same reservation check as the rest.

    `_drop_unhealthy_sandbox` untracks before it claims, so in that window the
    container is running, untracked, and still carries our own `own:` lease —
    exactly the shape this loop adopts. Neither guard in the loop excludes it:
    the claim succeeds because the lease is ours, and on the `memory` store the
    recovery grace is skipped outright (`supports_cross_process = False`), so
    nothing stands in the way at all. Adopting parks a container in the warm pool
    moments before its stop lands, leaving a dead entry for the next reclaim.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("reap1")
    _active_sandbox(provider, "reap1", info)
    provider._ownership.take("reap1")
    provider._backend.list_running.return_value = [info]

    at_claim, let_claim = threading.Event(), threading.Event()
    real_claim = provider._claim_ownership

    def gated_claim(sandbox_id, *, for_destroy=False):
        if for_destroy:
            at_claim.set()
            assert let_claim.wait(timeout=5)
        return real_claim(sandbox_id, for_destroy=for_destroy)

    provider._claim_ownership = gated_claim
    reaper = threading.Thread(target=lambda: provider._drop_unhealthy_sandbox("reap1", "failed health check"), daemon=True)
    reaper.start()
    try:
        assert at_claim.wait(timeout=5), "the drop never reached its claim"
        # Untracked, still running, and the `del:` marker is not written yet.
        provider._reconcile_orphans()
        assert "reap1" not in provider._warm_pool, "reconcile adopted a container this instance is stopping"
    finally:
        let_claim.set()
        reaper.join(timeout=5)


def test_teardown_reservation_is_cleared_once_the_stop_returns():
    """The reservation must not outlive the stop it guards.

    Release ordering is the mirror of acquire ordering: the heartbeat drops the
    `del:` lease before `_finish_local_teardown` clears the local mark. That
    direction is the safe one — the mark only refuses *our* promotes, and the
    container is gone anyway — but a mark left behind would keep refusing this
    thread's cold-start until the process restarts.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("cleared")
    provider._warm_pool["cleared"] = (info, time.time() - 10_000)
    provider._ownership.take("cleared")

    assert provider._destroy_warm_entry("cleared", info, reason="idle_timeout", still_reapable=lambda: True) is True
    assert provider._local_teardown == set(), "a teardown reservation outlived the stop it guarded"
    assert provider._acquire_inflight == {}, "an acquire intent mark leaked"


def test_reclaim_does_not_hand_out_a_container_a_reaper_is_stopping():
    """Reclaim's reservation check runs before its round trip, so it must run again.

    A reaper can reserve *after* that first check — the warm entry is still
    there, since the pop is deferred so a refused stop cannot lose the container
    — and then claim `del:`, which succeeds because reclaim's own `take()` just
    made the lease ours. With the reaper parked inside the container stop, the
    entry is still in `_warm_pool` and still reserved, so only the re-check
    stands between reclaim and installing a client for a dying container.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("handout")
    provider._warm_pool["handout"] = (info, time.time() - 10_000)
    provider._ownership.take("handout")

    # Park reclaim between its `take()` and its install.
    published, let_install = threading.Event(), threading.Event()
    real_publish = provider._publish_ownership

    def gated_publish(sandbox_id):
        result = real_publish(sandbox_id)
        published.set()
        assert let_install.wait(timeout=5)
        return result

    provider._publish_ownership = gated_publish
    out = {}
    reclaim = threading.Thread(target=lambda: out.update(id=provider._reclaim_warm_pool_sandbox("t1", "handout", user_id="u")), daemon=True)
    reclaim.start()
    assert published.wait(timeout=5), "reclaim never published ownership"

    # Reaper reserves and enters the stop, then parks there -- so when reclaim
    # resumes the entry is still in `_warm_pool` and the reservation is held.
    in_stop, let_stop_finish = threading.Event(), threading.Event()
    provider._backend.destroy = MagicMock(side_effect=lambda entry: (in_stop.set(), let_stop_finish.wait(timeout=5)))
    reaper = threading.Thread(target=lambda: provider._reap_expired_warm(600), daemon=True)
    reaper.start()
    try:
        assert in_stop.wait(timeout=5), "the reaper never reached the container stop"
        assert "handout" in provider._warm_pool, "precondition: the pop must still be pending"
        let_install.set()
        reclaim.join(timeout=5)
    finally:
        let_stop_finish.set()
        reaper.join(timeout=5)

    assert out.get("id") is None, "reclaim handed out a container a reaper was stopping"
    assert "handout" not in provider._sandboxes


def test_warm_entry_is_removed_before_its_teardown_reservation_is_released():
    """Removal must happen under the reservation, not after it.

    The reservation is what keeps promotes off the entry. If it is released when
    the stop returns and the entry is removed afterwards, there is a gap in which
    the container is already stopped, the entry is still in `_warm_pool`, and
    nothing marks it — so a reclaim in that gap re-checks the reservation, finds
    it clear, and hands out a dead container. Ordering is the guarantee, so the
    assertion is on the ordering rather than on one interleaving.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("ordered")
    provider._warm_pool["ordered"] = (info, time.time() - 10_000)
    provider._ownership.take("ordered")

    observed = {}
    real_finish = provider._finish_local_teardown

    def spy(sandbox_id):
        observed["warm_at_release"] = sandbox_id in provider._warm_pool
        return real_finish(sandbox_id)

    provider._finish_local_teardown = spy
    assert provider._destroy_warm_entry("ordered", info, reason="idle_timeout", still_reapable=lambda: True) is True

    assert observed["warm_at_release"] is False, "the warm entry outlived the reservation that protected it"
    assert "ordered" not in provider._warm_pool


def test_reclaim_short_circuits_a_reserved_entry_without_touching_the_backend():
    """The pre-round-trip reservation check earns its keep as an early-out.

    Correctness is covered by the re-check after the publish, so this first check
    is not what makes reclaim safe. What it does is refuse a doomed entry before
    spending a container health check and an ownership round trip on it, so that
    is what this pins — otherwise the clause is unanchored and a later edit can
    drop it silently.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("shortcut")
    provider._warm_pool["shortcut"] = (info, time.time())
    provider._ownership = MagicMock(wraps=provider._ownership)
    with provider._lock:
        provider._local_teardown.add("shortcut")

    assert provider._reclaim_warm_pool_sandbox("t1", "shortcut", user_id="u") is None
    provider._backend.is_alive.assert_not_called()
    provider._ownership.take.assert_not_called()


def test_forget_without_an_epoch_still_refuses_an_id_being_acquired():
    """ "No epoch supplied" must not read as "no guard at all".

    The epoch-less callers are the two `SandboxBeingDestroyedError` handlers,
    which cannot collide with a publish for the same id today. This pins the
    primitive's contract rather than that reachability argument: an id whose
    acquire is mid-publish is never dropped, whoever asks.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("guarded")
    _active_sandbox(provider, "guarded", info)
    with provider._lock:
        provider._acquire_inflight["guarded"] = 1

    provider._forget_lost_sandbox("guarded")

    assert "guarded" in provider._sandboxes, "an id being acquired was dropped by an epoch-less forget"


@pytest.mark.parametrize("register", ["discovered", "created"])
def test_registering_a_sandbox_clears_any_stale_warm_entry(register):
    """Active and warm are exclusive states, and only register can violate it.

    Reconciliation adopts an untracked-but-running container into the warm pool,
    and on the `memory` store that happens on sight — `_adoptable_after_grace`
    short-circuits when `supports_cross_process` is False, so an id carrying this
    process's *own* lease is treated as adoptable. That is reachable during the
    publish → track window of either register path, which this branch introduced
    (on `main` the track was a single locked insert with nothing before it).

    Leaving both entries gives one container two reapers: `_reap_expired_warm`
    judges it by the warm timestamp and never consults `_last_activity`, so it
    stops a container an agent is using while `_sandboxes` still hands out its
    client — #4206's symptom on the default backend.

    The exclusivity is what actually fixes this, so it is asserted directly and
    on both register paths rather than through one interleaving.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("stale")
    provider._warm_pool["stale"] = (info, time.time())

    if register == "discovered":
        provider._register_discovered_sandbox("t1", info, user_id="u")
    else:
        provider._register_created_sandbox("t1", "stale", info, user_id="u")

    assert "stale" in provider._sandboxes
    assert "stale" not in provider._warm_pool, "a stale warm entry survived the id becoming active"


def test_a_container_adopted_during_register_is_not_reaped_from_the_warm_pool():
    """The end-to-end harm the exclusivity rule removes.

    Reconcile runs inside the register's publish → track window and adopts the
    container; once tracking lands, the warm entry must be gone, otherwise warm
    expiry stops the container this turn is holding.
    """
    provider = _make_provider_for_reconciliation()
    info = _info("adopted")
    provider._backend.list_running.return_value = [info]
    provider._backend.is_alive.return_value = True

    at_gap, go = threading.Event(), threading.Event()
    real_publish = provider._publish_ownership

    def gated_publish(sandbox_id):
        result = real_publish(sandbox_id)
        at_gap.set()
        assert go.wait(timeout=5)
        return result

    provider._publish_ownership = gated_publish
    out = {}
    registrar = threading.Thread(target=lambda: out.update(id=provider._register_discovered_sandbox("t1", info, user_id="u")), daemon=True)
    registrar.start()
    try:
        assert at_gap.wait(timeout=5), "register never reached the publish → track window"
        provider._reconcile_orphans()
    finally:
        go.set()
        registrar.join(timeout=5)

    assert out.get("id") == "adopted"
    assert "adopted" not in provider._warm_pool, "the adopted entry survived, giving the container two reapers"

    time.sleep(0.02)
    provider._reap_expired_warm(0.01)
    provider._backend.destroy.assert_not_called()
    assert provider.get("adopted") is not None, "warm expiry stopped a container this turn is holding"