"""Pure runtime helpers for the multi-agent orchestrator."""

from __future__ import annotations

import asyncio
import math
import random
import time
from contextlib import suppress
from dataclasses import dataclass, field
from functools import partial
from typing import TYPE_CHECKING, Any

import httpx

from mindroom import constants
from mindroom.cancellation import (
    SYNC_RESTART_CANCEL_MSG,
    USER_STOP_CANCEL_MSG,
    CancelSource,
    cancel_failure_reason,
    cancel_source_from_failure_reason,
    classify_cancel_source,
    request_task_cancel,
)
from mindroom.constants import RuntimePaths, runtime_matrix_ssl_verify
from mindroom.logging_config import get_logger
from mindroom.matrix.health import (
    MATRIX_SYNC_CACHE_WRITE_GRACE_SECONDS,
    MATRIX_SYNC_STARTUP_GRACE_SECONDS,
    MATRIX_SYNC_WATCHDOG_TIMEOUT_SECONDS,
    matrix_versions_url,
    response_has_matrix_versions,
)
from mindroom.runtime_shutdown import (
    GENERIC_SHUTDOWN,
    SYNC_RESTART_SHUTDOWN,
    RestartReasonCategory,
    RuntimeLifecycleAction,
    RuntimeShutdownIntent,
    shutdown_intent_for_entity,
)
from mindroom.runtime_state import set_runtime_starting
from mindroom.startup_errors import PermanentStartupError

if TYPE_CHECKING:
    from collections.abc import Awaitable, Callable, Coroutine
    from types import TracebackType

    import structlog

    from mindroom.bot import AgentBot, TeamBot

logger = get_logger(__name__)

_MATRIX_HOMESERVER_STARTUP_TIMEOUT_ENV = "MINDROOM_MATRIX_HOMESERVER_STARTUP_TIMEOUT_SECONDS"
_MATRIX_HOMESERVER_REQUEST_TIMEOUT_SECONDS = 5.0
_MATRIX_HOMESERVER_RETRY_INTERVAL_SECONDS = 2.0
STARTUP_RETRY_INITIAL_DELAY_SECONDS = 2.0
STARTUP_RETRY_MAX_DELAY_SECONDS = 60.0
_CANCELLING_LOGGED_TASKS: set[asyncio.Task[Any]] = set()
_MATRIX_SYNC_WATCHDOG_POLL_INTERVAL_SECONDS = 5.0
_MATRIX_SYNC_STARTUP_TIMEOUT_ENV = "MINDROOM_MATRIX_SYNC_STARTUP_TIMEOUT_SECONDS"
_MATRIX_SYNC_CACHE_WRITE_GRACE_ENV = "MINDROOM_MATRIX_SYNC_CACHE_WRITE_GRACE_SECONDS"
_STALLED_RESTART_MAX_JITTER_SECONDS = 10.0


def _stalled_restart_jitter_seconds() -> float:
    """Return random extra delay for restarting one stalled sync loop.

    A loop-wide stall trips every agent's watchdog in the same tick; restarting
    all sync loops simultaneously triggers a thundering herd of initial syncs
    that re-starves the loop and produces repeating stall/restart waves.
    """
    return random.uniform(0.0, _STALLED_RESTART_MAX_JITTER_SECONDS)  # noqa: S311


__all__ = [
    "STARTUP_RETRY_INITIAL_DELAY_SECONDS",
    "STARTUP_RETRY_MAX_DELAY_SECONDS",
    "SYNC_RESTART_CANCEL_MSG",
    "USER_STOP_CANCEL_MSG",
    "CancelSource",
    "EntityStartResults",
    "cancel_failure_reason",
    "cancel_logged_task",
    "cancel_source_from_failure_reason",
    "cancel_sync_task",
    "cancel_task",
    "classify_cancel_source",
    "create_logged_task",
    "is_permanent_startup_error",
    "is_sync_restart_cancel",
    "log_cancelled_response",
    "log_cancelled_response_source",
    "log_startup_phase_finished",
    "log_startup_phase_started",
    "matrix_sync_cache_write_grace_seconds",
    "matrix_sync_startup_timeout_seconds",
    "request_task_cancel",
    "retry_delay_seconds",
    "run_with_retry",
    "stop_entities",
    "sync_forever_with_restart",
    "wait_for_matrix_homeserver",
]


