Codex mobile notifications with Bark and Hammerspoon (sanitized example config)

README.md

Codex Mobile Notify

将 Codex 生命周期事件通过 Bark 推送到手机，并在 Hammerspoon 菜单栏中提供开关。

文件

• codex_mobile_notify.py：Codex hook 与通知命令。
• config.json：通知规则与 Bark 地址。发布的版本不含个人信息。
• hammerspoon.lua：菜单栏开关。

运行时文件（events.jsonl 与 state/）会在本机自动创建，不应提交或发布。

配置 Bark

在 Bark App 中复制自己的推送地址，然后编辑 config.json 的 bark_url_template。示例中的 yourkey 必须替换为自己的 key，${message} 必须保留：

"bark_url_template": "https://api.day.app/yourkey/${message}"

该地址格式与 Bark 的官方用法 一致。

安装与菜单栏

将这些文件放到 ~/.codex/mobile-notify/，并创建命令入口：

mkdir -p ~/.local/bin
ln -sf ~/.codex/mobile-notify/codex_mobile_notify.py ~/.local/bin/codex-mobile-notify
chmod +x ~/.codex/mobile-notify/codex_mobile_notify.py

在 ~/.hammerspoon/init.lua 中加载菜单栏脚本：

dofile(os.getenv("HOME") .. "/.codex/mobile-notify/hammerspoon.lua")

重载 Hammerspoon 后，点击铃铛即可在 1m 与 never 之间切换任务完成通知。可用下列命令查看或测试配置：

codex-mobile-notify status
codex-mobile-notify test alarm

配置 Codex Hooks

将下面的内容保存到 ~/.codex/hooks.json。如果该文件已经存在，请把这三项事件合并到现有的 hooks 对象中，不要直接覆盖其他 hook：

{
  "hooks": {
    "UserPromptSubmit": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "/usr/bin/python3 \"$HOME/.codex/mobile-notify/codex_mobile_notify.py\" hook UserPromptSubmit",
            "timeout": 5
          }
        ]
      }
    ],
    "PermissionRequest": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "/usr/bin/python3 \"$HOME/.codex/mobile-notify/codex_mobile_notify.py\" hook PermissionRequest",
            "timeout": 12
          }
        ]
      }
    ],
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "/usr/bin/python3 \"$HOME/.codex/mobile-notify/codex_mobile_notify.py\" hook Stop",
            "timeout": 12
          }
        ]
      }
    ]
  }
}

Codex 默认启用 hooks。如果你曾在 ~/.codex/config.toml 中关闭它，请改为：

[features]
hooks = true

重新启动 Codex，然后使用 /hooks 检查并信任这些命令。Codex 会按 hook 内容的哈希记录信任状态；以后修改命令时，需要再次审核。

codex_mobile_notify.py

#!/usr/bin/env python3
"""Bark notifications for Codex lifecycle hooks.

The hook always fails open: notification errors are logged locally and never
block or continue a Codex turn.
"""

# Hammerspoon 配置关联：~/.hammerspoon/init.lua
# 菜单栏通过 `codex-mobile-notify time-unit 1m|5m|never` 调整通知状态。

from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path
import re
import sys
import time
from typing import Any
from urllib import error, parse, request


ROOT = Path.home() / ".codex" / "mobile-notify"
CONFIG_PATH = ROOT / "config.json"
LOG_PATH = ROOT / "events.jsonl"
STATE_DIR = ROOT / "state"
DEFAULT_TASK_TIME_UNIT = "1m"
TASK_TIME_UNITS_SECONDS = {"1m": 60.0, "5m": 300.0}
TASK_RULE_PATTERN = re.compile(r"^(<|>)(\d+(?:\.\d+)?)x$")
TEST_DURATION_PATTERN = re.compile(r"^(\d+(?:\.\d+)?)(s|m|h)?$")


def ensure_layout() -> None:
    ROOT.mkdir(parents=True, exist_ok=True)
    STATE_DIR.mkdir(parents=True, exist_ok=True)


