feat(channels): add health telemetry and active connection probe to E…

src/agentos/channels/email.py

@@ -291,6 +291,8 @@ class EmailChannel:
     _connected: bool = field(default=False, init=False, repr=False)
     _last_message_at: datetime | None = field(default=None, init=False, repr=False)
     _last_error: str = field(default="", init=False, repr=False)
+    _last_poll_at: datetime | None = field(default=None, init=False, repr=False)
+    _last_poll_count: int | None = field(default=None, init=False, repr=False)
 
     # ------------------------------------------------------------------
     # Capability declaration
@@ -404,13 +406,54 @@ async def stop(self) -> None:
         log.info("email.stopped", name=self.config.name)
 
     async def health_check(self) -> ChannelHealth:
+        extra: dict[str, Any] = {
+            "imap_folder": self.config.imap_folder,
+            "poll_interval_s": self.config.poll_interval_s,
+        }
+        if self._last_poll_at is not None:
+            extra["last_poll_at"] = self._last_poll_at.isoformat()
+        if self._last_poll_count is not None:
+            extra["last_poll_count"] = self._last_poll_count
+        if self._last_error:
+            extra["last_error"] = self._last_error
         return ChannelHealth(
             connected=self._connected,
             bot_user_id=self.config.from_address or None,
             last_message_at=self._last_message_at,
-            extra={"last_error": self._last_error} if self._last_error else {},
+            extra=extra,
         )
 
+    async def probe(self) -> dict[str, Any]:
+        """Perform active connectivity check against configured IMAP and SMTP endpoints."""
+
+        def _check_imap() -> str:
+            client = self._imap_connect()
+            try:
+                client.noop()
+                return "ok"
+            finally:
+                with contextlib.suppress(Exception):
+                    client.logout()
+
+        def _check_smtp() -> str:
+            with self._smtp_connect() as server:
+                server.noop()
+                return "ok"
+
+        imap_status = "unknown"
+        smtp_status = "unknown"
+        try:
+            imap_status = await asyncio.to_thread(_check_imap)
+        except Exception as exc:
+            imap_status = f"error: {exc}"
+
+        try:
+            smtp_status = await asyncio.to_thread(_check_smtp)
+        except Exception as exc:
+            smtp_status = f"error: {exc}"
+
+        return {"imap": imap_status, "smtp": smtp_status}
+
     # ------------------------------------------------------------------
     # Inbound
     # ------------------------------------------------------------------
@@ -428,6 +471,8 @@ async def _poll_loop(self) -> None:
             else:
                 self._connected = True
                 self._last_error = ""
+                self._last_poll_at = datetime.now(UTC)
+                self._last_poll_count = len(messages)
                 for message in messages:
                     self.enqueue(message)
             await asyncio.sleep(max(1.0, self.config.poll_interval_s))
@@ -670,9 +715,7 @@ def _compose(
         message.set_content(body or "")
         return message
 
-    def _smtp_send(self, message: EmailMessage) -> None:
-        """Blocking SMTP send — always called through ``asyncio.to_thread``."""
-
+    def _smtp_connect(self) -> smtplib.SMTP:
         context = ssl.create_default_context()
         timeout = self.config.connect_timeout_s
         if self.config.smtp_ssl:
@@ -684,11 +727,15 @@ def _smtp_send(self, message: EmailMessage) -> None:
             )
         else:
             server = smtplib.SMTP(self.config.smtp_host, self.config.smtp_port, timeout=timeout)
-        with server:
-            if not self.config.smtp_ssl and self.config.smtp_starttls:
-                server.starttls(context=context)
-            if self.config.smtp_username:
-                server.login(self.config.smtp_username, self.config.smtp_password)
+        if not self.config.smtp_ssl and self.config.smtp_starttls:
+            server.starttls(context=context)
+        if self.config.smtp_username:
+            server.login(self.config.smtp_username, self.config.smtp_password)
+        return server
+
+    def _smtp_send(self, message: EmailMessage) -> None:
+        """Blocking SMTP send — always called through ``asyncio.to_thread``."""
+        with self._smtp_connect() as server:
             server.send_message(message)
 
     async def send(self, message: OutgoingMessage) -> None:

tests/test_channels/test_email_channel.py

@@ -484,6 +484,68 @@ def _boom() -> list[IncomingMessage]:
     assert channel._task is None
 
 
+async def test_health_check_reports_poll_telemetry(monkeypatch: pytest.MonkeyPatch) -> None:
+    channel = EmailChannel(config=_config(poll_interval_s=1.0, imap_folder="INBOX"))
+    inbound = channel._to_incoming(_raw())
+    assert inbound is not None
+    monkeypatch.setattr(channel, "_fetch_unseen", lambda: [inbound])
+
+    await channel.start()
+    await asyncio.sleep(0.05)
+    health = await channel.health_check()
+    await channel.stop()
+
+    assert health.connected is True
+    assert health.extra["imap_folder"] == "INBOX"
+    assert health.extra["poll_interval_s"] == 1.0
+    assert health.extra["last_poll_count"] == 1
+    assert "last_poll_at" in health.extra
+
+
+async def test_probe_success(monkeypatch: pytest.MonkeyPatch) -> None:
+    channel = EmailChannel(config=_config())
+
+    class _MockIMAP:
+        def noop(self) -> tuple[str, list[bytes]]:
+            return "OK", [b""]
+
+        def logout(self) -> None:
+            pass
+
+    class _MockSMTP:
+        def __enter__(self) -> _MockSMTP:
+            return self
+
+        def __exit__(self, *args: object) -> None:
+            pass
+
+        def noop(self) -> tuple[int, bytes]:
+            return 250, b"OK"
+
+    monkeypatch.setattr(channel, "_imap_connect", lambda: _MockIMAP())
+    monkeypatch.setattr(channel, "_smtp_connect", lambda: _MockSMTP())
+
+    result = await channel.probe()
+    assert result == {"imap": "ok", "smtp": "ok"}
+
+
+async def test_probe_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+    channel = EmailChannel(config=_config())
+
+    def _fail_imap() -> Any:
+        raise OSError("connection refused")
+
+    def _fail_smtp() -> Any:
+        raise OSError("auth failed")
+
+    monkeypatch.setattr(channel, "_imap_connect", _fail_imap)
+    monkeypatch.setattr(channel, "_smtp_connect", _fail_smtp)
+
+    result = await channel.probe()
+    assert "connection refused" in result["imap"]
+    assert "auth failed" in result["smtp"]
+
+
 async def test_receive_yields_polled_messages(monkeypatch: pytest.MonkeyPatch) -> None:
     channel = EmailChannel(config=_config(poll_interval_s=1.0))
     inbound = channel._to_incoming(_raw())