def is_sync_restart_cancel(exc: asyncio.CancelledError) -> bool:
    """Return whether one cancellation was caused by a sync restart."""
    return classify_cancel_source(exc) == "sync_restart"


def log_cancelled_response(
    logger: structlog.stdlib.BoundLogger,
    *,
    exc: asyncio.CancelledError,
    message_id: str | None,
    restart_message: str,
    user_stop_message: str,
    interrupted_message: str,
) -> None:
    """Log one CancelledError with the right provenance label."""
    log_cancelled_response_source(
        logger,
        cancel_source=classify_cancel_source(exc),
        message_id=message_id,
        restart_message=restart_message,
        user_stop_message=user_stop_message,
        interrupted_message=interrupted_message,
        exc_info=(type(exc), exc, exc.__traceback__),
    )


def log_cancelled_response_source(
    logger: structlog.stdlib.BoundLogger,
    *,
    cancel_source: CancelSource,
    message_id: str | None,
    restart_message: str,
    user_stop_message: str,
    interrupted_message: str,
    exc_info: tuple[type[BaseException], BaseException, TracebackType | None] | bool | None = None,
) -> None:
    """Log one resolved cancellation source with caller-specific text."""
    if cancel_source == "sync_restart":
        logger.info(restart_message, message_id=message_id)
    elif cancel_source == "user_stop":
        logger.info(user_stop_message, message_id=message_id)
    else:
        kwargs: dict[str, Any] = {"message_id": message_id}
        if exc_info is not None:
            kwargs["exc_info"] = exc_info
        logger.warning(interrupted_message, **kwargs)


def matrix_sync_startup_timeout_seconds(runtime_paths: RuntimePaths) -> float:
    """Return the sync startup timeout resolved via ``RuntimePaths``."""
    raw = (runtime_paths.env_value(_MATRIX_SYNC_STARTUP_TIMEOUT_ENV) or "").strip()
    if not raw:
        return MATRIX_SYNC_STARTUP_GRACE_SECONDS
    value = float(raw)
    if value <= 0:
        msg = f"{_MATRIX_SYNC_STARTUP_TIMEOUT_ENV} must be a positive number"
        raise ValueError(msg)
    return value


def matrix_sync_cache_write_grace_seconds(runtime_paths: RuntimePaths) -> float:
    """Return the finite grace for one active durable sync-cache phase."""
    raw = (runtime_paths.env_value(_MATRIX_SYNC_CACHE_WRITE_GRACE_ENV) or "").strip()
    if not raw:
        return MATRIX_SYNC_CACHE_WRITE_GRACE_SECONDS
    try:
        value = float(raw)
    except ValueError:
        msg = f"{_MATRIX_SYNC_CACHE_WRITE_GRACE_ENV} must be a finite positive number"
        raise ValueError(msg) from None
    if not math.isfinite(value) or value <= 0:
        msg = f"{_MATRIX_SYNC_CACHE_WRITE_GRACE_ENV} must be a finite positive number"
        raise ValueError(msg)
    return value


def _matrix_homeserver_startup_timeout_seconds_from_env(
    runtime_paths: RuntimePaths,
) -> int | None:
    """Return the startup wait timeout from the environment, if configured."""
    raw_timeout = (runtime_paths.env_value(_MATRIX_HOMESERVER_STARTUP_TIMEOUT_ENV) or "").strip()
    if not raw_timeout:
        return None
    timeout_seconds = int(raw_timeout)
    if timeout_seconds == 0:
        return None
    if timeout_seconds < 0:
        msg = f"{_MATRIX_HOMESERVER_STARTUP_TIMEOUT_ENV} must be 0 or a positive integer"
        raise ValueError(msg)
    return timeout_seconds


