diff --git a/EvoScientist/channels/email/__init__.py b/EvoScientist/channels/email/__init__.py index c48e91f..3e2e216 100644 --- a/EvoScientist/channels/email/__init__.py +++ b/EvoScientist/channels/email/__init__.py @@ -28,7 +28,7 @@ def create_from_config(config) -> EmailChannel: smtp_port=config.email_smtp_port, smtp_username=config.email_smtp_username, smtp_password=config.email_smtp_password, - smtp_use_tls=config.email_smtp_use_tls, + smtp_starttls=config.email_smtp_use_tls, from_address=config.email_from_address, poll_interval=config.email_poll_interval, mark_seen=config.email_mark_seen, diff --git a/EvoScientist/channels/email/channel.py b/EvoScientist/channels/email/channel.py index e3bb69c..1336a07 100644 --- a/EvoScientist/channels/email/channel.py +++ b/EvoScientist/channels/email/channel.py @@ -1,6 +1,7 @@ """Email channel implementation using IMAP + SMTP.""" import asyncio +import contextlib import email as email_lib import email.utils import html @@ -55,7 +56,7 @@ class EmailConfig(BaseChannelConfig): smtp_port: int = 587 smtp_username: str = "" smtp_password: str = "" - smtp_use_tls: bool = True + smtp_starttls: bool = True # True=STARTTLS (port 587), False=implicit SSL (port 465) from_address: str = "" poll_interval: int = 30 mark_seen: bool = True @@ -237,6 +238,37 @@ class EmailChannel(Channel, PollingMixin): def _is_ready(self) -> bool: return bool(self.config.smtp_host) + @contextlib.contextmanager + def _smtp_connect(self): + """Open an SMTP connection as a context manager. + + Ensures the connection is closed even if login or send raises. + Uses STARTTLS (port 587) when smtp_starttls=True, otherwise + implicit SSL (port 465). + """ + cfg = self.config + srv = None + try: + if cfg.smtp_starttls: + srv = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) + srv.starttls() + else: + srv = smtplib.SMTP_SSL( + cfg.smtp_host, cfg.smtp_port, + context=ssl.create_default_context(), timeout=30, + ) + srv.login(cfg.smtp_username, cfg.smtp_password) + yield srv + finally: + if srv is not None: + try: + srv.quit() + except Exception: + try: + srv.close() + except Exception: + pass + async def _send_chunk(self, chat_id, formatted_text, raw_text, reply_to, metadata): loop = asyncio.get_event_loop() try: @@ -268,17 +300,11 @@ class EmailChannel(Channel, PollingMixin): msg["References"] = f"{meta.get('references', '')} {orig_id}".strip() msg.set_content(content) try: - if cfg.smtp_use_tls: - srv = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) - srv.starttls() - else: - srv = smtplib.SMTP_SSL(cfg.smtp_host, cfg.smtp_port, context=ssl.create_default_context(), timeout=30) - srv.login(cfg.smtp_username, cfg.smtp_password) - srv.sendmail(from_addr, [to], msg.as_string()) - srv.quit() + with self._smtp_connect() as srv: + srv.sendmail(from_addr, [to], msg.as_string()) except Exception as e: - logger.error(f"SMTP send failed: from={from_addr} to={to} error={e}") - raise RuntimeError(f"SMTP: {e}") + logger.error(f"SMTP send failed: from={from_addr} to={to}") + raise RuntimeError("SMTP send failed") from e def _smtp_send_html(self, to: str, html_content: str, plain_content: str, meta: dict) -> None: """Send an email with both HTML and plain-text parts.""" @@ -297,17 +323,11 @@ class EmailChannel(Channel, PollingMixin): msg.attach(MIMEText(plain_content, "plain", "utf-8")) msg.attach(MIMEText(html_content, "html", "utf-8")) try: - if cfg.smtp_use_tls: - srv = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) - srv.starttls() - else: - srv = smtplib.SMTP_SSL(cfg.smtp_host, cfg.smtp_port, context=ssl.create_default_context(), timeout=30) - srv.login(cfg.smtp_username, cfg.smtp_password) - srv.sendmail(from_addr, [to], msg.as_string()) - srv.quit() + with self._smtp_connect() as srv: + srv.sendmail(from_addr, [to], msg.as_string()) except Exception as e: - logger.error(f"SMTP HTML send failed: from={from_addr} to={to} error={e}") - raise RuntimeError(f"SMTP HTML: {e}") + logger.error(f"SMTP HTML send failed: from={from_addr} to={to}") + raise RuntimeError("SMTP HTML send failed") from e # ── Media send (email attachment) ───────────────────────────── @@ -353,14 +373,8 @@ class EmailChannel(Channel, PollingMixin): msg.attach(part) try: - if cfg.smtp_use_tls: - srv = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) - srv.starttls() - else: - srv = smtplib.SMTP_SSL(cfg.smtp_host, cfg.smtp_port, context=ssl.create_default_context(), timeout=30) - srv.login(cfg.smtp_username, cfg.smtp_password) - srv.sendmail(from_addr, [to], msg.as_string()) - srv.quit() + with self._smtp_connect() as srv: + srv.sendmail(from_addr, [to], msg.as_string()) except Exception as e: - logger.error(f"SMTP attachment send failed: from={from_addr} to={to} error={e}") - raise RuntimeError(f"SMTP attachment: {e}") + logger.error(f"SMTP attachment send failed: from={from_addr} to={to}") + raise RuntimeError("SMTP attachment send failed") from e diff --git a/EvoScientist/channels/email/probe.py b/EvoScientist/channels/email/probe.py index 32e2fc9..735aa57 100644 --- a/EvoScientist/channels/email/probe.py +++ b/EvoScientist/channels/email/probe.py @@ -57,6 +57,7 @@ async def validate_email_smtp( loop = asyncio.get_event_loop() def _check(): + server = None try: if use_tls: server = smtplib.SMTP(host, port, timeout=10) @@ -65,11 +66,19 @@ async def validate_email_smtp( ctx = ssl.create_default_context() server = smtplib.SMTP_SSL(host, port, context=ctx, timeout=10) server.login(username, password) - server.quit() return True, "SMTP credentials valid" - except smtplib.SMTPAuthenticationError as e: - return False, f"SMTP auth failed: {e}" + except smtplib.SMTPAuthenticationError: + return False, "SMTP auth failed" except Exception as e: return False, f"SMTP error: {e}" + finally: + if server is not None: + try: + server.quit() + except Exception: + try: + server.close() + except Exception: + pass return await loop.run_in_executor(None, _check) diff --git a/EvoScientist/channels/signal/channel.py b/EvoScientist/channels/signal/channel.py index c43a34c..e21986b 100644 --- a/EvoScientist/channels/signal/channel.py +++ b/EvoScientist/channels/signal/channel.py @@ -40,6 +40,8 @@ class SignalChannel(Channel): self._writer: asyncio.StreamWriter | None = None self._rpc_id = 0 self._daemon_proc = None + # Pending RPC responses: rpc_id -> Future + self._pending_rpcs: dict[int, asyncio.Future] = {} # Cache message_id → sender for reaction targetAuthor (bounded) self._msg_senders: dict[str, str] = {} self._msg_senders_order: deque = deque(maxlen=200) @@ -51,8 +53,13 @@ class SignalChannel(Channel): # Try to start signal-cli daemon if not already running await self._ensure_daemon() - # Connect to JSON RPC socket - await self._connect() + try: + # Connect to JSON RPC socket + await self._connect() + except Exception: + # If connect fails after daemon was started, clean up the daemon + await self._cleanup() + raise self._running = True logger.info(f"Signal channel started (phone: {self.config.phone_number})") @@ -65,6 +72,11 @@ class SignalChannel(Channel): if hasattr(self, "_listen_task") and self._listen_task: self._listen_task.cancel() self._listen_task = None + # Cancel any pending RPC futures + for fut in self._pending_rpcs.values(): + if not fut.done(): + fut.cancel() + self._pending_rpcs.clear() if self._writer: self._writer.close() try: @@ -136,13 +148,25 @@ class SignalChannel(Channel): raise ChannelError(f"Cannot connect to signal-cli: {e}") async def _listen_loop(self) -> None: - """Listen for incoming JSON RPC notifications.""" + """Listen for incoming JSON RPC notifications and responses.""" while self._running and self._reader: try: line = await self._reader.readline() if not line: break data = json.loads(line.decode()) + # Dispatch RPC response if it has an 'id' matching a pending call + rpc_id = data.get("id") + if rpc_id is not None and rpc_id in self._pending_rpcs: + fut = self._pending_rpcs.pop(rpc_id) + if not fut.done(): + if "error" in data: + fut.set_exception( + RuntimeError(f"signal-cli RPC error: {data['error']}") + ) + else: + fut.set_result(data.get("result")) + continue await self._handle_rpc(data) except asyncio.CancelledError: break @@ -343,22 +367,35 @@ class SignalChannel(Channel): def _is_ready(self) -> bool: return self._writer is not None and not self._writer.is_closing() - async def _rpc_call(self, method: str, params: dict) -> dict | None: - """Send a JSON RPC call to signal-cli.""" + async def _rpc_call(self, method: str, params: dict, timeout: float = 10.0) -> dict | None: + """Send a JSON RPC call to signal-cli and wait for the response.""" if not self._writer: return None self._rpc_id += 1 + rpc_id = self._rpc_id request = { "jsonrpc": "2.0", - "id": self._rpc_id, + "id": rpc_id, "method": method, "params": params, } + + # Register a Future before sending so the listen loop can resolve it + loop = asyncio.get_running_loop() + fut: asyncio.Future = loop.create_future() + self._pending_rpcs[rpc_id] = fut + line = json.dumps(request) + "\n" self._writer.write(line.encode()) await self._writer.drain() - return None # We don't wait for response in this simple impl + + try: + return await asyncio.wait_for(fut, timeout=timeout) + except asyncio.TimeoutError: + self._pending_rpcs.pop(rpc_id, None) + logger.warning(f"Signal RPC '{method}' timed out after {timeout}s") + return None async def _send_chunk( self, chat_id, formatted_text, raw_text, reply_to, metadata,