codex完成对话时，自动发送提醒邮件到指定QQ邮箱中，手机安装QQ邮箱客户端并将自己设置为特别关心的发件人，即可实时收到推送. Forward Codex turn-complete notifications to email and the legacy UDP port

codex-config.toml

model = "deepseek-v4-flash"
notify = [
    "python3",
    "/home/ubuntu/.codex/notify.py",
]

notify.env

NOTIFY_SMTP_PASSWORD=exampleabcdefg

notify.py

#!/usr/bin/env python3
"""Forward Codex turn-complete notifications to email and the legacy UDP port."""

from __future__ import annotations

import concurrent.futures
import json
import logging
from logging.handlers import RotatingFileHandler
import os
from pathlib import Path
import subprocess
import sys
from typing import Any


SMTP_FROM = "example@qq.com"
SMTP_TO = "example@qq.com"
SMTP_HOST = "smtp.qq.com:465"
SMTP_USER = "example@qq.com"
SMTP_PASSWORD_FILE = Path(__file__).with_name("notify.env")
LOG_FILE = Path(__file__).with_name("notify.log")


class SecureRotatingFileHandler(RotatingFileHandler):
    """Keep the current and rotated notification logs private to this user."""

    def _open(self) -> Any:
        stream = super()._open()
        os.chmod(self.baseFilename, 0o600)
        return stream


def configure_logger() -> logging.Logger:
    """Write enough detail to debug a notification without exposing secrets."""
    LOG_FILE.touch(mode=0o600, exist_ok=True)
    os.chmod(LOG_FILE, 0o600)

    logger = logging.getLogger("codex_notify")
    logger.setLevel(logging.INFO)
    logger.propagate = False
    if not logger.handlers:
        handler = SecureRotatingFileHandler(
            LOG_FILE, maxBytes=1_000_000, backupCount=3, encoding="utf-8"
        )
        handler.setFormatter(
            logging.Formatter("%(asctime)s %(levelname)s %(message)s")
        )
        logger.addHandler(handler)
    return logger


LOGGER = configure_logger()


def read_notify_env() -> dict[str, str]:
    """Read the private SMTP secret without putting it in this script."""
    password_from_process = os.environ.get("NOTIFY_SMTP_PASSWORD")
    if password_from_process:
        return {"NOTIFY_SMTP_PASSWORD": password_from_process}

    if not SMTP_PASSWORD_FILE.exists():
        raise RuntimeError(f"missing {SMTP_PASSWORD_FILE}")
    if SMTP_PASSWORD_FILE.stat().st_mode & 0o077:
        raise RuntimeError(f"refusing to read {SMTP_PASSWORD_FILE}: permissions must be 600")

    values: dict[str, str] = {}
    for raw_line in SMTP_PASSWORD_FILE.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip().strip("\"'")
    if not values.get("NOTIFY_SMTP_PASSWORD"):
        raise RuntimeError(f"NOTIFY_SMTP_PASSWORD is missing from {SMTP_PASSWORD_FILE}")
    return values


def redact_output(text: str, password: str | None = None) -> str:
    """Keep subprocess diagnostics while removing SMTP authentication data."""
    if not text:
        return "<empty>"

    safe_lines: list[str] = []
    for line in text.splitlines():
        # sendemail.py enables smtplib debug output. `send:` records can contain
        # MIME data or a Base64-encoded SMTP credential, so retain replies and
        # failures but never persist outbound SMTP protocol data.
        if line.lstrip().lower().startswith("send:"):
            safe_lines.append("<redacted outbound SMTP protocol data>")
        elif "auth " in line.lower() or "password" in line.lower():
            safe_lines.append("<redacted SMTP authentication data>")
        else:
            safe_lines.append(line)

    safe_text = "\n".join(safe_lines)
    if password:
        safe_text = safe_text.replace(password, "<redacted SMTP password>")
    return safe_text


def payload_text(value: Any) -> str:
    """Keep a non-string payload readable without exposing unrelated fields."""
    if isinstance(value, str):
        return value.strip()
    if value is None:
        return ""
    return json.dumps(value, ensure_ascii=False, indent=2)


def notification_text(notification: dict[str, Any]) -> tuple[str, str]:
    """Format one user request and its final assistant response for email."""
    cwd = payload_text(notification.get("cwd")).rstrip("/")
    project_name = Path(cwd).name if cwd else "Codex"
    subject = f"Codex | {project_name[:60] or 'Codex'}"

    input_messages = notification.get("input-messages")
    if isinstance(input_messages, list) and input_messages:
        last_input = payload_text(input_messages[-1])
    else:
        last_input = ""
    last_response = payload_text(notification.get("last-assistant-message"))

    # The event can contain prior user messages and internal metadata. Email
    # only needs the final user request and the answer that completed this turn.
    body = "\n\n".join(
        [
            f"User request:\n{last_input or '<unavailable>'}",
            f"Assistant response:\n{last_response or '<unavailable>'}",
        ]
    )
    return subject, body


