#!/usr/bin/env python3
"""
Agent Notifier — production-friendly push notification CLI for AI agents.

Sends notifications to the Agent Notifier service (POST /api/v1/messages),
the Pushover-class push platform purpose-built for agents. The CLI mirrors
the public API while keeping customer machines private by default.

Quick usage:
    notify.py "Build finished"
    notify.py "Tests failed" --title "CI" --priority 1 --type alert
    notify.py "Deploying..." --progress 0.4
    notify.py "Running tests" --step 3/5
    notify.py "Review PR" --url https://example.com --url-title "Open PR"
    notify.py "Need input" --button approve --button reject --button defer
    notify.py "See screenshot" --file-path ./screenshot.png
    notify.py "Done" --project my-repo --metadata branch=main --metadata commit=abc123

Authentication:
    No shared API key is embedded. Provide an account API key using one of:
      1. AGENT_NOTIFIER_API_KEY in the environment
      2. AGENT_NOTIFIER_API_KEY in .env.local or .env in the current directory
      3. ~/.config/agent-notifier/config.json (or --config)
      4. --api-key for one-off use (less private: shell history/process list)
      5. --api-key-stdin to pipe it in without exposing it in the process list

Config example:
    printf %s 'an_key_REPLACE_WITH_YOUR_KEY' | notify.py --save-config --api-key-stdin
    Saving without --project is the recommended setup: the project is then
    auto-detected per repository on every send. Pass --project only to pin one.

Project name:
    If --project is omitted, the CLI checks AGENT_NOTIFIER_PROJECT (environment,
    then the repo's .env files), then the git repo name (origin remote, else the
    repo folder). A config `project` only applies outside git repositories; the
    last resort is the current directory name.

Privacy defaults:
    --file-path uploads bytes to /api/v1/media, but does not send the local
    absolute path in the message payload. Use --reference-file-path for
    metadata-only path context, or --include-local-paths to explicitly send
    upload paths as file_paths metadata.
"""

from __future__ import annotations

import argparse
import json
import mimetypes
import os
import random
import re
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from dataclasses import dataclass
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Any, Callable

__version__ = "0.8.0"

DEFAULT_BASE_URL = "https://notifier.aicrew.in"
DEFAULT_API_PATH = "/api/v1"
VALID_TYPES = ("alert", "progress", "session_start", "session_end", "url", "file")
VALID_PRIORITIES = (-1, 0, 1, 2)
VALID_COLORS = ("red", "orange", "amber", "lime", "green", "teal", "cyan", "blue", "purple", "pink")
MAX_BUTTONS = 4
MAX_BUTTON_LABEL = 64
MAX_TITLE = 256
MAX_MESSAGE = 4096
MAX_MEDIA_UPLOAD_BYTES = 25 << 20
MAX_METADATA_ENTRIES = 64
MAX_METADATA_KEY = 64
MAX_FILE_PATHS = 12
DEFAULT_TTS_ENGINE = "azure"
DEFAULT_TTS_VOICE = "en-US-Olivia:MAI-Voice-2"
DEFAULT_TTS_SPEED = 1.0
MIN_TTS_SPEED = 0.5
MAX_TTS_SPEED = 2.0
MAX_TTS_TEXT_CHARS = 12_000
VALID_TTS_ENGINES = ("azure",)
# Safe charset for Azure AI Speech voice ids like en-US-Olivia:MAI-Voice-2.
TTS_VOICE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{1,63}$")
TRANSIENT_STATUSES = {408, 425, 429, 500, 502, 503, 504}
USER_AGENT = f"AgentNotifierCLI/{__version__}"


class NotifyError(Exception):
    """User-facing CLI error."""


class ButtonAction(argparse.Action):
    """Append one or many button labels while preserving argument order."""

    def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
                 values: str | list[str], option_string: str | None = None) -> None:
        current = getattr(namespace, self.dest, None)
        if current is None:
            current = []
        if isinstance(values, list):
            current.extend(values)
        else:
            current.append(values)
        setattr(namespace, self.dest, current)


@dataclass
class HTTPResult:
    status: int
    body: dict[str, Any] | str
    headers: dict[str, str]


def default_config_path() -> Path:
    raw = os.environ.get("AGENT_NOTIFIER_CONFIG", "").strip()
    if raw:
        return Path(raw).expanduser()
    return Path.home() / ".config" / "agent-notifier" / "config.json"


def load_config(path: Path) -> dict[str, str]:
    if not path.is_file():
        return {}
    text = path.read_text(encoding="utf-8")
    if not text.strip():
        return {}
    try:
        raw = json.loads(text)
    except (OSError, json.JSONDecodeError) as exc:
        raise NotifyError(f"could not read config {path}: {exc}") from exc
    if not isinstance(raw, dict):
        raise NotifyError(f"config {path} must contain a JSON object")
    return {str(k): str(v) for k, v in raw.items() if v is not None}


def load_env_file(path: Path) -> dict[str, str]:
    if not path.is_file():
        return {}
    values: dict[str, str] = {}
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError:
        return {}
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if stripped.startswith("export "):
            stripped = stripped[len("export ") :].strip()
        key, sep, value = stripped.partition("=")
        if not sep:
            continue
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key.startswith("AGENT_NOTIFIER_") and key not in values:
            values[key] = value
    return values


def load_local_env() -> dict[str, str]:
    explicit = os.environ.get("AGENT_NOTIFIER_ENV_FILE", "").strip()
    candidates = [Path(explicit).expanduser()] if explicit else [Path.cwd() / ".env.local", Path.cwd() / ".env"]
    merged: dict[str, str] = {}
    for path in candidates:
        merged.update({k: v for k, v in load_env_file(path).items() if k not in merged})
    return merged


def pick_setting(*values: Any, default: str | None = None) -> str | None:
    for value in values:
        if value is None:
            continue
        text = str(value).strip()
        if text:
            return text
    return default


def read_api_key_from_stdin() -> str:
    """Read the API key from stdin.

    A key passed as `--api-key <secret>` is visible in `ps` to every other user
    on the machine for the lifetime of the process. Piping it from a shell
    builtin (`printf %s '<key>' | notify.py --api-key-stdin`) keeps it out of
    the process list entirely.
    """
    key = sys.stdin.readline().strip()
    if not key:
        raise NotifyError("--api-key-stdin was given but stdin was empty")
    return key