def retry_delay_seconds(
    attempt: int,
    *,
    initial_delay_seconds: float,
    max_delay_seconds: float,
) -> float:
    """Return capped exponential backoff delay for a retry attempt."""
    exponent = max(0, attempt - 1)
    return min(max_delay_seconds, initial_delay_seconds * (2**exponent))


def is_permanent_startup_error(exc: Exception) -> bool:
    """Return whether a startup exception is clearly non-retryable."""
    return isinstance(exc, PermanentStartupError)


async def cancel_task(
    task: asyncio.Task | None,
    *,
    suppress_exceptions: tuple[type[BaseException], ...] = (asyncio.CancelledError,),
    shutdown_intent: RuntimeShutdownIntent = GENERIC_SHUTDOWN,
) -> None:
    """Cancel a detached task and wait for it to finish."""
    if task is None:
        return
    request_task_cancel(task, cancel_source=shutdown_intent.cancel_source)
    with suppress(*suppress_exceptions):
        await task


async def cancel_logged_task(task: asyncio.Task | None) -> None:
    """Cancel a detached logged task without re-raising its completion exception."""
    if task is None:
        return
    _CANCELLING_LOGGED_TASKS.add(task)
    try:
        await cancel_task(task)
    except Exception:
        # `_log_detached_task_result` downgrades these cancellation-time failures to
        # debug logging, so shutdown/reload should not re-raise them here.
        return


class _MatrixSyncStalledError(RuntimeError):
    """Raised when the watchdog detects a stalled Matrix sync loop."""


