connector-runner: heartbeat attribution converges after restart + receive-liveness health surface + image HEALTHCHECKs (#2153, carved out of #2208)

connectors/echo-http/Dockerfile

@@ -31,4 +31,5 @@ RUN uv pip install --system --no-cache --no-sources \
     /app/packages/aios-connector-http \
     /app/connectors/echo-http
 
+HEALTHCHECK --interval=10s --timeout=5s --retries=3 CMD ["python", "-m", "aios_connector_http.healthcheck"]
 CMD ["python", "-m", "aios_echo_http"]

connectors/matrix/aios_matrix/connector.py

@@ -225,6 +225,9 @@ async def serve_connection(self, connection_id: str, secrets: dict[str, str]) ->
         intent = self.az.intent.user(self._mxid(localpart))
         await intent.ensure_registered()
         await self._reconcile_intent(intent)
+        # The appservice listener is started in setup; registration, routing,
+        # and initial membership reconciliation make this ghost receivable.
+        self.mark_transport_ready(connection_id)
         try:
             await asyncio.Event().wait()
         finally:

connectors/matrix/healthcheck.py

@@ -4,6 +4,13 @@
 import os
 import urllib.request
 
+from aios_connector_http.healthcheck import main as check_connector_heartbeat
+
+# Matrix's appservice endpoint can remain responsive while every connection
+# worker is starting or restarting.  Require the SDK's serving heartbeat first
+# so Docker health reflects inbound transport readiness as well as HTTP reachability.
+check_connector_heartbeat()
+
 host, _, port = os.environ.get("MATRIX_LISTEN_ADDR", "0.0.0.0:29328").rpartition(":")
 if host in {"", "0.0.0.0", "::"}:
     host = "127.0.0.1"

connectors/signal/Dockerfile

@@ -146,4 +146,5 @@ RUN uv pip install --system --no-cache --no-sources \
 # and connector containers run as root so the chown succeeds; the actual
 # signal-cli process inside aios_signal still runs as root.
 RUN chmod 0755 /app/connectors/signal/entrypoint.sh
+HEALTHCHECK --interval=10s --timeout=5s --retries=3 CMD ["python", "-m", "aios_connector_http.healthcheck"]
 ENTRYPOINT ["/app/connectors/signal/entrypoint.sh"]

connectors/signal/src/aios_signal/connector.py

@@ -206,6 +206,9 @@ async def serve_connection(self, connection_id: str, secrets: dict[str, str]) ->
         )
         self.state[connection_id] = state
         queue = self._queue_for(phone)
+        # The shared daemon dispatcher is already live; once this account's
+        # queue and state exist, inbound envelopes can be routed and drained.
+        self.mark_transport_ready(connection_id)
         log.info(
             "signal.connection.ready",
             connection_id=connection_id,

connectors/slack/Dockerfile

@@ -33,4 +33,5 @@ RUN uv pip install --system --no-cache --no-sources \
     /app/packages/aios-connector-http \
     /app/connectors/slack
 
+HEALTHCHECK --interval=10s --timeout=5s --retries=3 CMD ["python", "-m", "aios_connector_http.healthcheck"]
 CMD ["python", "-m", "aios_slack"]

connectors/slack/src/aios_slack/connector.py

@@ -193,7 +193,7 @@ async def serve_connection(self, connection_id: str, secrets: dict[str, str]) ->
 
             async with asyncio.TaskGroup() as tg:
                 tg.create_task(
-                    self._run_socket(state),
+                    self._run_socket(connection_id, state),
                     name=f"slack-socket-{connection_id}",
                 )
                 tg.create_task(
@@ -242,7 +242,7 @@ async def on_request(client: AsyncSocketModeClient, req: SocketModeRequest) -> N
 
         state.socket_client.socket_mode_request_listeners.append(on_request)
 
-    async def _run_socket(self, state: _SlackConnectionState) -> None:
+    async def _run_socket(self, connection_id: str, state: _SlackConnectionState) -> None:
         """Open the Socket-Mode connection and keep the task alive.
 
         ``connect()`` establishes the WebSocket and returns once the
@@ -252,6 +252,9 @@ async def _run_socket(self, state: _SlackConnectionState) -> None:
         into ``serve_connection``'s ``finally`` which closes the client.
         """
         await state.socket_client.connect()
+        # ``connect`` returns only after Socket Mode's receiver is running.
+        # Keep the heartbeat fail-closed until that concrete receive path exists.
+        self.mark_transport_ready(connection_id)
         await asyncio.Event().wait()
 
     async def _drain_queue(self, connection_id: str, state: _SlackConnectionState) -> None:

connectors/slack/tests/test_serve_connection.py

@@ -151,6 +151,7 @@ async def test_identity_match_serves_and_registers_listener(
         assert state.bot_user_id == BOT_USER_ID
         assert len(socket.socket_mode_request_listeners) == 1
         socket.connect.assert_awaited()
+        assert connector._connections[CONNECTION_ID].serve_status == "serving"
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):

connectors/sms/Dockerfile

@@ -39,4 +39,5 @@ RUN uv pip install --system --no-cache --no-sources \
 # Default listener port (overridable via AIOS_SMS_PORT).
 EXPOSE 8080
 
+HEALTHCHECK --interval=10s --timeout=5s --retries=3 CMD ["python", "-m", "aios_connector_http.healthcheck"]
 CMD ["python", "-m", "aios_sms"]

connectors/sms/src/aios_sms/connector.py

@@ -126,6 +126,10 @@ async def serve_connection(self, connection_id: str, secrets: dict[str, str]) ->
                 queue=state.inbound_queue,
             ),
         )