def load_config() -> dict[str, Any]:
    with CONFIG_PATH.open("r", encoding="utf-8") as handle:
        config = json.load(handle)
    if not isinstance(config, dict):
        raise ValueError("Config root must be an object")
    return config


def save_config(config: dict[str, Any]) -> None:
    ensure_layout()
    temporary = CONFIG_PATH.with_suffix(".tmp")
    with temporary.open("w", encoding="utf-8") as handle:
        json.dump(config, handle, ensure_ascii=False, indent=2)
        handle.write("\n")
    os.chmod(temporary, 0o600)
    os.replace(temporary, CONFIG_PATH)


def task_time_unit(config: dict[str, Any]) -> str:
    value = str(config.get("task_time_unit", DEFAULT_TASK_TIME_UNIT)).strip().lower()
    if value != "never" and value not in TASK_TIME_UNITS_SECONDS:
        raise ValueError("task_time_unit must be 1m, 5m, or never")
    return value


def set_task_time_unit(value: str) -> None:
    normalized = value.strip().lower()
    if normalized != "never" and normalized not in TASK_TIME_UNITS_SECONDS:
        raise ValueError("task_time_unit must be 1m, 5m, or never")
    config = load_config()
    config["task_time_unit"] = normalized
    save_config(config)


def append_log(event: str, **fields: Any) -> None:
    ensure_layout()
    record = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "event": event,
        **fields,
    }
    with LOG_PATH.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, ensure_ascii=False) + "\n")


def safe_id(payload: dict[str, Any], suffix: str = "") -> str:
    raw = "|".join(
        [
            str(payload.get("session_id", "unknown")),
            str(payload.get("turn_id", "unknown")),
            suffix,
        ]
    )
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def state_path(payload: dict[str, Any]) -> Path:
    return STATE_DIR / f"turn-{safe_id(payload)}.json"