@dataclass(slots=True)
class _SyncIteration:
    """Own the lifecycle of one sync task and its watchdog."""

    bot: AgentBot | TeamBot
    sync_task: asyncio.Task[Any] | None
    watchdog_task: asyncio.Task[Any] | None
    watchdog_cancelled_sync: asyncio.Event = field(default_factory=asyncio.Event)

    @staticmethod
    async def _watch(
        bot: AgentBot | TeamBot,
        sync_task: asyncio.Task[Any],
        watchdog_cancelled_sync: asyncio.Event,
    ) -> None:
        """Cancel a sync task when it stops reporting successful sync responses.

        Before the first ``SyncResponse`` (or ``SyncError``) arrives, the monotonic
        watchdog clock is not armed (``seconds_since_last_sync_activity`` returns
        ``None``). During that startup window a separate, longer timeout protects
        against a first sync that never completes.
        """
        startup_timeout_seconds = matrix_sync_startup_timeout_seconds(bot.runtime_paths)
        cache_write_grace_seconds = matrix_sync_cache_write_grace_seconds(bot.runtime_paths)
        startup_monotonic = time.monotonic()
        while bot.running and not sync_task.done():
            await asyncio.sleep(_MATRIX_SYNC_WATCHDOG_POLL_INTERVAL_SECONDS)
            sync_age_seconds = bot.seconds_since_last_sync_activity()

            if sync_age_seconds is None:
                # Still waiting for the first SyncResponse/SyncError.
                elapsed = time.monotonic() - startup_monotonic
                if elapsed <= startup_timeout_seconds:
                    continue
                logger.error(
                    "matrix_sync_watchdog_stalled",
                    agent=bot.agent_name,
                    active_response_count=bot.in_flight_response_count,
                    elapsed_seconds=elapsed,
                    restart_reason_category="first_sync_timeout",
                    resulting_action="cancel_receive_loop",
                    startup_timeout_seconds=startup_timeout_seconds,
                )
            elif sync_age_seconds <= MATRIX_SYNC_WATCHDOG_TIMEOUT_SECONDS:
                continue
            else:
                cache_write_progress = bot.sync_cache_write_progress()
                now_monotonic = time.monotonic()
                cache_write_seconds = (
                    cache_write_progress.seconds_in_flight(now_monotonic) if cache_write_progress is not None else None
                )
                if cache_write_seconds is not None and cache_write_seconds <= cache_write_grace_seconds:
                    logger.debug(
                        "matrix_sync_watchdog_awaiting_cache_write",
                        agent=bot.agent_name,
                        active_response_count=bot.in_flight_response_count,
                        cache_write_grace_seconds=cache_write_grace_seconds,
                        cache_write_seconds_in_flight=cache_write_seconds,
                        resulting_action="await_sync_cache_write",
                        stale_for_seconds=sync_age_seconds,
                    )
                    continue
                if cache_write_seconds is not None:
                    logger.error(
                        "matrix_sync_watchdog_stalled",
                        agent=bot.agent_name,
                        active_response_count=bot.in_flight_response_count,
                        cache_write_grace_seconds=cache_write_grace_seconds,
                        cache_write_seconds_in_flight=cache_write_seconds,
                        last_sync_time=bot.last_sync_time.isoformat() if bot.last_sync_time is not None else None,
                        restart_reason_category="cache_write_grace_exhausted",
                        resulting_action="cancel_receive_loop",
                        stale_for_seconds=sync_age_seconds,
                    )
                else:
                    logger.error(
                        "matrix_sync_watchdog_stalled",
                        agent=bot.agent_name,
                        active_response_count=bot.in_flight_response_count,
                        last_sync_time=bot.last_sync_time.isoformat() if bot.last_sync_time is not None else None,
                        restart_reason_category="sync_activity_timeout",
                        resulting_action="cancel_receive_loop",
                        stale_for_seconds=sync_age_seconds,
                    )

            watchdog_cancelled_sync.set()
            request_task_cancel(sync_task, cancel_source="sync_restart")
            with suppress(asyncio.CancelledError):
                await sync_task
            msg = f"Matrix sync loop stalled for {bot.agent_name}"
            raise _MatrixSyncStalledError(msg)

    @classmethod
    def start(cls, bot: AgentBot | TeamBot) -> _SyncIteration:
        """Create the sync task and its watchdog for one loop iteration."""
        bot.mark_sync_loop_started()
        # Reset the monotonic watchdog clock so a fresh iteration gets the full
        # startup timeout instead of inheriting a stale timestamp from a previous
        # stall/restart cycle.
        bot.reset_watchdog_clock()
        watchdog_cancelled_sync = asyncio.Event()
        sync_task = asyncio.create_task(bot.sync_forever(), name=f"matrix_sync_{bot.agent_name}")
        watchdog_coro = cls._watch(bot, sync_task, watchdog_cancelled_sync)
        try:
            watchdog_task = asyncio.create_task(
                watchdog_coro,
                name=f"matrix_sync_watchdog_{bot.agent_name}",
            )
        except BaseException:
            watchdog_coro.close()
            sync_task.cancel()
            raise
        return cls(
            bot=bot,
            sync_task=sync_task,
            watchdog_task=watchdog_task,
            watchdog_cancelled_sync=watchdog_cancelled_sync,
        )

    async def wait(self) -> None:
        """Wait for the first task to finish and surface the real failure."""
        if self.sync_task is None or self.watchdog_task is None:
            return
        done, _ = await asyncio.wait(
            {self.sync_task, self.watchdog_task},
            return_when=asyncio.FIRST_COMPLETED,
        )
        if self.sync_task in done:
            if not self.sync_task.cancelled():
                await self.sync_task  # raises if sync_forever failed; returns if clean
                return
            # Let the watchdog surface its stalled-sync error when it initiated the
            # cancellation; otherwise preserve the original sync-task cancellation.
            if self.watchdog_cancelled_sync.is_set():
                await self.watchdog_task
            await self.sync_task
            return
        await self.watchdog_task

    async def cancel(self, *, shutdown_intent: RuntimeShutdownIntent = GENERIC_SHUTDOWN) -> None:
        """Cancel child tasks without masking the original failure."""
        for attr in ("watchdog_task", "sync_task"):
            task = getattr(self, attr)
            if task is None:
                continue
            setattr(self, attr, None)
            if attr == "sync_task":
                request_task_cancel(task, cancel_source=shutdown_intent.cancel_source)
            else:
                task.cancel()
            try:
                await task
            except (asyncio.CancelledError, _MatrixSyncStalledError):
                pass
            except Exception:
                logger.warning("sync_iteration_cleanup_failed", agent=self.bot.agent_name, exc_info=True)