+        # Registration is the point at which the already-running webhook can
+        # route an inbound request to this connection.  Do not advertise the
+        # connection before that concrete receive path exists.
+        self.mark_transport_ready(connection_id)
         log.info(
             "sms.connection.ready",
             connection_id=connection_id,

connectors/telegram/Dockerfile

@@ -32,4 +32,5 @@ RUN uv pip install --system --no-cache --no-sources \
     /app/packages/aios-connector-http \
     /app/connectors/telegram
 
+HEALTHCHECK --interval=10s --timeout=5s --retries=3 CMD ["python", "-m", "aios_connector_http.healthcheck"]
 CMD ["python", "-m", "aios_telegram"]

connectors/telegram/src/aios_telegram/connector.py

@@ -126,7 +126,7 @@ async def serve_connection(self, connection_id: str, secrets: dict[str, str]) ->
         try:
             async with asyncio.TaskGroup() as tg:
                 tg.create_task(
-                    self._run_polling(state),
+                    self._run_polling(connection_id, state),
                     name=f"telegram-polling-{connection_id}",
                 )
                 tg.create_task(
@@ -203,14 +203,17 @@ async def _shutdown_application(application: Application) -> None:  # type: igno
         with contextlib.suppress(Exception):
             await application.shutdown()
 
-    async def _run_polling(self, state: _TelegramConnectionState) -> None:
+    async def _run_polling(self, connection_id: str, state: _TelegramConnectionState) -> None:
         await state.application.start()
         assert state.application.updater is not None
         # ``allowed_updates`` is opt-in — Telegram only delivers update
         # types we explicitly subscribe to.  Without this, edits and
         # reactions never reach the bot regardless of which handlers we
         # register locally.
         await state.application.updater.start_polling(allowed_updates=_ALLOWED_UPDATES)
+        # PTB returns after its polling task has started and can receive updates.
+        # Do not advertise this connection before that transport startup succeeds.
+        self.mark_transport_ready(connection_id)
         await asyncio.Event().wait()
 
     async def _drain_queue(self, connection_id: str, state: _TelegramConnectionState) -> None:

connectors/telegram/tests/test_transport_readiness.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from aios_connector_http.runner import _ConnectionState
+
+from aios_telegram.connector import TelegramConnector, _TelegramConnectionState
+
+CONNECTION_ID = "conn_ready"
+
+
+def _connector_and_state() -> tuple[TelegramConnector, _TelegramConnectionState]:
+    connector = TelegramConnector()
+    connector._connections[CONNECTION_ID] = _ConnectionState(CONNECTION_ID, "bot")
+    updater = MagicMock()
+    updater.start_polling = AsyncMock()
+    application = MagicMock()
+    application.start = AsyncMock()
+    application.updater = updater
+    state = _TelegramConnectionState(
+        application=application,
+        bot_id=1,
+        first_name="Bot",
+        username="bot",
+        inbound_queue=asyncio.Queue(),
+    )
+    return connector, state
+
+
+async def test_polling_marks_ready_only_after_transport_starts() -> None:
+    connector, state = _connector_and_state()
+    release = asyncio.Event()
+
+    async def start_polling(**_kwargs: object) -> None:
+        await release.wait()
+
+    state.application.updater.start_polling.side_effect = start_polling
+    task = asyncio.create_task(connector._run_polling(CONNECTION_ID, state))
+    await asyncio.sleep(0)
+    assert connector._connections[CONNECTION_ID].serve_status == "starting"
+
+    release.set()
+    await asyncio.sleep(0)
+    assert connector._connections[CONNECTION_ID].serve_status == "serving"
+
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+
+async def test_polling_start_failure_remains_unhealthy() -> None:
+    connector, state = _connector_and_state()
+    state.application.updater.start_polling.side_effect = RuntimeError("polling failed")
+
+    with pytest.raises(RuntimeError, match="polling failed"):
+        await connector._run_polling(CONNECTION_ID, state)
+
+    assert connector._connections[CONNECTION_ID].serve_status == "starting"

connectors/whatsapp/Dockerfile

@@ -77,4 +77,5 @@ RUN uv pip install --system --no-cache --no-sources \
 # runs as root inside this container.
 COPY connectors/whatsapp/entrypoint.sh /app/connectors/whatsapp/entrypoint.sh
 RUN chmod 0755 /app/connectors/whatsapp/entrypoint.sh
+HEALTHCHECK --interval=10s --timeout=5s --retries=3 CMD ["python", "-m", "aios_connector_http.healthcheck"]
 ENTRYPOINT ["/app/connectors/whatsapp/entrypoint.sh"]

connectors/whatsapp/src/aios_whatsapp/connector.py

@@ -62,6 +62,9 @@ async def serve_connection(self, connection_id: str, secrets: dict[str, str]) ->
             store_dir=store_dir,
         ) as daemon:
             self.state[connection_id] = _WhatsappConnectionState(phone=phone, daemon=daemon)
+            # Entering the daemon context establishes its notification
+            # listener; state publication completes the inbound receive path.
+            self.mark_transport_ready(connection_id)
             log.info(
                 "whatsapp.connection.ready",
                 connection_id=connection_id,

packages/aios-connector-http/aios_connector_http/healthcheck.py

@@ -0,0 +1,92 @@
+"""Container health probe for connectors built on :mod:`aios_connector_http`."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import time
+from pathlib import Path
+
+DEFAULT_HEARTBEAT_PATH = Path("/var/run/aios-connector-alive")
+DEFAULT_MAX_AGE_SECONDS = 30.0
+
+
+def resolve_heartbeat_path() -> Path:
+    """Return the probe path, using a writable location outside containers."""
+    configured = os.environ.get("AIOS_CONNECTOR_HEARTBEAT_PATH")
+    if configured:
+        return Path(configured)
+    if Path("/.dockerenv").exists():
+        return DEFAULT_HEARTBEAT_PATH
+    return Path(os.environ.get("TMPDIR", tempfile.gettempdir())) / DEFAULT_HEARTBEAT_PATH.name
+
+
+def heartbeat_max_age_seconds() -> float:
+    """Return the age at which the container probe considers a heartbeat stale."""
+    return float(
+        os.environ.get("AIOS_CONNECTOR_HEARTBEAT_MAX_AGE_SECONDS", DEFAULT_MAX_AGE_SECONDS)
+    )
+
+
+def heartbeat_is_fresh(path: Path, *, max_age_seconds: float) -> bool:
+    """Return whether ``path`` was touched within ``max_age_seconds``."""
+    try:
+        age = time.time() - path.stat().st_mtime
+    except FileNotFoundError:
+        return False
+    return age <= max_age_seconds
+
+
+def _parse_connection_health(path: Path) -> tuple[list[str], list[str]] | None:
+    """Return validated connection state, or ``None`` for unreadable content."""
+    try:
+        payload = json.loads(path.read_text())
+    except (FileNotFoundError, UnicodeError, json.JSONDecodeError, OSError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+
+    healthy = payload.get("healthy_connection_ids")
+    unhealthy = payload.get("unhealthy_connection_ids")
+    if not isinstance(healthy, list) or not all(isinstance(value, str) for value in healthy):
+        return None
+    if not isinstance(unhealthy, list) or not all(isinstance(value, str) for value in unhealthy):
+        return None
+    return healthy, unhealthy
+
+
+def read_connection_health(path: Path) -> tuple[list[str], list[str]]:
+    """Read connection-correlated transport state from a heartbeat.
+
+    Invalid content remains represented as empty state for callers which only
+    consume attribution.  The health probe uses the validity-aware parser and
+    fails closed instead of confusing invalid content with valid empty state.
+    """
+    return _parse_connection_health(path) or ([], [])
+
+
+def main() -> None:
+    path = resolve_heartbeat_path()
+    max_age = heartbeat_max_age_seconds()
+    parsed = _parse_connection_health(path)
+    valid = parsed is not None
+    healthy, unhealthy = parsed or ([], [])
+    # Docker retains probe output in State.Health.Log.  The external reader
+    # consumes this machine-readable line to attribute container health to the
+    # affected connection rather than its healthy siblings.
+    print(
+        json.dumps(
+            {
+                "healthy_connection_ids": healthy,
+                "unhealthy_connection_ids": unhealthy,
+            },
+            sort_keys=True,
+        )
+    )
+    if not valid or unhealthy or not heartbeat_is_fresh(path, max_age_seconds=max_age):
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()