def send_email(subject: str, body: str) -> tuple[subprocess.CompletedProcess[str], str]:
    password = read_notify_env()["NOTIFY_SMTP_PASSWORD"]
    # sendemail.py needs the password as an argument. Do not log this command
    # directly: its argument list would contain the SMTP authorization code.
    command = [
        "sendemail.py",
        "--ssl",
        "-f",
        SMTP_FROM,
        "-t",
        SMTP_TO,
        "-u",
        subject,
        "-m",
        body,
        "-s",
        SMTP_HOST,
        "-xu",
        SMTP_USER,
        "-xp",
        password,
    ]
    LOGGER.info("email: starting sendemail.py with SSL to %s", SMTP_HOST)
    result = subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        timeout=30,
    )
    return result, password


def send_legacy_udp() -> subprocess.CompletedProcess[str]:
    """Preserve the existing `echo test | nc -u -w 1 ...` notification."""
    LOGGER.info("udp: starting nc -u -w 1 127.0.0.1 54321")
    return subprocess.run(
        ["nc", "-u", "-w", "1", "127.0.0.1", "54321"],
        input="test\n",
        check=False,
        capture_output=True,
        text=True,
        timeout=2,
    )


def log_result(
    name: str, result: subprocess.CompletedProcess[str], password: str | None = None
) -> bool:
    """Record every child process result and identify sendemail.py's false success."""
    LOGGER.info("%s: exit code %d", name, result.returncode)
    LOGGER.info("%s stdout:\n%s", name, redact_output(result.stdout, password))
    LOGGER.info("%s stderr:\n%s", name, redact_output(result.stderr, password))

    if result.returncode != 0:
        return False
    # sendemail.py catches SMTPException itself and can exit 0 after printing a
    # failure. Its explicit success line is therefore the reliable signal.
    if name == "email" and "Email sent successfully." not in result.stdout:
        LOGGER.error("email: sendemail.py exited 0 without its success marker")
        return False
    return True


def safe_error(error: BaseException, password: str | None = None) -> str:
    """Avoid putting a subprocess command line, including its password, in logs."""
    if isinstance(error, subprocess.TimeoutExpired):
        return f"timed out after {error.timeout} seconds"
    return redact_output(str(error), password)


def main() -> int:
    if len(sys.argv) != 2:
        message = "usage: notify.py '<notification-json>'"
        LOGGER.error(message)
        print(message, file=sys.stderr)
        return 2

    try:
        notification = json.loads(sys.argv[1])
    except json.JSONDecodeError as error:
        message = f"codex notify received invalid JSON: {error}"
        LOGGER.error(message)
        print(message, file=sys.stderr)
        return 2
    if not isinstance(notification, dict):
        message = "codex notify received a JSON value other than an object"
        LOGGER.error(message)
        print(message, file=sys.stderr)
        return 2
    if notification.get("type") != "agent-turn-complete":
        LOGGER.info("ignored notification type: %r", notification.get("type"))
        return 0

    subject, body = notification_text(notification)
    LOGGER.info(
        "notification received: thread-id=%r turn-id=%r cwd=%r subject=%r",
        notification.get("thread-id"),
        notification.get("turn-id"),
        notification.get("cwd"),
        subject,
    )
    LOGGER.info("notification body prepared: %d characters", len(body))

    # Run both transports independently so a slow or failed SMTP connection
    # cannot prevent the legacy UDP listener from receiving its datagram.
    transports = {
        "email": lambda: send_email(subject, body),
        "udp": send_legacy_udp,
    }
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        futures = {name: executor.submit(task) for name, task in transports.items()}
        for name, future in futures.items():
            try:
                outcome = future.result()
                if name == "email":
                    result, password = outcome
                else:
                    result, password = outcome, None
                if not log_result(name, result, password):
                    print(
                        f"codex notify {name} failed; inspect {LOG_FILE}",
                        file=sys.stderr,
                    )
            except Exception as error:  # Notification failures must not break Codex.
                message = safe_error(error)
                LOGGER.error("%s failed: %s", name, message)
                print(
                    f"codex notify {name} failed: {message}; inspect {LOG_FILE}",
                    file=sys.stderr,
                )
    LOGGER.info("notification handling finished")
    return 0


if __name__ == "__main__":
    sys.exit(main())