@dataclass(frozen=True, slots=True)
class _SyncIterationOutcome:
    """Why one receive-loop iteration ended, and how its runtime must be settled."""

    # None means the supervisor is done with this bot: it either stopped or was
    # cancelled, so no replacement receive loop should start.
    restart_reason_category: RestartReasonCategory | None = None
    stalled_restart: bool = False
    shutdown_intent: RuntimeShutdownIntent = GENERIC_SHUTDOWN


async def _run_sync_iteration(bot: AgentBot | TeamBot, *, attempt: int) -> _SyncIterationOutcome:
    """Run one receive-loop iteration and settle only the tasks that iteration owns.

    Live responses outlive a transport restart: the response runtime is torn down
    only once the bot itself has stopped, which is a config, agent, or process
    shutdown driven by the caller rather than a restart of the receive loop.
    """
    iteration: _SyncIteration | None = None
    outcome = _SyncIterationOutcome()
    try:
        logger.info("starting_sync_loop", agent=bot.agent_name)
        iteration = _SyncIteration.start(bot)
        await iteration.wait()
        if bot.running:
            logger.warning("sync_loop_returned_while_bot_running", agent=bot.agent_name, retry_count=attempt)
            outcome = _SyncIterationOutcome(restart_reason_category="unexpected_sync_return")
    except asyncio.CancelledError as exc:
        logger.info("sync_task_cancelled", agent=bot.agent_name)
        if is_sync_restart_cancel(exc):
            # A replacement runtime is taking over, so its predecessor's responses
            # must carry sync-restart provenance into the drain below.
            outcome = _SyncIterationOutcome(shutdown_intent=SYNC_RESTART_SHUTDOWN)
    except _MatrixSyncStalledError:
        outcome = _SyncIterationOutcome(restart_reason_category="watchdog_stall", stalled_restart=True)
    except Exception:
        logger.exception("sync_loop_failed", agent=bot.agent_name, retry_count=attempt)
        outcome = _SyncIterationOutcome(restart_reason_category="sync_failure")
    finally:
        if iteration is not None:
            await iteration.cancel(shutdown_intent=outcome.shutdown_intent)
            if not bot.running:
                await bot.prepare_for_sync_shutdown(shutdown_intent=outcome.shutdown_intent)
    return outcome


def _log_detached_task_result(task: asyncio.Task, *, message: str) -> None:
    """Log failures from a detached background task."""
    try:
        task.result()
    except asyncio.CancelledError:
        return
    except Exception:
        if task in _CANCELLING_LOGGED_TASKS:
            logger.debug(
                "Detached task failed while being cancelled",
                task_name=task.get_name(),
                exc_info=True,
            )
            return
        logger.exception(message)
    finally:
        _CANCELLING_LOGGED_TASKS.discard(task)


def create_logged_task(
    coro: Coroutine[Any, Any, None],
    *,
    name: str,
    failure_message: str,
) -> asyncio.Task:
    """Create a detached task that logs failures on completion."""
    task = asyncio.create_task(coro, name=name)
    task.add_done_callback(partial(_log_detached_task_result, message=failure_message))
    return task