def load_state(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as handle:
            return json.load(handle)
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        return {}


def save_state(path: Path, state: dict[str, Any]) -> None:
    ensure_layout()
    temporary = path.with_suffix(".tmp")
    with temporary.open("w", encoding="utf-8") as handle:
        json.dump(state, handle, ensure_ascii=False)
        handle.write("\n")
    os.replace(temporary, path)


def collapse_text(value: Any, limit: int) -> str:
    text = re.sub(r"\s+", " ", str(value or "")).strip()
    if len(text) > limit:
        return text[: max(0, limit - 1)].rstrip() + "…"
    return text


def format_duration(total_seconds: int) -> str:
    remaining = max(0, int(total_seconds))
    parts: list[str] = []
    for label, unit_seconds in (
        ("天", 86_400),
        ("小时", 3_600),
        ("分", 60),
        ("秒", 1),
    ):
        value, remaining = divmod(remaining, unit_seconds)
        if value:
            parts.append(f"{value} {label}")
        if len(parts) == 2:
            break
    return " ".join(parts) if parts else "0 秒"


def parse_test_duration(value: str) -> float:
    match = TEST_DURATION_PATTERN.fullmatch(value.strip().lower())
    if not match:
        raise ValueError("Duration must look like 30s, 2m, or 1.5h")
    amount_text, unit = match.groups()
    multiplier = {None: 1, "s": 1, "m": 60, "h": 3_600}[unit]
    return float(amount_text) * multiplier


def project_name(payload: dict[str, Any]) -> str:
    cwd = str(payload.get("cwd", "")).strip()
    return Path(cwd).name if cwd else "Codex"


def opener_for(config: dict[str, Any]) -> request.OpenerDirector:
    proxy_url = str(config.get("proxy_url", "")).strip()
    if not proxy_url:
        return request.build_opener()
    return request.build_opener(
        request.ProxyHandler({"http": proxy_url, "https": proxy_url})
    )


def build_url(template: str, message: str, notification_query: str) -> str:
    encoded = parse.quote(message, safe="")
    if "${message}" in template:
        url = template.replace("${message}", encoded)
    elif "{message}" in template:
        url = template.replace("{message}", encoded)
    else:
        raise ValueError("Bark URL template is missing a message placeholder")

    query = notification_query.strip().lstrip("?&")
    if not query:
        return url
    separator = "&" if "?" in url else "?"
    return f"{url}{separator}{query}"


def task_notification_query(
    config: dict[str, Any], elapsed_seconds: float | None
) -> str | bool:
    unit = task_time_unit(config)
    if unit == "never" or elapsed_seconds is None:
        return False

    rules = config.get("task_notifications")
    if not isinstance(rules, dict) or not rules:
        raise ValueError("task_notifications must be a non-empty object")

    greater_matches: list[tuple[float, Any]] = []
    less_matches: list[tuple[float, Any]] = []
    unit_seconds = TASK_TIME_UNITS_SECONDS[unit]
    for condition, action in rules.items():
        match = TASK_RULE_PATTERN.fullmatch(str(condition).strip())
        if not match:
            raise ValueError(f"Invalid task notification condition: {condition}")
        operator, multiple_text = match.groups()
        multiple = float(multiple_text)
        threshold = multiple * unit_seconds
        if operator == ">" and elapsed_seconds > threshold:
            greater_matches.append((multiple, action))
        elif operator == "<" and elapsed_seconds < threshold:
            less_matches.append((multiple, action))

    if greater_matches:
        action = max(greater_matches, key=lambda item: item[0])[1]
    elif less_matches:
        action = min(less_matches, key=lambda item: item[0])[1]
    else:
        return False

    if action is False:
        return False
    if not isinstance(action, str) or not action.strip():
        raise ValueError("Notification action must be false or a non-empty string")
    return action.strip()


def send_bark(
    message: str,
    notification_query: str,
    source: str,
    config: dict[str, Any],
) -> None:
    template = str(config.get("bark_url_template", ""))
    url = build_url(template, message, notification_query)
    timeout = int(config.get("request_timeout_seconds", 10))
    req = request.Request(url, headers={"User-Agent": "codex-mobile-notify/1.0"})
    try:
        with opener_for(config).open(req, timeout=timeout) as response:
            status = int(getattr(response, "status", 200))
        if status < 200 or status >= 300:
            raise RuntimeError(f"HTTP {status}")
        append_log(
            "notification_sent",
            source=source,
            notification=notification_query,
            http_status=status,
        )
    except error.HTTPError as exc:
        append_log(
            "notification_error",
            source=source,
            notification=notification_query,
            error=f"HTTP {exc.code}",
        )
        raise RuntimeError(f"Bark returned HTTP {exc.code}") from None
    except error.URLError as exc:
        reason = type(exc.reason).__name__
        append_log(
            "notification_error",
            source=source,
            notification=notification_query,
            error=f"network:{reason}",
        )
        raise RuntimeError(f"Bark network error: {reason}") from None


def record_prompt_start(payload: dict[str, Any]) -> None:
    path = state_path(payload)
    state = load_state(path)
    state.update(
        {
            "started_at": time.time(),
            "session_id": payload.get("session_id"),
            "turn_id": payload.get("turn_id"),
        }
    )
    for key in ("sent_at", "handled_at", "elapsed_seconds", "urgent", "notification"):
        state.pop(key, None)
    save_state(path, state)
    append_log("turn_started", state=path.name)


def notify_stop(payload: dict[str, Any]) -> None:
    path = state_path(payload)
    state = load_state(path)
    if state.get("handled_at") or state.get("sent_at"):
        append_log("notification_skipped", source="Stop", reason="duplicate")
        return

    config = load_config()
    started_at = state.get("started_at")
    elapsed_seconds: float | None = None
    if isinstance(started_at, (int, float)):
        elapsed_seconds = max(0.0, time.time() - float(started_at))

    notification = task_notification_query(config, elapsed_seconds)
    elapsed = int(elapsed_seconds) if elapsed_seconds is not None else None
    if notification is False:
        now = time.time()
        unit = task_time_unit(config)
        if unit == "never":
            reason = "never"
        elif elapsed_seconds is None:
            reason = "missing_duration"
        else:
            reason = "notification_rule"
        append_log("notification_skipped", source="Stop", reason=reason)
        state["handled_at"] = now
        state["elapsed_seconds"] = elapsed
        state["notification"] = False
        save_state(path, state)
        return

    preview = collapse_text(
        payload.get("last_assistant_message"), int(config.get("max_preview_chars", 180))
    )
    duration_text = f" · {format_duration(elapsed)}" if elapsed is not None else ""
    message = f"Codex 任务完成 · {project_name(payload)}{duration_text}"
    if preview:
        message += f"\n{preview}"

    send_bark(message, notification, source="Stop", config=config)
    now = time.time()
    state["handled_at"] = now
    state["sent_at"] = now
    state["elapsed_seconds"] = elapsed
    state["notification"] = notification
    save_state(path, state)


def notify_permission(payload: dict[str, Any]) -> None:
    config = load_config()
    if task_time_unit(config) == "never":
        append_log("notification_skipped", source="PermissionRequest", reason="never")
        return

    description = collapse_text(
        (payload.get("tool_input") or {}).get("description")
        if isinstance(payload.get("tool_input"), dict)
        else "",
        int(config.get("max_preview_chars", 180)),
    )
    suffix = f"permission|{payload.get('tool_name', '')}|{description}"
    path = STATE_DIR / f"permission-{safe_id(payload, suffix)}.json"
    state = load_state(path)
    sent_at = state.get("sent_at")
    dedupe_seconds = int(config.get("permission_dedupe_seconds", 60))
    if isinstance(sent_at, (int, float)) and time.time() - float(sent_at) < dedupe_seconds:
        append_log("notification_skipped", source="PermissionRequest", reason="duplicate")
        return

    message = f"Codex 需要你接手 · {project_name(payload)}"
    if description:
        message += f"\n{description}"
    notification = str(config.get("permission_notification", "call=1")).strip()
    if not notification:
        raise ValueError("permission_notification must not be empty")
    send_bark(message, notification, source="PermissionRequest", config=config)
    save_state(path, {"sent_at": time.time()})


def hook_main(event_name: str) -> int:
    ensure_layout()
    try:
        payload = json.load(sys.stdin)
        if event_name == "UserPromptSubmit":
            record_prompt_start(payload)
        elif event_name == "Stop":
            notify_stop(payload)
        elif event_name == "PermissionRequest":
            notify_permission(payload)
        else:
            append_log("hook_ignored", source=event_name)
    except Exception as exc:  # Notifications must never interfere with Codex.
        append_log("hook_error", source=event_name, error=type(exc).__name__)
    print("{}")
    return 0


def control_main(args: list[str]) -> int:
    command = args[0] if args else "status"
    if command == "time-unit":
        if len(args) == 1:
            print(task_time_unit(load_config()))
            return 0
        if len(args) == 2:
            set_task_time_unit(args[1])
            print(task_time_unit(load_config()))
            return 0
        print("Usage: codex-mobile-notify time-unit [1m|5m|never]", file=sys.stderr)
        return 2
    if command == "status":
        config = load_config()
        unit = task_time_unit(config)
        print(f"Codex mobile notifications: {'OFF' if unit == 'never' else 'ON'}")
        print(f"Task time unit: {unit}")
        print(f"Log: {LOG_PATH}")
        return 0
    if command == "test":
        config = load_config()
        test_name = args[1] if len(args) > 1 else "alarm"
        notifications = {
            "birdsong": "sound=birdsong",
            "alarm": "sound=alarm",
            "call": "call=1",
        }
        notification = notifications.get(test_name)
        if notification is None:
            print(
                "Usage: codex-mobile-notify test [birdsong|alarm|call]",
                file=sys.stderr,
            )
            return 2
        send_bark(
            f"Codex 手机通知测试 · {test_name}",
            notification,
            source="manual_test",
            config=config,
        )
        print(f"Bark test sent ({test_name})")
        return 0
    if command == "test-duration":
        if len(args) != 2:
            print(
                "Usage: codex-mobile-notify test-duration <30s|2m|1.5h>",
                file=sys.stderr,
            )
            return 2
        try:
            elapsed_seconds = parse_test_duration(args[1])
        except ValueError as exc:
            print(str(exc), file=sys.stderr)
            return 2

        duration = format_duration(int(elapsed_seconds))
        test_id = f"manual-duration-{time.time_ns()}"
        payload = {
            "session_id": "manual-duration-test",
            "turn_id": test_id,
            "cwd": str(ROOT),
            "last_assistant_message": "手动时长规则测试",
        }
        path = state_path(payload)
        save_state(
            path,
            {
                "started_at": time.time() - elapsed_seconds,
                "session_id": payload["session_id"],
                "turn_id": payload["turn_id"],
            },
        )
        notify_stop(payload)

        result = load_state(path)
        notification = result.get("notification", False)
        if notification is False:
            print(f"Bark duration test skipped by rule ({duration})")
        else:
            print(f"Bark duration test sent ({duration} -> {notification})")
        return 0
    print(
        "Usage: codex-mobile-notify status|time-unit [1m|5m|never]|"
        "test [birdsong|alarm|call]|test-duration <30s|2m|1.5h>",
        file=sys.stderr,
    )
    return 2


def main() -> int:
    ensure_layout()
    if len(sys.argv) >= 3 and sys.argv[1] == "hook":
        return hook_main(sys.argv[2])
    return control_main(sys.argv[1:])


if __name__ == "__main__":
    raise SystemExit(main())

config.json

{
  "task_time_unit": "never",
  "task_notifications": {
    "<1x": false,
    ">1x": "sound=birdsong",
    ">3x": "sound=alarm",
    ">10x": "sound=sherwoodforest",
    ">20x": "call=1"
  },
  "bark_url_template": "https://api.day.app/yourkey/${message}",
  "permission_notification": "call=1",
  "request_timeout_seconds": 10,
  "max_preview_chars": 180,
  "permission_dedupe_seconds": 60,
  "proxy_url": ""
}

hammerspoon.lua

-- Codex 手机通知菜单栏开关（无需 OnlySwitch）

local codexMobileNotify = {
    cli = "/Users/sam/.local/bin/codex-mobile-notify",
    menu = nil
}

local function readTaskTimeUnit()
    local output, status = hs.execute(codexMobileNotify.cli .. " time-unit", false)
    if status == true then
        local unit = output:match("^%s*(.-)%s*$")
        if unit == "1m" or unit == "5m" or unit == "never" then
            return unit
        end
    end
    return "1m"
end

local function refreshCodexMenuTitle(unit)
    local enabled = unit ~= "never"
    codexMobileNotify.menu:setTitle(enabled and "🔔" or "🔕")
    codexMobileNotify.menu:setTooltip(
        enabled and "Codex 手机通知 · ON（1m）" or "Codex 手机通知 · OFF"
    )
end

local function toggleCodexNotification()
    local current = readTaskTimeUnit()
    local nextUnit = current == "never" and "1m" or "never"
    local output, status = hs.execute(
        codexMobileNotify.cli .. " time-unit " .. nextUnit,
        false
    )
    if status ~= true then
        hs.notify.show("Codex 手机通知", "设置失败", output or "执行脚本失败")
        return
    end
    hs.reload()
end

local function buildCodexMenu(unit)
    local enabled = unit ~= "never"
    return {
        {
            title = enabled and "通知：ON（点击关闭）" or "通知：OFF（点击开启）",
            fn = toggleCodexNotification
        },
        {title = "-"},
        {
            title = "刷新状态",
            fn = function()
                hs.reload()
            end
        }
    }
end

codexMobileNotify.menu = hs.menubar.new()
if codexMobileNotify.menu then
    local current = readTaskTimeUnit()
    refreshCodexMenuTitle(current)
    codexMobileNotify.menu:setMenu(buildCodexMenu(current))
end