def save_config(path: Path, values: dict[str, str]) -> None:
    existing = load_config(path)
    for key, value in values.items():
        if value:
            existing[key] = value
    if not existing.get("api_key"):
        raise NotifyError("--save-config requires --api-key, --api-key-stdin, or AGENT_NOTIFIER_API_KEY")

    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(json.dumps(existing, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    os.chmod(tmp, 0o600)
    tmp.replace(path)
    os.chmod(path, 0o600)


def _git(args: list[str], cwd: Path) -> str:
    try:
        return subprocess.run(
            ["git", *args], cwd=cwd, check=True, stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL, text=True, timeout=1.0,
        ).stdout.strip()
    except (subprocess.SubprocessError, FileNotFoundError, OSError):
        return ""


def repo_name_from_remote(url: str) -> str | None:
    """`git@github.com:owner/name.git`, `https://host/group/name` -> `name`."""
    tail = url.strip().rstrip("/").rsplit("/", 1)[-1].rsplit(":", 1)[-1]
    name = tail[:-4] if tail.endswith(".git") else tail
    return name or None


def detect_project(cwd: Path | None = None) -> str | None:
    """The repository's name: its origin remote, else its top-level folder.

    Returns None outside a git repository so a configured default can apply.
    """
    here = cwd or Path.cwd()
    root = _git(["rev-parse", "--show-toplevel"], here)
    if not root:
        return None
    remote = _git(["remote", "get-url", "origin"], here) or _git(["config", "--get", "remote.origin.url"], here)
    return repo_name_from_remote(remote) if remote else Path(root).name


def resolve_project(cli: str | None, local_env: dict[str, str], config: dict[str, str], cwd: Path | None = None) -> str:
    """--project > AGENT_NOTIFIER_PROJECT (env, then repo .env files) > the git
    repo > the config default > the folder name.

    A config `project` is a machine-wide fallback for non-repo folders only; it
    must never rename every repository's notifications to one name.
    """
    here = cwd or Path.cwd()
    chosen = pick_setting(cli, os.environ.get("AGENT_NOTIFIER_PROJECT"), local_env.get("AGENT_NOTIFIER_PROJECT"))
    return chosen or detect_project(here) or config.get("project") or here.name or "unknown"


def parse_metadata(items: list[str]) -> dict[str, str]:
    """Parse `key=value` pairs into a flat str->str map."""
    md: dict[str, str] = {}
    for item in items:
        if "=" not in item:
            raise NotifyError(f"--metadata expects key=value, got: {item!r}")
        key, _, value = item.partition("=")
        key = key.strip()
        if not key:
            raise NotifyError(f"--metadata key cannot be empty: {item!r}")
        if len(key) > MAX_METADATA_KEY:
            raise NotifyError(f"--metadata key is too long: {key!r}")
        if key in md:
            raise NotifyError(f"duplicate --metadata key: {key}")
        md[key] = value
    if len(md) > MAX_METADATA_ENTRIES:
        raise NotifyError(f"metadata: maximum {MAX_METADATA_ENTRIES} entries")
    return md


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="notify.py",
        description="Send a push notification via Agent Notifier (POST /api/v1/messages).",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
        allow_abbrev=False,
    )
    p.add_argument("message", nargs="?", default="", help="Notification body (max 4096 chars)")
    p.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
    p.add_argument("--title", "-t", default=None, help="Push title (max 256 chars). Defaults to project name when omitted.")
    p.add_argument("--type", choices=VALID_TYPES, default=None, help="Notification type (default: alert, progress when --progress is set, file when --file-path is set)")
    p.add_argument("--priority", "-p", type=int, choices=VALID_PRIORITIES, default=0, help="-1=silent, 0=normal, 1=high (bypass DND), 2=critical (default: 0)")
    p.add_argument("--project", default=None, help="Project name (default: env/config/current git repo)")
    p.add_argument("--session-id", default=None, help="Existing server session UUID to attach to")
    p.add_argument("--client-event-id", default=None, help="UUID idempotency key for retries/replays (default: generated per invocation)")
    p.add_argument("--no-client-event-id", action="store_true", help="Do not send a client_event_id; disables safe send retries")
    p.add_argument("--check-status", metavar="ID", default=None, help="Fetch status for a previously accepted notification ID and exit")
    p.add_argument("--progress", type=float, default=None, help="Progress 0.0–1.0 (sets default type=progress)")
    p.add_argument("--step", default=None, metavar="N/M", help="Step N of M, e.g. 3/5: shown as '3 of 5' on Live Activities and sessions; implies --progress N/M")
    p.add_argument("--image-url", default=None, help="Deprecated HTTPS image URL; prefer --file-path or POST /media attachment")
    p.add_argument("--as-audio", action="store_true", help="Ask the API to generate speech audio asynchronously from message text or the first Markdown/text --file-path")
    p.add_argument("--voice", "--tts-voice", dest="voice", default=None, help=f"TTS voice id when --as-audio is set (default: {DEFAULT_TTS_VOICE})")
    p.add_argument("--voice-engine", "--tts-engine", dest="voice_engine", default=None, help=f"TTS engine when --as-audio is set (default: {DEFAULT_TTS_ENGINE})")
    p.add_argument("--speed", "--tts-speed", dest="speed", type=float, default=None, help=f"TTS speech speed {MIN_TTS_SPEED}–{MAX_TTS_SPEED} when --as-audio is set (default: {DEFAULT_TTS_SPEED})")
    p.add_argument("--rewrite-for-tts", action="store_true", help="Request server-side text rewriting before TTS (currently an opt-in future flag)")
    p.add_argument("--url", default=None, help="Clickable HTTP(S) URL")
    p.add_argument("--url-title", default=None, help="Display text for --url")
    p.add_argument("--button", "-b", action=ButtonAction, default=[], dest="buttons", help="Action button label (repeat up to 4)")
    p.add_argument("--buttons", nargs="+", action=ButtonAction, default=[], dest="buttons", help="Compatibility alias for multiple action button labels; prefer repeated --button")
    ask = p.add_argument_group("ask (blocking)")
    ask.add_argument("--ask", metavar="QUESTION", default=None, help="Ask a human and block until they answer. Prints the answer; exit code is the chosen index, or 124 on timeout.")
    ask.add_argument("--ask-text", metavar="QUESTION", default=None, help="Ask for a typed answer rather than a choice")
    ask.add_argument("--choice", "-c", action="append", default=[], help="An option for --ask (repeatable). Allow/Deny, Yes/No and Continue/Stop earn real Lock Screen buttons.")
    ask.add_argument("--allow-free-form", action="store_true", help="Accept a typed answer alongside --choice options")
    ask.add_argument("--ask-timeout", type=float, default=300, help="Seconds to wait for an answer (default: 300, max 3600)")
    p.add_argument("--sound", default=None, help="Push sound name (default: server default)")
    p.add_argument("--ttl", type=int, default=None, help="TTL seconds (0 = immediate-only)")
    p.add_argument("--device", default=None, help="Target a specific device by name")
    p.add_argument("--color", choices=VALID_COLORS, default=None, help="Notification background color (fixed accessible palette; alerts default orange)")
    p.add_argument("--metadata", "-m", action="append", default=[], help="Flat metadata key=value (repeatable, max 64). Example: -m branch=main")

    # Substance fields
    p.add_argument("--event", default=None, help="Substance: hook event name (e.g., PostToolUse)")
    p.add_argument("--tool-name", default=None, help="Substance: tool name")
    p.add_argument("--command", default=None, help="Substance: redacted command/tool input")
    p.add_argument("--prompt", default=None, help="Substance: redacted user prompt")
    p.add_argument("--file-path", action="append", default=[], dest="file_paths", help="Local file to upload and attach. Local paths are not sent unless --include-local-paths is set. Repeatable; the API renders one primary attachment plus legacy image/link fallbacks.")
    p.add_argument("--reference-file-path", action="append", default=[], dest="reference_file_paths", help="Metadata-only referenced file path; does not upload bytes (repeatable)")
    p.add_argument("--include-local-paths", action="store_true", help="Also send --file-path values as file_paths metadata. Off by default for privacy.")
    p.add_argument("--parent-id", default=None, help="Substance: parent message ID")
    p.add_argument("--subagent-id", default=None, help="Substance: subagent ID")
    p.add_argument("--status", default=None, help="Substance: status string (e.g., success/failed)")

    # Connection / behavior
    p.add_argument("--api-key", default=None, help="Override API key (prefer env/config; command-line args can appear in shell history)")
    p.add_argument("--api-key-stdin", action="store_true", help="Read the API key from stdin so it never appears in the process list")
    p.add_argument("--base-url", default=None, help=f"Override host base (default: {DEFAULT_BASE_URL})")
    p.add_argument("--api-path", default=None, help=f"Override API path (default: {DEFAULT_API_PATH})")
    p.add_argument("--config", default=None, help="Config JSON path (default: ~/.config/agent-notifier/config.json)")
    p.add_argument("--save-config", action="store_true", help="Save api_key/base_url/api_path (and --project only if given) to config and exit")
    p.add_argument("--timeout", type=float, default=15.0, help="HTTP timeout per attempt in seconds (default: 15)")
    p.add_argument("--max-retries", type=int, default=2, help="Retries for transient network/5xx/429 failures (default: 2)")
    p.add_argument("--dry-run", action="store_true", help="Print payload + endpoint, do not upload or send")
    p.add_argument("--json", action="store_true", help="Print machine-readable JSON for success or dry-run output")
    p.add_argument("--id-only", action="store_true", help="Print only the notification ID on success")
    p.add_argument("--quiet", "-q", action="store_true", help="Suppress success output (errors still printed)")
    return p


def validate_url(value: str, field: str, *, https_only: bool) -> str:
    url = value.strip()
    parsed = urllib.parse.urlparse(url)
    schemes = {"https"} if https_only else {"http", "https"}
    if parsed.scheme.lower() not in schemes or not parsed.netloc:
        kind = "HTTPS" if https_only else "HTTP(S)"
        raise NotifyError(f"{field} must be an {kind} URL")
    if parsed.username or parsed.password:
        raise NotifyError(f"{field} must not contain credentials")
    return url


def validate_https_url(value: str, field: str) -> str:
    return validate_url(value, field, https_only=True)


def validate_click_url(value: str, field: str) -> str:
    return validate_url(value, field, https_only=False)


def validate_uuid(value: str, field: str) -> str:
    try:
        return str(uuid.UUID(value.strip()))
    except (ValueError, AttributeError) as exc:
        raise NotifyError(f"{field} must be a UUID") from exc


def resolve_client_event_id(args: argparse.Namespace) -> str | None:
    if args.no_client_event_id:
        return None
    if args.client_event_id:
        return validate_uuid(args.client_event_id, "client_event_id")
    return str(uuid.uuid4())


def parse_step(value: str) -> tuple[int, int]:
    """Parse `N/M` with 0 <= N <= M and M >= 1."""
    done, sep, total = value.partition("/")
    if not sep or not done.strip().isdigit() or not total.strip().isdigit():
        raise NotifyError(f"--step expects N/M, e.g. 3/5, got: {value!r}")
    n, m = int(done), int(total)
    if m < 1 or n > m:
        raise NotifyError(f"--step must satisfy 0 <= N <= M and M >= 1, got: {value!r}")
    return n, m


def build_payload(args: argparse.Namespace, project: str, client_event_id: str | None) -> dict[str, Any]:
    step = parse_step(args.step) if getattr(args, "step", None) else None
    if step and args.progress is None:
        args.progress = round(step[0] / step[1], 4)
    title = args.title or project
    if len(title) > MAX_TITLE:
        raise NotifyError(f"title exceeds {MAX_TITLE} chars")
    if len(args.message) > MAX_MESSAGE:
        raise NotifyError(f"message exceeds {MAX_MESSAGE} chars")
    if args.progress is not None and not 0.0 <= args.progress <= 1.0:
        raise NotifyError("progress must be between 0.0 and 1.0")
    if args.ttl is not None and args.ttl < 0:
        raise NotifyError("ttl must be non-negative")
    tts_option_used = any(
        [
            args.voice is not None,
            args.voice_engine is not None,
            args.speed is not None,
            args.rewrite_for_tts,
        ]
    )
    if tts_option_used and not args.as_audio:
        raise NotifyError("TTS options require --as-audio")
    tts_voice = DEFAULT_TTS_VOICE
    tts_engine = DEFAULT_TTS_ENGINE
    tts_speed = DEFAULT_TTS_SPEED
    if args.as_audio:
        tts_engine = (args.voice_engine or DEFAULT_TTS_ENGINE).strip().lower()
        if tts_engine == "kokoro":
            # Kokoro has been removed; coerce legacy requests to Azure MAI-Voice.
            tts_engine = "azure"
        if tts_engine not in VALID_TTS_ENGINES:
            raise NotifyError("--voice-engine must be one of: " + ", ".join(VALID_TTS_ENGINES))
        tts_voice = args.voice.strip() if args.voice else DEFAULT_TTS_VOICE
        if not TTS_VOICE_RE.match(tts_voice):
            raise NotifyError("--voice must be a safe voice id like en-US-Olivia:MAI-Voice-2")
        tts_speed = args.speed if args.speed is not None else DEFAULT_TTS_SPEED
        if not MIN_TTS_SPEED <= tts_speed <= MAX_TTS_SPEED:
            raise NotifyError(f"--speed must be between {MIN_TTS_SPEED} and {MAX_TTS_SPEED}")

    buttons = args.buttons or []
    if len(buttons) > MAX_BUTTONS:
        raise NotifyError(f"maximum {MAX_BUTTONS} buttons allowed")
    cleaned_buttons = []
    for label in buttons:
        text = str(label).strip()
        if not text:
            raise NotifyError("button labels cannot be empty")
        if len(text) > MAX_BUTTON_LABEL:
            raise NotifyError(f"button label exceeds {MAX_BUTTON_LABEL} chars: {text[:24]!r}...")
        cleaned_buttons.append(text)

    referenced_paths = [*getattr(args, "reference_file_paths", [])]
    if args.include_local_paths:
        referenced_paths.extend(args.file_paths)
    if len(referenced_paths) > MAX_FILE_PATHS:
        raise NotifyError(f"file_paths metadata supports at most {MAX_FILE_PATHS} entries")

    notification_type = args.type
    if notification_type is None:
        if args.progress is not None:
            notification_type = "progress"
        elif args.file_paths:
            notification_type = "file"
        else:
            notification_type = "alert"

    payload: dict[str, Any] = {
        "project": project,
        "type": notification_type,
        "title": title,
        "message": args.message,
        "priority": args.priority,
    }
    if client_event_id:
        payload["client_event_id"] = client_event_id
    if args.session_id:
        payload["session_id"] = validate_uuid(args.session_id, "session_id")
    if args.progress is not None:
        payload["progress"] = args.progress
    if args.image_url:
        payload["image_url"] = validate_https_url(args.image_url, "image_url")
    if args.as_audio:
        payload["as_audio"] = True
        payload["voice"] = tts_voice
        payload["voice_engine"] = tts_engine
        payload["speed"] = tts_speed
        payload["rewrite_for_tts"] = bool(args.rewrite_for_tts)
        tts_text = tts_text_from_file_paths(args.file_paths)
        if tts_text:
            payload["tts_text"] = tts_text
    if args.url:
        payload["url"] = validate_click_url(args.url, "url")
    if args.url_title:
        if not args.url and not args.file_paths:
            raise NotifyError("--url-title requires --url unless --file-path will populate a file link")
        payload["url_title"] = args.url_title.strip()
    if cleaned_buttons:
        payload["buttons"] = cleaned_buttons
    if args.sound:
        payload["sound"] = args.sound.strip()
    if args.ttl is not None:
        payload["ttl"] = args.ttl
    if args.device:
        payload["device"] = args.device.strip()
    if args.color:
        payload["color"] = args.color.strip().lower()
    if args.metadata or step:
        metadata = parse_metadata(args.metadata)
        if step:
            metadata.update({"step": str(step[0]), "steps": str(step[1])})
        payload["metadata"] = metadata

    # Substance fields
    if args.event:
        payload["event"] = args.event.strip()
    if args.tool_name:
        payload["tool_name"] = args.tool_name.strip()
    if args.command:
        payload["command"] = args.command.strip()
    if args.prompt:
        payload["prompt"] = args.prompt.strip()
    if referenced_paths:
        payload["file_paths"] = referenced_paths
    if args.parent_id:
        payload["parent_id"] = validate_uuid(args.parent_id, "parent_id")
    if args.subagent_id:
        payload["subagent_id"] = args.subagent_id.strip()
    if args.status:
        payload["status"] = args.status.strip()
    return payload


def guess_upload_content_type(path: Path) -> str:
    """Return a stable content type for the local upload."""
    mimetypes.add_type("text/markdown", ".md")
    mimetypes.add_type("image/webp", ".webp")
    mimetypes.add_type("image/heic", ".heic")
    mimetypes.add_type("audio/mp4", ".m4a")
    guessed, _ = mimetypes.guess_type(path.name)
    return guessed or "application/octet-stream"


def upload_kind(upload: dict[str, Any]) -> str:
    kind = str(upload.get("kind") or "").strip().lower()
    if kind in {"image", "audio", "video", "file"}:
        return kind
    media_type = str(upload.get("mime_type") or upload.get("content_type") or "").split(";", 1)[0].strip().lower()
    if media_type.startswith("image/"):
        return "image"
    if media_type.startswith("audio/"):
        return "audio"
    if media_type.startswith("video/"):
        return "video"
    return "file"


def is_tts_text_path(path: Path) -> bool:
    suffix = path.suffix.lower()
    if suffix in {".md", ".markdown", ".txt"}:
        return True
    content_type = guess_upload_content_type(path).split(";", 1)[0].strip().lower()
    return content_type.startswith("text/") or content_type in {"application/markdown", "text/markdown"}


def tts_text_from_file_paths(raw_paths: list[str]) -> str | None:
    for raw_path in raw_paths:
        path = resolve_upload_path(raw_path)
        if not path.is_file() or not is_tts_text_path(path):
            continue
        if path.stat().st_size > MAX_MEDIA_UPLOAD_BYTES:
            raise NotifyError(f"file exceeds {MAX_MEDIA_UPLOAD_BYTES // (1024 * 1024)} MiB upload limit: {raw_path}")
        text = path.read_text(encoding="utf-8", errors="replace").strip()
        if not text:
            continue
        return text[:MAX_TTS_TEXT_CHARS]
    return None


def media_endpoint_from_messages(endpoint: str) -> str:
    if endpoint.endswith("/messages"):
        return endpoint[: -len("/messages")] + "/media"
    return endpoint.rstrip("/") + "/media"


def resolve_upload_path(raw_path: str) -> Path:
    path = Path(raw_path).expanduser()
    if not path.is_absolute():
        path = Path.cwd() / path
    return path.resolve()


def probe_duration_ms(path: Path, content_type: str) -> int | None:
    media_type = (content_type or "").split(";", 1)[0].strip().lower()
    if not (media_type.startswith("audio/") or media_type.startswith("video/")):
        return None
    try:
        from mutagen import File as MutagenFile  # type: ignore
    except Exception:
        return None
    try:
        media = MutagenFile(path)
        length = getattr(getattr(media, "info", None), "length", None)
        if length is None:
            return None
        value = int(float(length) * 1000)
        return value if value >= 0 else None
    except Exception:
        return None


def multipart_body(path: Path, boundary: str, content_type: str, duration_ms: int | None = None) -> bytes:
    filename = path.name.replace("\\", "_").replace('"', "_") or "attachment"
    body = bytearray()
    body.extend(f"--{boundary}\r\n".encode("utf-8"))
    body.extend(f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'.encode("utf-8"))
    body.extend(f"Content-Type: {content_type}\r\n\r\n".encode("utf-8"))
    body.extend(path.read_bytes())
    if duration_ms is not None:
        body.extend(f"\r\n--{boundary}\r\n".encode("utf-8"))
        body.extend(b'Content-Disposition: form-data; name="duration_ms"\r\n\r\n')
        body.extend(str(duration_ms).encode("utf-8"))
    body.extend(f"\r\n--{boundary}--\r\n".encode("utf-8"))
    return bytes(body)


def response_body(raw: bytes) -> dict[str, Any] | str:
    text = raw.decode("utf-8", errors="replace")
    if not text:
        return ""
    try:
        decoded = json.loads(text)
    except json.JSONDecodeError:
        return text
    return decoded if isinstance(decoded, dict) else text


def headers_dict(headers: Any) -> dict[str, str]:
    return {str(k).lower(): str(v) for k, v in headers.items()}


def retry_after_seconds(headers: dict[str, str]) -> float | None:
    raw = headers.get("retry-after", "").strip()
    if not raw:
        return None
    if raw.isdigit():
        return float(raw)
    try:
        dt = parsedate_to_datetime(raw)
        return max(0.0, dt.timestamp() - time.time())
    except (TypeError, ValueError, OverflowError):
        return None


def retry_delay(attempt: int, headers: dict[str, str] | None = None) -> float:
    if headers:
        advertised = retry_after_seconds(headers)
        if advertised is not None:
            return min(advertised, 30.0)
    base = min(0.5 * (2**attempt), 8.0)
    return base + random.uniform(0.0, 0.25)


def open_json_request(req: urllib.request.Request, timeout: float) -> HTTPResult:
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return HTTPResult(resp.status, response_body(resp.read()), headers_dict(resp.headers))
    except urllib.error.HTTPError as exc:
        return HTTPResult(exc.code, response_body(exc.read()), headers_dict(exc.headers))
    except urllib.error.URLError as exc:
        raise NotifyError(f"network error: {exc.reason}") from exc


def upload_file_once(endpoint: str, api_key: str, raw_path: str, timeout: float) -> HTTPResult:
    path = resolve_upload_path(raw_path)
    if not path.is_file():
        raise NotifyError(f"file upload path does not exist or is not a file: {raw_path}")
    size = path.stat().st_size
    if size > MAX_MEDIA_UPLOAD_BYTES:
        raise NotifyError(f"file exceeds {MAX_MEDIA_UPLOAD_BYTES // (1024 * 1024)} MiB upload limit: {raw_path}")

    content_type = guess_upload_content_type(path)
    duration_ms = probe_duration_ms(path, content_type)
    boundary = f"Boundary-{uuid.uuid4().hex}"
    req = urllib.request.Request(
        endpoint,
        data=multipart_body(path, boundary, content_type, duration_ms),
        headers={
            "Content-Type": f"multipart/form-data; boundary={boundary}",
            "X-API-Key": api_key,
            "User-Agent": USER_AGENT,
        },
        method="POST",
    )
    result = open_json_request(req, timeout=max(timeout, 30.0))
    if isinstance(result.body, dict):
        result.body["filename"] = path.name
        result.body["content_type"] = result.body.get("content_type") or content_type
        result.body["mime_type"] = result.body.get("mime_type") or result.body["content_type"]
        result.body["kind"] = upload_kind(result.body)
        result.body["size_bytes"] = result.body.get("size_bytes") or size
        if duration_ms is not None and result.body.get("duration_ms") is None:
            result.body["duration_ms"] = duration_ms
    return result


def format_http_error(context: str, status: int, body: dict[str, Any] | str) -> str:
    detail = body.get("error") if isinstance(body, dict) else str(body)
    detail = str(detail or "").strip()
    if status == 400:
        return f"{context} failed validation: {detail or 'bad request'}"
    if status == 401:
        return f"{context} authentication failed: check AGENT_NOTIFIER_API_KEY or run --save-config"
    if status == 413:
        return f"{context} file is too large for the current media limit"
    if status == 429:
        if isinstance(body, dict) and body.get("is_lifetime"):
            limit = body.get("limit")
            used = body.get("used")
            url = body.get("upgrade_url") or "https://agentnotifier.app/pricing"
            usage = f" ({used}/{limit} used)" if limit and used is not None else ""
            return (
                f"{context} blocked: free Agent Notifier quota exhausted{usage}. "
                f"Upgrade to Pro for unlimited delivery: {url}"
            )
        return f"{context} was rate limited; retry after the advertised delay"
    if status in {500, 502, 503, 504}:
        return f"{context} service is temporarily unavailable (HTTP {status})"
    return f"{context} failed HTTP {status}: {detail[:240] or 'unexpected response'}"


def upload_file(endpoint: str, api_key: str, raw_path: str, timeout: float, max_retries: int) -> dict[str, Any]:
    result = retry_request(lambda: upload_file_once(endpoint, api_key, raw_path, timeout), max_retries)
    if 200 <= result.status < 300:
        if not isinstance(result.body, dict):
            raise NotifyError("media upload returned a non-JSON response")
        return result.body
    raise NotifyError(format_http_error("media upload", result.status, result.body))


def attachment_from_upload(upload: dict[str, Any]) -> dict[str, Any]:
    attachment = {
        "id": upload.get("id"),
        "kind": upload_kind(upload),
        "url": upload.get("blob_url") or upload.get("image_url"),
        "mime_type": upload.get("mime_type") or upload.get("content_type") or "application/octet-stream",
        "size_bytes": int(upload.get("size_bytes") or 0),
    }
    if upload.get("duration_ms") is not None:
        attachment["duration_ms"] = int(upload["duration_ms"])
    if upload.get("filename"):
        attachment["file_name"] = upload["filename"]
    return {key: value for key, value in attachment.items() if value not in (None, "")}


def attach_uploaded_files(payload: dict[str, Any], uploads: list[dict[str, Any]]) -> None:
    """Map uploaded blobs onto the /messages multi-attachment contract.

    The full upload set is sent as `attachments`. For compatibility, the first
    upload also fills legacy `attachment`, the first image fills `image_url`, and
    the first non-image file fills `url`/`url_title` when the caller did not
    supply an explicit link. Local absolute paths are never added here.
    """
    if not uploads:
        return

    first_upload = uploads[0]
    first_image = next((item for item in uploads if upload_kind(item) == "image"), None)
    first_file = next((item for item in uploads if upload_kind(item) != "image"), None)

    attachments = [attachment for attachment in (attachment_from_upload(item) for item in uploads) if attachment.get("url")]
    if attachments:
        payload["attachments"] = attachments

    if not payload.get("attachment"):
        attachment = attachments[0] if attachments else attachment_from_upload(first_upload)
        if attachment.get("url"):
            payload["attachment"] = attachment

    if first_image and not payload.get("image_url"):
        payload["image_url"] = first_image.get("image_url") or first_image.get("blob_url")
    if first_file and not payload.get("url"):
        payload["url"] = first_file.get("blob_url") or first_file.get("image_url")
        payload.setdefault("url_title", first_file.get("filename") or "Open uploaded file")

    metadata = dict(payload.get("metadata") or {})
    metadata.setdefault("media_upload_count", str(len(uploads)))
    metadata.setdefault("media_upload_filenames", ", ".join(str(item.get("filename") or "attachment") for item in uploads)[:512])
    if len(uploads) > 1:
        metadata.setdefault("media_upload_extra_count", str(len(uploads) - 1))
    if first_image and first_file:
        metadata.setdefault("media_upload_kind", "mixed")
    elif first_image:
        metadata.setdefault("media_upload_kind", "image")
    else:
        metadata.setdefault("media_upload_kind", "file")
    payload["metadata"] = metadata


def send_once(endpoint: str, api_key: str, payload: dict[str, Any], timeout: float,
              *, user_agent: str = USER_AGENT) -> HTTPResult:
    body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        endpoint,
        data=body,
        headers={"Content-Type": "application/json", "X-API-Key": api_key, "User-Agent": user_agent},
        method="POST",
    )
    return open_json_request(req, timeout=timeout)


def send(endpoint: str, api_key: str, payload: dict[str, Any], timeout: float, max_retries: int) -> HTTPResult:
    return retry_request(lambda: send_once(endpoint, api_key, payload, timeout), max_retries,
                         stop_on_lifetime_quota=True)


def retry_request(attempt_request: Callable[[], HTTPResult], max_retries: int,
                  *, stop_on_lifetime_quota: bool = False) -> HTTPResult:
    last: HTTPResult | None = None
    for attempt in range(max_retries + 1):
        result = attempt_request()
        if 200 <= result.status < 300:
            return result
        last = result
        if result.status not in TRANSIENT_STATUSES or attempt >= max_retries:
            break
        if stop_on_lifetime_quota and result.status == 429 and isinstance(result.body, dict) and result.body.get("is_lifetime"):
            # Lifetime quota will not reset — retrying is futile.
            break
        time.sleep(retry_delay(attempt, result.headers))
    assert last is not None
    return last


def fetch_message_status(endpoint: str, api_key: str, timeout: float) -> HTTPResult:
    req = urllib.request.Request(
        endpoint,
        headers={"X-API-Key": api_key, "User-Agent": USER_AGENT},
        method="GET",
    )
    return open_json_request(req, timeout=timeout)


# --------------------------------------------------------------------------- #
# Asking a human
#
# The Lock Screen can only draw buttons whose titles the app registered at
# launch, so the server picks a "category" by matching an approval's ordered
# decision *keys*. Using one of the pairs below is what earns real verbs there
# instead of "Option 1" / "Option 2"; anything else is still answerable, just in
# the expanded view or the app.
#
# Mirrors `push.semanticPairs` in the Go backend. Order is part of the contract:
# a tap is reported by index, so a reversed pair deliberately does not match.

SEMANTIC_DECISION_PAIRS: dict[tuple[str, str], str] = {
    ("allow", "deny"): "AN_ALLOW_DENY",
    ("approve", "reject"): "AN_ALLOW_DENY",
    ("continue", "stop"): "AN_CONTINUE_STOP",
    ("yes", "no"): "AN_YES_NO",
    ("ok", "cancel"): "AN_YES_NO",
}

# Labels that map onto a semantic key, so `--choice Allow --choice Deny` earns
# real Lock Screen buttons without the caller having to know the vocabulary.
_LABEL_KEYS = {
    "allow": "allow", "allow once": "allow", "approve": "approve", "yes": "yes",
    "ok": "ok", "okay": "ok", "continue": "continue", "keep going": "continue",
    "deny": "deny", "reject": "reject", "no": "no", "cancel": "cancel",
    "stop": "stop",
}

ASK_TIMEOUT_EXIT = 124  # Matches timeout(1), so scripts can branch on it.
ASK_POLL_SECONDS = 2.0


def decision_key_for(label: str, index: int) -> str:
    """Maps a label onto a stable key, falling back to its position.

    The key is what the server matches to choose a category, and it is also what
    a caller reads back. Deriving it from the label keeps the common
    Allow/Deny case on the rich path without making every caller learn the
    vocabulary, while an unrecognised label still gets a usable identifier.
    """
    return _LABEL_KEYS.get(label.strip().lower(), f"choice_{index}")


def sessions_endpoint(messages_endpoint: str) -> str:
    return messages_endpoint.rsplit("/messages", 1)[0] + "/sessions"


def find_active_session(start: Path) -> str | None:
    """Reads the session sidecar the agent hooks write, if there is one.

    Walking up from the working directory matches how `steer` finds it, so a
    question asked from inside a live agent run is attributed to *that* run
    rather than opening a second, unrelated one.
    """
    current = start.resolve()
    for directory in [current, *current.parents]:
        sidecar = directory / ".agent-notifier" / "active-session.json"
        if not sidecar.is_file():
            continue
        try:
            data = json.loads(sidecar.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        session_id = str(data.get("server_session_id") or "").strip()
        if session_id:
            return session_id
    return None


def ask_session_label() -> str:
    """Names the run that is asking: "agent-notifier · master".

    This is the session's title on the phone and the difference between
    "something wants you" and "the deploy on main wants you" when several things
    can ask at once. The tool name stays out; the session's agent field says it.
    """
    parts = [Path.cwd().name or "notify.py"]
    try:
        branch = subprocess.run(
            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
            capture_output=True, text=True, timeout=2, check=False,
        ).stdout.strip()
        if branch and branch != "HEAD" and branch not in parts:
            parts.append(branch)
    except Exception:
        pass
    return " · ".join(parts)


def ask_session_cache_path() -> Path:
    return default_config_path().parent / "ask-sessions.json"


def cached_ask_session(project: str, cwd: Path) -> str | None:
    """Returns the session this project+directory asked from last time."""
    try:
        cache = json.loads(ask_session_cache_path().read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    entry = cache.get(f"{project}:{cwd}")
    if not isinstance(entry, dict):
        return None
    # Sessions the server has long since reaped should not be reused; a day is
    # far longer than any agent run and short enough to stay tidy.
    if time.time() - float(entry.get("created_at") or 0) > 86400:
        return None
    session_id = str(entry.get("session_id") or "").strip()
    return session_id or None


def remember_ask_session(project: str, cwd: Path, session_id: str) -> None:
    path = ask_session_cache_path()
    try:
        cache = json.loads(path.read_text(encoding="utf-8"))
        if not isinstance(cache, dict):
            cache = {}
    except (OSError, json.JSONDecodeError):
        cache = {}
    cache[f"{project}:{cwd}"] = {"session_id": session_id, "created_at": time.time()}
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(cache, indent=2, sort_keys=True), encoding="utf-8")
    except OSError:
        pass


def resolve_ask_session(args: argparse.Namespace, endpoint: str, api_key: str,
                        project: str, timeout: float) -> str:
    """Finds or creates the session a question belongs to.

    An approval must be attributable: the phone shows which run is asking, and
    with several agents live on one repo that is what makes an answer safe to
    give. A bare script has no session of its own, so one is created for it.

    It is then *reused*. Registering one per question left a new "notify.py"
    session behind on every ask, which piled up in the app's session list with
    no way to tell them apart or clear them out.
    """
    if args.session_id:
        return validate_uuid(args.session_id, "session_id")

    existing = find_active_session(Path.cwd())
    if existing:
        return existing

    cached = cached_ask_session(project, Path.cwd())
    if cached:
        return cached

    result = send_once(sessions_endpoint(endpoint), api_key, {
        "project": project, "label": ask_session_label(), "agent": "notify.py",
    }, timeout, user_agent=f"notify.py/{__version__}")
    if not (200 <= result.status < 300) or not isinstance(result.body, dict):
        raise NotifyError(format_http_error("create session", result.status, result.body))
    session_id = str(result.body.get("id") or "")
    if not session_id:
        raise NotifyError("server created a session but returned no id")
    remember_ask_session(project, Path.cwd(), session_id)
    return session_id


def ask_title(question: str) -> str:
    """The push title for a question: its first line, at most 120 characters.

    A generic title ("Agent needs you") hid the question on the Lock Screen and
    in the app, which then showed every request as "Alert".
    """
    first = next((line.strip() for line in question.splitlines() if line.strip()), "") or "Agent needs you"
    return first if len(first) <= 120 else first[:119].rstrip() + "…"


def build_ask_payload(args: argparse.Namespace, project: str, session_id: str, client_event_id: str | None) -> dict[str, Any]:
    """Builds the POST /approvals body for a blocking question."""
    labels = [c.strip() for c in (args.choice or []) if c.strip()]
    free_form = bool(args.ask_text) or bool(args.allow_free_form)

    if not labels and not free_form:
        raise NotifyError("--ask needs at least one --choice, or use --ask-text")

    decisions = [
        {"key": decision_key_for(label, index), "label": label}
        for index, label in enumerate(labels)
    ]
    # An unanswered question must resolve to *something*, or the agent blocks
    # until its own deadline with no signal. The last option is the safe default
    # by convention: refusals and cancels are conventionally listed last.
    if decisions:
        decisions[-1]["is_default_on_timeout"] = True

    metadata: dict[str, str] = dict(parse_metadata(args.metadata))
    metadata["hitl_kind"] = "question" if labels else "elicitation"
    if free_form and labels:
        # Read by the server to keep a text field when a device cannot draw the
        # options, and by the clients to offer typing alongside them.
        metadata["allow_free_form"] = "true"

    payload: dict[str, Any] = {
        "project": project,
        "title": args.title or ask_title(args.ask or args.ask_text or args.message or ""),
        "message": args.message or args.ask or args.ask_text or "",
        "type": "alert",
        "priority": 1,
        "response_format": "answers" if labels else "text",
        "wait": "none",
        "timeout_seconds": max(10, min(int(args.ask_timeout), 3600)),
        "decisions": decisions or [{"key": "reply", "label": "Reply"}],
        "metadata": metadata,
    }
    payload["session_id"] = session_id
    if client_event_id:
        payload["client_event_id"] = client_event_id
    return payload


def approvals_endpoint(messages_endpoint: str) -> str:
    return messages_endpoint.rsplit("/messages", 1)[0] + "/approvals"


def action_responses_endpoint(messages_endpoint: str) -> str:
    return messages_endpoint.rsplit("/messages", 1)[0] + "/action-responses"


def poll_for_answer(
    endpoint: str,
    api_key: str,
    notification_id: str,
    project: str,
    deadline: float,
    timeout: float,
) -> dict[str, Any] | None:
    """Waits for the human's answer, or returns None at the deadline.

    Transient failures are retried rather than surfaced: a blip while waiting is
    a reason to keep waiting, not to abandon a question the user may already
    have answered on their phone.
    """
    query = urllib.parse.urlencode({"notification_id": notification_id, "project": project})
    url = f"{endpoint}?{query}"
    while time.time() < deadline:
        try:
            request = urllib.request.Request(url, headers={"X-API-Key": api_key})
            result = open_json_request(request, timeout)
            if 200 <= result.status < 300 and isinstance(result.body, dict):
                items = result.body.get("items") or []
                if items:
                    return items[0]
        except Exception:
            pass
        time.sleep(ASK_POLL_SECONDS)
    return None


def ack_answer(endpoint: str, api_key: str, action_id: str, timeout: float) -> None:
    """Consumes the answer so it cannot be served to a later question."""
    try:
        url = f"{endpoint}/{urllib.parse.quote(action_id, safe='')}/ack"
        request = urllib.request.Request(url, data=b"{}", method="POST",
                                         headers={"X-API-Key": api_key, "Content-Type": "application/json"})
        open_json_request(request, timeout)
    except Exception:
        pass


def run_ask(args: argparse.Namespace, endpoint: str, api_key: str, project: str) -> int:
    """Asks a human and blocks until they answer.

    Prints the answer on stdout so it can be captured directly:

        answer=$(notify.py --ask "Deploy to prod?" -c "Ship it" -c "Hold")

    The exit code carries the choice index so a script can branch without
    string-matching, and ASK_TIMEOUT_EXIT (124, as timeout(1) uses) means
    nobody answered.
    """
    client_event_id = resolve_client_event_id(args)
    session_id = "00000000-0000-0000-0000-000000000000" if args.dry_run else resolve_ask_session(
        args, endpoint, api_key, project, args.timeout)
    payload = build_ask_payload(args, project, session_id, client_event_id)

    if args.dry_run:
        print(json.dumps({"endpoint": approvals_endpoint(endpoint), "payload": payload},
                         indent=2, ensure_ascii=False))
        return 0

    result = send_once(approvals_endpoint(endpoint), api_key, payload, args.timeout,
                       user_agent=f"notify.py/{__version__}")
    if not (200 <= result.status < 300) or not isinstance(result.body, dict):
        raise NotifyError(format_http_error("ask", result.status, result.body))

    notification_id = str(result.body.get("notification_id") or "")
    if not notification_id:
        raise NotifyError("server accepted the question but returned no notification id")

    deadline = time.time() + payload["timeout_seconds"]
    answer = poll_for_answer(action_responses_endpoint(endpoint), api_key,
                             notification_id, project, deadline, args.timeout)
    if answer is None:
        if not args.quiet:
            print("no answer before the deadline", file=sys.stderr)
        return ASK_TIMEOUT_EXIT

    if answer.get("id"):
        ack_answer(action_responses_endpoint(endpoint), api_key, str(answer["id"]), args.timeout)

    text = str(answer.get("response_text") or answer.get("button_label") or "").strip()
    print(text)
    # Index-based exit codes let a caller branch on *which* option was picked
    # without parsing prose. 0 is the first choice, which is the affirmative one
    # by convention.
    index = answer.get("button_index")
    if isinstance(index, int) and 0 <= index < 100:
        return index
    return 0


def build_endpoints(base_url: str, api_path: str) -> tuple[str, str]:
    parsed = urllib.parse.urlparse(base_url)
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise NotifyError("base URL must be an HTTP(S) URL")
    base = base_url.rstrip("/")
    path = "/" + api_path.strip().strip("/")
    messages_endpoint = f"{base}{path}/messages"
    return messages_endpoint, media_endpoint_from_messages(messages_endpoint)


def message_status_endpoint(messages_endpoint: str, message_id: str) -> str:
    return f"{messages_endpoint}/{urllib.parse.quote(message_id.strip(), safe='')}/status"


def print_dry_run(endpoint: str, media_endpoint: str, payload: dict[str, Any], args: argparse.Namespace) -> None:
    upload_plan = [
        {
            "path": str(Path(raw).expanduser()),
            "filename": Path(raw).expanduser().name or "attachment",
            "local_path_sent": bool(args.include_local_paths),
        }
        for raw in args.file_paths
    ]
    if args.json:
        print(json.dumps({"messages_endpoint": endpoint, "media_endpoint": media_endpoint, "payload": payload, "uploads": upload_plan}, indent=2, ensure_ascii=False))
        return
    print(f"POST {endpoint}")
    if args.file_paths:
        print(f"UPLOAD {media_endpoint}")
        print(json.dumps({"files": upload_plan}, indent=2, ensure_ascii=False))
        if not args.include_local_paths:
            print("NOTE local upload paths are not included in payload.file_paths unless --include-local-paths is set", file=sys.stderr)
    print(json.dumps(payload, indent=2, ensure_ascii=False))


def warn_multiple_uploads(args: argparse.Namespace) -> None:
    if len(args.file_paths) <= 1 or args.quiet or args.json or args.id_only:
        return
    print(
        "warning: sending multiple uploads as attachments; older clients may only render the primary attachment.",
        file=sys.stderr,
    )


def print_success(args: argparse.Namespace, body: dict[str, Any] | str, status: int) -> None:
    if args.quiet:
        return
    if args.id_only:
        if isinstance(body, dict) and body.get("id"):
            print(body["id"])
        return
    if args.json:
        print(json.dumps(body, indent=2, ensure_ascii=False) if isinstance(body, dict) else json.dumps({"status": status, "body": body}, indent=2))
        return
    if isinstance(body, dict):
        msg_id = body.get("id", "")
        project = body.get("project_name", "")
        suffix = f" project={project}" if project else ""
        print(f"sent ok status={status} id={msg_id}{suffix}")
    else:
        print(f"sent ok status={status}")


def print_status_result(args: argparse.Namespace, body: dict[str, Any] | str, status: int) -> None:
    if args.quiet:
        return
    if args.json:
        print(json.dumps(body, indent=2, ensure_ascii=False) if isinstance(body, dict) else json.dumps({"status": status, "body": body}, indent=2))
        return
    if isinstance(body, dict):
        msg_id = body.get("id", "")
        audio_status = body.get("audio_status") or "none"
        attachment_count = body.get("attachments_count", 0)
        print(f"status ok http={status} id={msg_id} audio_status={audio_status} attachments={attachment_count}")
    else:
        print(f"status ok http={status}")


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    try:
        config_path = Path(args.config).expanduser() if args.config else default_config_path()
        config = load_config(config_path)
        local_env = load_local_env()

        cli_api_key = read_api_key_from_stdin() if args.api_key_stdin else args.api_key
        api_key = pick_setting(cli_api_key, os.environ.get("AGENT_NOTIFIER_API_KEY"), local_env.get("AGENT_NOTIFIER_API_KEY"), config.get("api_key"))
        base_url = pick_setting(args.base_url, os.environ.get("AGENT_NOTIFIER_URL"), local_env.get("AGENT_NOTIFIER_URL"), config.get("base_url"), default=DEFAULT_BASE_URL)
        api_path = pick_setting(args.api_path, os.environ.get("AGENT_NOTIFIER_API_PATH"), local_env.get("AGENT_NOTIFIER_API_PATH"), config.get("api_path"), default=DEFAULT_API_PATH)
        # Kept separate from `project` below: only an explicitly chosen project
        # may be persisted by --save-config. Persisting an auto-detected one
        # would pin every future notification to whatever directory the setup
        # command happened to run in.
        chosen_project = pick_setting(args.project, os.environ.get("AGENT_NOTIFIER_PROJECT"), local_env.get("AGENT_NOTIFIER_PROJECT"), config.get("project"))
        project = resolve_project(args.project, local_env, config)

        if args.max_retries < 0:
            raise NotifyError("--max-retries must be non-negative")
        if args.timeout <= 0:
            raise NotifyError("--timeout must be positive")

        if args.save_config:
            save_config(
                config_path,
                {
                    "api_key": api_key or "",
                    "project": chosen_project or "",
                    "base_url": base_url or DEFAULT_BASE_URL,
                    "api_path": api_path or DEFAULT_API_PATH,
                },
            )
            if not args.quiet:
                print(f"saved config to {config_path}")
            return 0

        endpoint, media_endpoint = build_endpoints(base_url or DEFAULT_BASE_URL, api_path or DEFAULT_API_PATH)

        if args.check_status:
            status_endpoint = message_status_endpoint(endpoint, args.check_status)
            if args.dry_run:
                if args.json:
                    print(json.dumps({"status_endpoint": status_endpoint}, indent=2, ensure_ascii=False))
                else:
                    print(f"GET {status_endpoint}")
                return 0
            if not api_key:
                raise NotifyError("AGENT_NOTIFIER_API_KEY is required. Set the env var, add it to .env.local, or run --save-config --api-key <key>.")
            result = fetch_message_status(status_endpoint, api_key, args.timeout)
            if 200 <= result.status < 300:
                print_status_result(args, result.body, result.status)
                return 0
            raise NotifyError(format_http_error("message status", result.status, result.body))

        if args.ask or args.ask_text:
            if not api_key:
                raise NotifyError("AGENT_NOTIFIER_API_KEY is required. Set the env var, add it to .env.local, or run --save-config --api-key <key>.")
            return run_ask(args, endpoint, api_key, project)

        if not args.message.strip() and not (args.as_audio and args.file_paths):
            raise NotifyError("message is required (positional argument)")

        client_event_id = resolve_client_event_id(args)
        payload = build_payload(args, project, client_event_id)

        if args.dry_run:
            print_dry_run(endpoint, media_endpoint, payload, args)
            return 0

        if not api_key:
            raise NotifyError("AGENT_NOTIFIER_API_KEY is required. Set the env var, add it to .env.local, or run --save-config --api-key <key>.")

        warn_multiple_uploads(args)
        uploads = [upload_file(media_endpoint, api_key, raw_path, args.timeout, args.max_retries) for raw_path in args.file_paths]
        attach_uploaded_files(payload, uploads)

        send_retries = args.max_retries if payload.get("client_event_id") else 0
        result = send(endpoint, api_key, payload, args.timeout, send_retries)
        if 200 <= result.status < 300:
            print_success(args, result.body, result.status)
            return 0
        raise NotifyError(format_http_error("message send", result.status, result.body))
    except NotifyError as exc:
        print(f"notify.py: {exc}", file=sys.stderr)
        return 1


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