def log_startup_phase_started(phase: str) -> float:
    """Log and time one startup phase."""
    logger.info("startup_phase_started", phase=phase)
    return time.monotonic()


def log_startup_phase_finished(phase: str, started_at: float, *, status: str = "completed") -> None:
    """Log elapsed time for one startup phase."""
    logger.info(
        "startup_phase_finished",
        phase=phase,
        status=status,
        elapsed_ms=round((time.monotonic() - started_at) * 1000, 1),
    )


async def run_with_retry(
    step_name: str,
    operation: Callable[[], Awaitable[None]],
    *,
    initial_delay_seconds: float = STARTUP_RETRY_INITIAL_DELAY_SECONDS,
    max_delay_seconds: float = STARTUP_RETRY_MAX_DELAY_SECONDS,
    permanent_error_check: Callable[[Exception], bool] | None = None,
    update_runtime_state: bool = True,
) -> None:
    """Run an async startup step until it succeeds or a permanent error occurs."""
    attempt = 0
    while True:
        try:
            await operation()
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            if permanent_error_check is not None and permanent_error_check(exc):
                logger.error("startup_step_failed_permanently", step_name=step_name, error=str(exc))  # noqa: TRY400
                raise
            attempt += 1
            retry_in_seconds = retry_delay_seconds(
                attempt,
                initial_delay_seconds=initial_delay_seconds,
                max_delay_seconds=max_delay_seconds,
            )
            logger.warning(
                "startup_step_retrying",
                step_name=step_name,
                attempt=attempt,
                retry_in_seconds=retry_in_seconds,
                exc_info=True,
            )
            if update_runtime_state:
                set_runtime_starting(f"{step_name} failed; retrying in {retry_in_seconds:.0f}s")
            await asyncio.sleep(retry_in_seconds)
        else:
            return


async def wait_for_matrix_homeserver(
    *,
    runtime_paths: RuntimePaths,
    timeout_seconds: float | None = None,
    request_timeout_seconds: float = _MATRIX_HOMESERVER_REQUEST_TIMEOUT_SECONDS,
    retry_interval_seconds: float = _MATRIX_HOMESERVER_RETRY_INTERVAL_SECONDS,
) -> None:
    """Wait for the configured Matrix homeserver to answer `/versions`."""
    if timeout_seconds is None:
        timeout_seconds = _matrix_homeserver_startup_timeout_seconds_from_env(runtime_paths)
    versions_url = matrix_versions_url(constants.runtime_matrix_homeserver(runtime_paths=runtime_paths))
    set_runtime_starting(f"Waiting for Matrix homeserver at {versions_url}")
    loop = asyncio.get_running_loop()
    deadline = None if timeout_seconds is None else loop.time() + timeout_seconds
    attempt = 0
    logger.info(
        "Waiting for Matrix homeserver",
        url=versions_url,
        timeout_seconds=timeout_seconds,
    )

    async with httpx.AsyncClient(
        timeout=request_timeout_seconds,
        verify=runtime_matrix_ssl_verify(runtime_paths=runtime_paths),
    ) as client:
        while deadline is None or loop.time() < deadline:
            attempt += 1
            try:
                response = await client.get(versions_url)
            except httpx.TransportError as exc:
                if attempt == 1 or attempt % 5 == 0:
                    logger.info(
                        "Matrix homeserver not ready yet",
                        url=versions_url,
                        attempt=attempt,
                        error=str(exc),
                    )
                await asyncio.sleep(retry_interval_seconds)
                continue

            if response_has_matrix_versions(response):
                logger.info("Matrix homeserver ready", url=versions_url)
                return

            if attempt == 1 or attempt % 5 == 0:
                logger.info(
                    "Matrix homeserver not ready yet",
                    url=versions_url,
                    attempt=attempt,
                    status_code=response.status_code,
                    body_preview=response.text[:200].replace("\n", " "),
                )
            await asyncio.sleep(retry_interval_seconds)

    msg = f"Timed out waiting for Matrix homeserver at {versions_url}"
    raise TimeoutError(msg)


@dataclass(slots=True)
class EntityStartResults:
    """Result of one pass trying to start a batch of entities."""

    started_bots: list[AgentBot | TeamBot] = field(default_factory=list)
    retryable_entities: list[str] = field(default_factory=list)
    permanently_failed_entities: list[str] = field(default_factory=list)


async def cancel_sync_task(
    entity_name: str,
    sync_tasks: dict[str, asyncio.Task],
    *,
    shutdown_intent: RuntimeShutdownIntent = GENERIC_SHUTDOWN,
) -> None:
    """Cancel and remove a sync task for an entity."""
    task = sync_tasks.pop(entity_name, None)
    await cancel_task(task, shutdown_intent=shutdown_intent)


async def stop_entities(
    entities_to_stop: set[str],
    agent_bots: dict[str, AgentBot | TeamBot],
    sync_tasks: dict[str, asyncio.Task],
    *,
    restart_entities: set[str] | None = None,
) -> None:
    """Stop a set of entities and remove them from runtime maps."""
    restart_entities = set() if restart_entities is None else restart_entities
    shutdown_intents = {
        entity_name: shutdown_intent_for_entity(entity_name, restart_entities=restart_entities)
        for entity_name in entities_to_stop
    }
    # Stop sync loops before certifying callback drains; otherwise fresh callbacks can
    # appear after the checkpoint decision.
    for entity_name in entities_to_stop:
        await cancel_sync_task(
            entity_name,
            sync_tasks,
            shutdown_intent=shutdown_intents[entity_name],
        )

    for entity_name in entities_to_stop:
        bot = agent_bots.get(entity_name)
        if bot is not None:
            await bot.prepare_for_sync_shutdown(shutdown_intent=shutdown_intents[entity_name])

    stop_tasks = [
        agent_bots[entity_name].stop(shutdown_intent=shutdown_intents[entity_name])
        for entity_name in entities_to_stop
        if entity_name in agent_bots
    ]
    if stop_tasks:
        await asyncio.gather(*stop_tasks)

    for entity_name in entities_to_stop:
        agent_bots.pop(entity_name, None)


async def sync_forever_with_restart(bot: AgentBot | TeamBot, max_retries: int = -1) -> None:
    """Run sync_forever with automatic restart on failure."""
    retry_count = 0
    while bot.running and (max_retries < 0 or retry_count < max_retries):
        outcome = await _run_sync_iteration(bot, attempt=retry_count + 1)
        if outcome.restart_reason_category is None or not bot.running:
            break
        retry_count += 1
        will_retry = max_retries < 0 or retry_count < max_retries
        resulting_action: RuntimeLifecycleAction = "restart_receive_loop" if will_retry else "preserve_response_runtime"
        logger.warning(
            "matrix_sync_transport_restart",
            agent=bot.agent_name,
            active_response_count=bot.in_flight_response_count,
            restart_reason_category=outcome.restart_reason_category,
            resulting_action=resulting_action,
            retry_count=retry_count,
            will_retry=will_retry,
        )
        if not will_retry:
            # The entity keeps its live responses but has no receive loop left, so
            # its stale sync clock surfaces it through the Matrix sync health snapshot.
            logger.error(
                "sync_loop_retries_exhausted",
                agent=bot.agent_name,
                retry_count=retry_count,
                max_retries=max_retries,
                restart_reason_category=outcome.restart_reason_category,
            )
            break

        wait_time = retry_delay_seconds(
            retry_count,
            initial_delay_seconds=5.0,
            max_delay_seconds=60.0,
        )
        if outcome.stalled_restart:
            wait_time += _stalled_restart_jitter_seconds()
        logger.info("restarting_sync_loop", agent=bot.agent_name, retry_count=retry_count, wait_seconds=wait_time)
        await asyncio.sleep(wait_time)
