Matrix Communication

{
  "skill_name": "matrix-communication",
  "evals": [
    {
      "id": 1,
      "eval_name": "send-basic-message",
      "prompt": "Send a message to the dev room on Matrix saying 'Deployment complete for v2.3.0'.",
      "expected_output": "Uses matrix-send-e2ee.py with room name and message text.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred over non-E2EE)",
        "Uses uv run to execute the script",
        "Passes 'dev' as room identifier",
        "Passes message text as second argument"
      ]
    },
    {
      "id": 2,
      "eval_name": "send-with-exclamation",
      "prompt": "Send 'Deploy complete! All tests passed!' to the ops room.",
      "expected_output": "Handles bash history expansion by using set +H before the command.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred)",
        "Prepends set +H && before the command",
        "Does not let bash history expansion corrupt the message"
      ]
    },
    {
      "id": 3,
      "eval_name": "send-emote",
      "prompt": "Send an emote message to the ops room saying 'is deploying to production'.",
      "expected_output": "Uses matrix-send-e2ee.py with --emote flag.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred)",
        "Uses --emote flag",
        "Passes 'ops' as room identifier",
        "Passes 'is deploying to production' as message"
      ]
    },
    {
      "id": 4,
      "eval_name": "send-no-prefix",
      "prompt": "Send a plain message to the team room without the bot prefix: 'Meeting moved to 3pm'.",
      "expected_output": "Uses matrix-send-e2ee.py with --no-prefix flag.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred)",
        "Uses --no-prefix flag",
        "Passes room identifier and message"
      ]
    },
    {
      "id": 5,
      "eval_name": "send-to-room-alias",
      "prompt": "Send 'Hello everyone' to the room #team:matrix.org.",
      "expected_output": "Uses matrix-send-e2ee.py with full room alias as ROOM argument.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred)",
        "Passes '#team:matrix.org' as room argument",
        "Passes 'Hello everyone' as message"
      ]
    },
    {
      "id": 6,
      "eval_name": "send-formatted-message",
      "prompt": "Send a deployment notification to the ops room: bold 'Deployed' header, version 1.5.0, with a Jira link to PROJ-456.",
      "expected_output": "Uses matrix-send-e2ee.py with markdown-formatted message.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred)",
        "Uses markdown bold for 'Deployed'",
        "Includes version information",
        "Includes Jira URL"
      ]
    },
    {
      "id": 7,
      "eval_name": "thread-reply",
      "prompt": "Reply in the thread started by event $root456 in the dev room with 'Tests passing, ready for review'.",
      "expected_output": "Uses matrix-send-e2ee.py with --thread flag and root event ID.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred)",
        "Uses --thread flag with '$root456'",
        "Passes 'dev' as room identifier",
        "Passes message text as argument"
      ]
    },
    {
      "id": 8,
      "eval_name": "direct-reply",
      "prompt": "Reply to the message $msg123 in the dev room saying 'Agreed, let's proceed with that approach'.",
      "expected_output": "Uses matrix-send-e2ee.py with --reply flag.",
      "files": [],
      "assertions": [
        "Uses matrix-send-e2ee.py (E2EE preferred)",
        "Uses --reply flag (not --thread) with '$msg123'",
        "Passes 'dev' as room identifier"
      ]
    },
    {
      "id": 9,
      "eval_name": "read-recent-messages",
      "prompt": "Read the last 5 messages from the agent-work Matrix room.",
      "expected_output": "Uses matrix-read-e2ee.py with room name and --limit flag.",
      "files": [],
      "assertions": [
        "Uses matrix-read-e2ee.py (E2EE preferred)",
        "Passes 'agent-work' as room identifier",
        "Uses --limit 5 flag"
      ]
    },
    {
      "id": 10,
      "eval_name": "read-messages-json",
      "prompt": "Read the last 20 messages from the ops room as JSON so I can analyze them.",
      "expected_output": "Uses matrix-read-e2ee.py with --json and --limit flags.",
      "files": [],
      "assertions": [
        "Uses matrix-read-e2ee.py (E2EE preferred)",
        "Uses --json flag for structured output",
        "Uses --limit 20 flag"
      ]
    },
    {
      "id": 11,
      "eval_name": "read-with-request-keys",
      "prompt": "Read the last 10 messages from the dev room and try to recover any undecryptable messages.",
      "expected_output": "Uses matrix-read-e2ee.py with --request-keys flag.",
      "files": [],
      "assertions": [
        "Uses matrix-read-e2ee.py with --request-keys flag",
        "Uses --limit 10 flag"
      ]
    },
    {
      "id": 12,
      "eval_name": "edit-message",
      "prompt": "Edit the message $evt789 in the general room to say 'Corrected: deploy at 3pm not 2pm'.",
      "expected_output": "Uses matrix-edit-e2ee.py with room, event ID, and new text.",
      "files": [],
      "assertions": [
        "Uses matrix-edit-e2ee.py (E2EE preferred over matrix-edit.py)",
        "Passes room identifier as first argument",
        "Passes event ID '$evt789' as second argument",
        "Passes corrected text as third argument"
      ]
    },
    {
      "id": 13,
      "eval_name": "delete-message",
      "prompt": "Delete the message $evt999 from the dev room because it was sent by mistake.",
      "expected_output": "Uses matrix-redact.py with room, event ID, and reason.",
      "files": [],
      "assertions": [
        "Uses matrix-redact.py script",
        "Passes room identifier",
        "Passes event ID '$evt999'",
        "Optionally passes a reason string"
      ]
    },
    {
      "id": 14,
      "eval_name": "react-to-message",
      "prompt": "React to the message with event ID $abc123 in the ops room with a checkmark emoji.",
      "expected_output": "Uses matrix-react.py with room name, event ID, and emoji.",
      "files": [],
      "assertions": [
        "Uses matrix-react.py script",
        "Passes 'ops' as room identifier",
        "Passes '$abc123' as event ID",
        "Passes checkmark emoji as reaction"
      ]
    },
    {
      "id": 15,
      "eval_name": "list-rooms",
      "prompt": "List my Matrix rooms so I can see which ones I'm joined to.",
      "expected_output": "Uses matrix-rooms.py to list joined rooms.",
      "files": [],
      "assertions": [
        "Uses matrix-rooms.py script",
        "Uses uv run to execute",
        "Does not call Matrix API directly"
      ]
    },
    {
      "id": 16,
      "eval_name": "search-rooms",
      "prompt": "Search my Matrix rooms for any room with 'ops' in the name.",
      "expected_output": "Uses matrix-rooms.py with --search flag.",
      "files": [],
      "assertions": [
        "Uses matrix-rooms.py with --search flag",
        "Passes 'ops' as search term"
      ]
    },
    {
      "id": 17,
      "eval_name": "resolve-room-alias",
      "prompt": "What is the room ID for #general:matrix.example.com?",
      "expected_output": "Uses matrix-resolve.py to resolve the alias.",
      "files": [],
      "assertions": [
        "Uses matrix-resolve.py script",
        "Passes '#general:matrix.example.com' as argument"
      ]
    },
    {
      "id": 18,
      "eval_name": "health-check",
      "prompt": "Check if my Matrix skill dependencies are properly installed.",
      "expected_output": "Uses matrix-doctor.py with python3 (not uv run).",
      "files": [],
      "assertions": [
        "Uses matrix-doctor.py script",
        "Uses python3 (not uv run) to execute",
        "Optionally uses --install flag"
      ]
    },
    {
      "id": 19,
      "eval_name": "e2ee-setup-status",
      "prompt": "Check if my E2EE device is properly set up for Matrix.",
      "expected_output": "Uses matrix-e2ee-setup.py with --status flag.",
      "files": [],
      "assertions": [
        "Uses matrix-e2ee-setup.py with --status flag",
        "Uses uv run to execute",
        "Does not create a new device (only checks status)"
      ]
    },
    {
      "id": 20,
      "eval_name": "e2ee-setup-new",
      "prompt": "I need to set up Matrix E2EE for the first time. My password has special characters.",
      "expected_output": "Uses MATRIX_PASSWORD env var with matrix-e2ee-setup.py.",
      "files": [],
      "assertions": [
        "Uses matrix-e2ee-setup.py script",
        "Uses MATRIX_PASSWORD environment variable (not command-line arg)",
        "References setup-guide.md or explains env var avoids shell escaping"
      ]
    },
    {
      "id": 21,
      "eval_name": "device-verification",
      "prompt": "I need to verify my Matrix E2EE device with Element Desktop.",
      "expected_output": "Uses matrix-e2ee-verify.py with --timeout flag.",
      "files": [],
      "assertions": [
        "Uses matrix-e2ee-verify.py script",
        "Uses --timeout flag with reasonable value",
        "Mentions Element Desktop (not Element X)",
        "References e2ee-guide.md for details"
      ]
    },
    {
      "id": 22,
      "eval_name": "restore-key-backup",
      "prompt": "I have a recovery key 'EsTj qRGp YB4C abcd'. Restore my key backup so I can read old messages.",
      "expected_output": "Uses matrix-key-backup.py with --recovery-key and --import-keys.",
      "files": [],
      "assertions": [
        "Uses matrix-key-backup.py script",
        "Uses --recovery-key flag with the provided key",
        "Uses --import-keys flag (must not forget this)",
        "Does not omit --import-keys"
      ]
    },
    {
      "id": 23,
      "eval_name": "fetch-missing-keys",
      "prompt": "I'm seeing '[Unable to decrypt]' for some messages in the IT room. Can you try fetching the missing keys?",
      "expected_output": "Uses matrix-fetch-keys.py to request keys from other devices.",
      "files": [],
      "assertions": [
        "Uses matrix-fetch-keys.py script",
        "Passes 'IT' as room identifier",
        "Uses --sync-time flag with reasonable timeout"
      ]
    },
    {
      "id": 24,
      "eval_name": "error-recovery-forbidden",
      "prompt": "I tried to send a message to the ops room but got M_FORBIDDEN. What should I do?",
      "expected_output": "Advises joining the room in Element first.",
      "files": [],
      "assertions": [
        "Suggests joining the room in Element first",
        "Does not retry the same command blindly",
        "References error handling guidance"
      ]
    },
    {
      "id": 25,
      "eval_name": "error-recovery-decrypt",
      "prompt": "I'm reading messages in the dev room and most show '[Unable to decrypt]'. How do I fix this?",
      "expected_output": "Suggests key backup restore or key fetching, not non-E2EE fallback.",
      "files": [],
      "assertions": [
        "Suggests matrix-key-backup.py with --recovery-key and --import-keys, or matrix-fetch-keys.py",
        "Does not suggest non-E2EE fallback as primary solution"
      ]
    }
  ]
}

"""Matrix Skill shared library.

This module provides common functionality for all Matrix scripts.
All modules use ONLY stdlib to ensure non-E2EE scripts work without dependencies.

Usage:
    # At the top of each script, add:
    import sys, os
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

    # Then import what you need:
    from _lib import load_config, matrix_request, find_room_by_name
"""

# Config
from _lib.config import get_config_path, load_config

# HTTP API
from _lib.http import matrix_request

# Room operations
from _lib.rooms import (
    resolve_room_alias,
    get_room_info,
    list_joined_rooms,
    find_room_by_name,
    find_room_in_nio_client,
)

# Formatting
from _lib.formatting import (
    shorten_service_urls,
    markdown_to_html,
    add_bot_prefix,
)

# Utilities
from _lib.utils import (
    clean_message,
    format_timestamp,
    prefer_ipv4,
    suppress_nio_logging,
)

# Dependency checking
from _lib.deps import check_e2ee_dependencies

# E2EE (only used by E2EE scripts, but still stdlib-only)
from _lib.e2ee import (
    get_store_path,
    get_credentials_path,
    load_credentials,
    save_credentials,
    delete_credentials,
)

__all__ = [
    # Config
    "get_config_path",
    "load_config",
    # HTTP
    "matrix_request",
    # Rooms
    "resolve_room_alias",
    "get_room_info",
    "list_joined_rooms",
    "find_room_by_name",
    "find_room_in_nio_client",
    # Formatting
    "shorten_service_urls",
    "markdown_to_html",
    "add_bot_prefix",
    # Utils
    "clean_message",
    "format_timestamp",
    "prefer_ipv4",
    "suppress_nio_logging",
    # Deps
    "check_e2ee_dependencies",
    # E2EE
    "get_store_path",
    "get_credentials_path",
    "load_credentials",
    "save_credentials",
    "delete_credentials",
]

"""Configuration loading for Matrix scripts.

All functions use ONLY stdlib.
"""

import json
import os
import sys
from pathlib import Path


def get_config_path() -> Path:
    """Get the Matrix configuration file path.

    Returns ~/.config/matrix/config.json (respects XDG_CONFIG_HOME if set).
    """
    xdg_config = os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")
    return Path(xdg_config) / "matrix" / "config.json"


def load_config(require_user_id: bool = False) -> dict:
    """Load Matrix config from ~/.config/matrix/config.json.

    Args:
        require_user_id: If True, require user_id field (for E2EE scripts)

    Returns:
        dict with homeserver, access_token, and optionally user_id, bot_prefix

    Exits with error if config not found or missing required fields.
    """
    config_path = get_config_path()
    if not config_path.exists():
        print(f"Error: Config file not found: {config_path}", file=sys.stderr)
        print("Create it with:", file=sys.stderr)
        example = {
            "homeserver": "https://matrix.org",
            "access_token": "syt_...",
        }
        if require_user_id:
            example["user_id"] = "@user:matrix.org"
        print(json.dumps(example, indent=2), file=sys.stderr)
        sys.exit(1)

    with open(config_path) as f:
        config = json.load(f)

    # Validate required fields
    required = ["homeserver"]
    if require_user_id:
        required.append("user_id")
    else:
        required.append("access_token")

    missing = [f for f in required if f not in config]
    if missing:
        print(
            f"Error: Config missing required fields: {', '.join(missing)}",
            file=sys.stderr,
        )
        sys.exit(1)

    return config

"""Dependency checking for E2EE scripts.

All functions use ONLY stdlib.
"""

import os
import sys


def check_e2ee_dependencies() -> None:
    """Check that matrix-nio[e2e] dependencies are available.

    Prints helpful installation instructions and exits with code 1
    if dependencies are missing. Call this before importing nio.
    """
    try:
        from nio import AsyncClient  # noqa: F401
    except ImportError as e:
        error_msg = str(e).lower()
        script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

        if "olm" in error_msg:
            print("Error: libolm library not found.", file=sys.stderr)
            print("", file=sys.stderr)
            print("Install libolm for your platform:", file=sys.stderr)
            print("  Debian/Ubuntu: sudo apt install libolm-dev", file=sys.stderr)
            print("  Fedora:        sudo dnf install libolm-devel", file=sys.stderr)
            print("  macOS 26+:     brew won't help; python-olm", file=sys.stderr)
            print("    build fails (Clang 17), see setup-guide.md.", file=sys.stderr)
        elif "nio" in error_msg or "matrix" in error_msg:
            print("Error: matrix-nio library not found.", file=sys.stderr)
            print("", file=sys.stderr)
            print("Install with (try in order):", file=sys.stderr)
            print("  uvx pip install 'matrix-nio[e2e]'", file=sys.stderr)
            print("  pip install 'matrix-nio[e2e]'", file=sys.stderr)
            print("  pip3 install 'matrix-nio[e2e]'", file=sys.stderr)
        else:
            print(f"Error: Missing dependency: {e}", file=sys.stderr)

        print("", file=sys.stderr)
        print("Or run the health check to diagnose and fix:", file=sys.stderr)
        print(f"  python3 {script_dir}/matrix-doctor.py --install", file=sys.stderr)
        sys.exit(1)

"""E2EE credential management for Matrix scripts.

All functions use ONLY stdlib - no nio dependencies here.
The actual E2EE functionality (using nio) is in the scripts themselves.
"""

import json
import os
from pathlib import Path


def get_store_path() -> Path:
    """Get or create the E2EE key store directory.

    Uses XDG_DATA_HOME or falls back to ~/.local/share/matrix-skill/store
    """
    xdg_data = os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share")
    store_path = Path(xdg_data) / "matrix-skill" / "store"
    store_path.mkdir(parents=True, exist_ok=True)
    return store_path


def get_credentials_path() -> Path:
    """Get path for stored E2EE device credentials."""
    return get_store_path() / "credentials.json"


def load_credentials() -> dict | None:
    """Load stored device credentials if they exist.

    Returns:
        Dict with user_id, device_id, access_token, or None if not found
    """
    creds_path = get_credentials_path()
    if creds_path.exists():
        with open(creds_path) as f:
            return json.load(f)
    return None


def save_credentials(user_id: str, device_id: str, access_token: str):
    """Save device credentials for future use.

    Credentials file is chmod 600 for security.
    """
    creds_path = get_credentials_path()
    with open(creds_path, "w") as f:
        json.dump(
            {
                "user_id": user_id,
                "device_id": device_id,
                "access_token": access_token,
            },
            f,
            indent=2,
        )
    os.chmod(creds_path, 0o600)


def delete_credentials():
    """Remove stored device credentials and key store files."""
    creds_path = get_credentials_path()
    if creds_path.exists():
        creds_path.unlink()

    # Also remove key store databases
    store_path = get_store_path()
    for db_file in store_path.glob("*.db"):
        db_file.unlink()
    for key_file in store_path.glob("*_devices"):
        key_file.unlink()

"""Message formatting for Matrix scripts.

All functions use ONLY stdlib.
"""

import re


def shorten_service_urls(text: str) -> str:
    """Convert service URLs to shorter linked text.

    Supported services:
    - Jira: https://jira.example.com/browse/PROJ-123 -> [PROJ-123](url)
    - GitHub Issues/PRs: https://github.com/owner/repo/issues/123 -> [owner/repo#123](url)
    - GitHub commits: https://github.com/owner/repo/commit/abc123 -> [owner/repo@abc123](url)
    - GitLab Issues/MRs: https://gitlab.example.com/group/project/-/issues/123 -> [group/project#123](url)

    URLs already inside `[text](url)` markdown links are left untouched so the
    caller's chosen link text is preserved (and to avoid producing broken
    nested `[text]([short](url))` shapes).
    """
    # Protect existing markdown links from being re-wrapped.
    protected: list[str] = []

    def _protect(match: "re.Match[str]") -> str:
        idx = len(protected)
        protected.append(match.group(0))
        return f"\x00MDLINK{idx}\x00"

    text = re.sub(r"\[[^\]]+\]\([^)]+\)", _protect, text)

    # Jira URLs: https://jira.*/browse/PROJ-123 or https://*.atlassian.net/browse/PROJ-123
    text = re.sub(
        r"https?://[^/]+/browse/([A-Z][A-Z0-9]+-\d+)", r"[\1](https://\g<0>)", text
    )
    # Fix double https
    text = re.sub(r"\(https://https?://", r"(https://", text)

    # GitHub Issues/PRs: https://github.com/owner/repo/issues/123 or /pull/123
    text = re.sub(
        r"https?://github\.com/([^/]+)/([^/]+)/(issues|pull)/(\d+)",
        r"[\1/\2#\4](\g<0>)",
        text,
    )

    # GitHub commits: https://github.com/owner/repo/commit/abc123...
    text = re.sub(
        r"https?://github\.com/([^/]+)/([^/]+)/commit/([a-f0-9]{7,40})",
        r"[\1/\2@\3](\g<0>)",
        text,
    )

    # GitLab Issues/MRs: https://gitlab.*/group/project/-/issues/123 or /-/merge_requests/123
    text = re.sub(
        r"https?://[^/]+/([^/]+/[^/]+)/-/(issues|merge_requests)/(\d+)",
        r"[\1#\3](\g<0>)",
        text,
    )

    # Restore protected markdown links. Bounds-check the index so a forged
    # placeholder in user input can't raise IndexError — unknown placeholders
    # are left as-is.
    def _restore(match: "re.Match[str]") -> str:
        idx = int(match.group(1))
        if 0 <= idx < len(protected):
            return protected[idx]
        return match.group(0)

    text = re.sub(r"\x00MDLINK(\d+)\x00", _restore, text)

    return text


def markdown_to_html(text: str) -> str:
    """Convert markdown to Matrix HTML with smart features.

    Supports:
    - ## headings (h1-h6)
    - **bold**, *italic*, `code`, ~~strikethrough~~
    - [text](url) links
    - ||spoiler|| text (Discord-style)
    - ```lang code blocks ```
    - > blockquotes
    - list items: `- item`, `* item`, `+ item`
    - | table | rows |
    - @user:server mentions (clickable pills)
    - #room:server room links (clickable)
    - Auto-shortens Jira, GitHub, GitLab URLs
    """
    # First, shorten service URLs (before other processing)
    html = shorten_service_urls(text)

    # Extract and protect code blocks from other processing
    code_blocks = []

    def save_code_block(match):
        lang = match.group(1) or ""
        code = match.group(2)
        idx = len(code_blocks)
        if lang:
            code_blocks.append(
                f'<pre><code class="language-{lang}">{code}</code></pre>'
            )
        else:
            code_blocks.append(f"<pre><code>{code}</code></pre>")
        return f"{{{{CODEBLOCK_{idx}}}}}"

    html = re.sub(r"```(\w*)\n(.*?)```", save_code_block, html, flags=re.DOTALL)

    # Spoilers: ||text|| -> <span data-mx-spoiler>text</span>
    # But not table separators - check for pipe at start/end of line
    html = re.sub(
        r"(?<!\|)\|\|(.+?)\|\|(?!\|)", r"<span data-mx-spoiler>\1</span>", html
    )

    # Markdown links: [text](url) -> <a href="url">text</a>
    html = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r'<a href="\2">\1</a>', html)

    # Matrix user mentions: @user:server -> clickable pill
    # Only match if not already inside a link
    html = re.sub(
        r'(?<!["\'/])(@[a-zA-Z0-9._=-]+:[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})',
        r'<a href="https://matrix.to/#/\1">\1</a>',
        html,
    )

    # Matrix room links: #room:server -> clickable link
    # Only match if not already inside a link
    html = re.sub(
        r'(?<!["\'/])(#[a-zA-Z0-9._=-]+:[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})',
        r'<a href="https://matrix.to/#/\1">\1</a>',
        html,
    )

    # Emphasis runs follow CommonMark left/right-flanking: the opening
    # delimiter must NOT be followed by whitespace, and the closing one
    # must NOT be preceded by whitespace. Without that, `* item` (a bullet)
    # would be parsed as an italic opener and chew up the line until the
    # next stray `*`.

    # Strikethrough: ~~text~~ -> <del>text</del>
    html = re.sub(r"~~(?=\S)(.+?)(?<=\S)~~", r"<del>\1</del>", html)

    # Bold: **text** -> <strong>text</strong>
    html = re.sub(r"\*\*(?=\S)(.+?)(?<=\S)\*\*", r"<strong>\1</strong>", html)

    # Italic: *text* -> <em>text</em>
    html = re.sub(r"\*(?=\S)([^*\n]+?)(?<=\S)\*", r"<em>\1</em>", html)

    # Inline code: `text` -> <code>text</code>
    html = re.sub(r"`(.+?)`", r"<code>\1</code>", html)

    # Normalize multiple newlines
    html = re.sub(r"\n{2,}", "\n", html)

    # Process line-based formatting (headings, lists, blockquotes, tables)
    lines = html.split("\n")
    in_list = False
    in_quote = False
    in_table = False
    result = []

    for line in lines:
        stripped = line.strip()

        # Headings: ## Heading -> <h2>
        heading_match = re.match(r"^(#{1,6})\s+(.+)$", stripped)
        if heading_match:
            if in_quote:
                result.append("</blockquote>")
                in_quote = False
            if in_list:
                result.append("</ul>")
                in_list = False
            if in_table:
                result.append("</table>")
                in_table = False
            level = len(heading_match.group(1))
            heading_text = heading_match.group(2)
            result.append(f"<h{level}>{heading_text}</h{level}>")
            continue

        # Tables: | col | col |
        if stripped.startswith("|") and stripped.endswith("|"):
            # Parse table cells
            cells = [c.strip() for c in stripped.split("|")[1:-1]]

            # Check if this is a separator line (|---|---|)
            if all(re.match(r"^[-:]+$", c) for c in cells if c):
                # Skip separator line, it's just formatting
                continue

            if in_quote:
                result.append("</blockquote>")
                in_quote = False
            if in_list:
                result.append("</ul>")
                in_list = False

            if not in_table:
                result.append("<table>")
                in_table = True
                # First row is header
                result.append(
                    "<tr>" + "".join(f"<th>{c}</th>" for c in cells) + "</tr>"
                )
            else:
                result.append(
                    "<tr>" + "".join(f"<td>{c}</td>" for c in cells) + "</tr>"
                )
            continue

        # Close table if we're leaving table context
        if in_table and not (stripped.startswith("|") and stripped.endswith("|")):
            result.append("</table>")
            in_table = False

        # Blockquotes: > text
        if stripped.startswith("> "):
            if not in_quote:
                if in_list:
                    result.append("</ul>")
                    in_list = False
                result.append("<blockquote>")
                in_quote = True
            result.append(stripped[2:])
        # Lists: `- item`, `* item`, `+ item` (all valid CommonMark bullets)
        elif stripped[:2] in ("- ", "* ", "+ "):
            if in_quote:
                result.append("</blockquote>")
                in_quote = False
            if not in_list:
                result.append("<ul>")
                in_list = True
            result.append(f"<li>{stripped[2:]}</li>")
        elif stripped == "":
            # End blockquote on empty line
            if in_quote:
                result.append("</blockquote>")
                in_quote = False
            continue
        else:
            if in_quote:
                result.append("</blockquote>")
                in_quote = False
            if in_list:
                result.append("</ul>")
                in_list = False
            result.append(line)

    # Close any open tags
    if in_quote:
        result.append("</blockquote>")
    if in_list:
        result.append("</ul>")
    if in_table:
        result.append("</table>")

    # Join with special marker, then convert to <br> only outside block elements
    html = "{{BR}}".join(result)
    # Don't add <br> around block elements
    html = re.sub(
        r"\{\{BR\}\}(?=<ul>|<li>|</ul>|</li>|<blockquote>|</blockquote>|<pre>|<table>|<tr>|</table>|<h[1-6]>)",
        "",
        html,
    )
    html = re.sub(
        r"(</ul>|</li>|</blockquote>|</pre>|</table>|</tr>|</h[1-6]>)\{\{BR\}\}",
        r"\1",
        html,
    )
    html = re.sub(r"(<blockquote>|<table>)\{\{BR\}\}", r"\1", html)
    html = html.replace("{{BR}}", "<br>")

    # Restore code blocks
    for idx, block in enumerate(code_blocks):
        html = html.replace(f"{{{{CODEBLOCK_{idx}}}}}", block)

    return html


def add_bot_prefix(message: str, prefix: str) -> str:
    """Add bot prefix intelligently.

    If message starts with a heading, insert prefix after the heading.
    Otherwise, prepend prefix to the message.
    """
    lines = message.split("\n")
    if not lines:
        return f"{prefix} {message}"

    first_line = lines[0].strip()

    # Check if first line is a heading
    if re.match(r"^#{1,6}\s+", first_line):
        # Insert prefix after heading on same line or next line
        lines[0] = first_line
        if len(lines) > 1:
            # Insert prefix at start of content after heading
            lines.insert(1, f"\n{prefix}")
        else:
            # Add prefix after heading
            lines.append(f"\n{prefix}")
        return "\n".join(lines)
    else:
        # Prepend prefix to message
        return f"{prefix} {message}"

"""HTTP-based Matrix API requests.

All functions use ONLY stdlib.
"""

import contextlib
import json
import socket
import urllib.parse
import urllib.request
import urllib.error


_ALLOWED_SCHEMES = frozenset({"http", "https"})


def _require_http_scheme(url: str) -> None:
    scheme = urllib.parse.urlparse(url).scheme
    if scheme not in _ALLOWED_SCHEMES:
        raise ValueError(
            f"Refusing to fetch URL with scheme {scheme!r}; only http/https allowed"
        )


@contextlib.contextmanager
def _prefer_ipv4():
    """Temporarily prefer IPv4 addresses in DNS resolution.

    Workaround for WSL2 environments where IPv6 routes are often
    unreachable while IPv4 works fine.
    """
    original = socket.getaddrinfo

    def patched(*args, **kwargs):
        results = original(*args, **kwargs)
        return sorted(results, key=lambda r: r[0] != socket.AF_INET)

    socket.getaddrinfo = patched
    try:
        yield
    finally:
        socket.getaddrinfo = original


def _do_request(req: urllib.request.Request) -> dict:
    """Execute a request and return parsed JSON response.

    The caller must have validated the URL scheme via
    ``_require_http_scheme`` before constructing ``req``.
    """
    _require_http_scheme(req.full_url)
    # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
    with urllib.request.urlopen(req) as response:  # noqa: S310 — scheme validated above
        return json.loads(response.read().decode())


def _parse_http_error(e: urllib.error.HTTPError) -> dict:
    """Parse an HTTPError into a result dict."""
    error_body = e.read().decode()
    try:
        error_json = json.loads(error_body)
        return {
            "error": error_json.get("error", error_body),
            "errcode": error_json.get("errcode"),
        }
    except json.JSONDecodeError:
        return {"error": error_body, "errcode": str(e.code)}


def matrix_request(config: dict, method: str, endpoint: str, data: dict = None) -> dict:
    """Make a Matrix API request.

    Args:
        config: Dict with homeserver and access_token
        method: HTTP method (GET, POST, PUT, DELETE)
        endpoint: API endpoint (e.g., /joined_rooms)
        data: Optional dict to send as JSON body

    Returns:
        Response dict, or dict with 'error' key on failure
    """
    url = f"{config['homeserver']}/_matrix/client/v3{endpoint}"
    _require_http_scheme(url)
    headers = {
        "Authorization": f"Bearer {config['access_token']}",
        "Content-Type": "application/json",
    }

    body = json.dumps(data).encode() if data is not None else None
    req = urllib.request.Request(url, data=body, headers=headers, method=method)

    try:
        return _do_request(req)
    except urllib.error.HTTPError as e:
        return _parse_http_error(e)
    except OSError as e:
        if e.errno not in (101, 113):  # ENETUNREACH, EHOSTUNREACH
            return {"error": str(e)}
        # IPv6 likely unreachable — retry with IPv4 preference
        try:
            with _prefer_ipv4():
                req2 = urllib.request.Request(
                    url, data=body, headers=headers, method=method
                )
                return _do_request(req2)
        except urllib.error.HTTPError as e2:
            return _parse_http_error(e2)
        except OSError as e2:
            return {"error": str(e2)}

"""Room operations for Matrix scripts.

All functions use ONLY stdlib.
"""

import urllib.parse
from _lib.http import matrix_request


def resolve_room_alias(config: dict, alias: str) -> str:
    """Resolve a room alias to room ID.

    Args:
        config: Matrix config with homeserver and access_token
        alias: Room alias (e.g., #room:server)

    Returns:
        Room ID (e.g., !abc123:server)

    Raises:
        ValueError if alias cannot be resolved
    """
    encoded_alias = urllib.parse.quote(alias, safe="")
    result = matrix_request(config, "GET", f"/directory/room/{encoded_alias}")
    if "room_id" in result:
        return result["room_id"]
    raise ValueError(
        f"Could not resolve room alias: {result.get('error', 'Unknown error')}"
    )


def get_room_info(config: dict, room_id: str) -> dict:
    """Get the display name and canonical alias of a room.

    Args:
        config: Matrix config with homeserver and access_token
        room_id: Room ID to query

    Returns:
        Dict with 'name' and 'alias' keys (values may be None)
    """
    info = {"name": None, "alias": None}

    result = matrix_request(config, "GET", f"/rooms/{room_id}/state/m.room.name")
    if "name" in result:
        info["name"] = result["name"]

    result = matrix_request(
        config, "GET", f"/rooms/{room_id}/state/m.room.canonical_alias"
    )
    if "alias" in result:
        info["alias"] = result["alias"]

    return info


def list_joined_rooms(config: dict) -> list:
    """List all joined rooms with names and aliases.

    Args:
        config: Matrix config with homeserver and access_token

    Returns:
        List of dicts with room_id, name, and alias keys
    """
    result = matrix_request(config, "GET", "/joined_rooms")
    if "error" in result:
        return []

    rooms = []
    for room_id in result.get("joined_rooms", []):
        info = get_room_info(config, room_id)
        display_name = info["name"] or info["alias"] or room_id
        rooms.append({"room_id": room_id, "name": display_name, "alias": info["alias"]})

    return rooms


def find_room_by_name(config: dict, search_term: str) -> tuple[str | None, list]:
    """Find a room by name or alias (case-insensitive).

    Match priority:
    1. Exact alias match (#room:server)
    2. Exact alias name match (without server part, e.g. "agent-work")
    3. Exact room name match (rooms with aliases preferred)
    4. Single partial match on name or alias

    When an exact name match has no alias but other rooms also have names
    containing the search term, all candidates are returned for disambiguation.

    Args:
        config: Matrix config with homeserver and access_token
        search_term: Search term to match against room names/aliases

    Returns:
        (room_id, matches) where:
        - room_id is the matched room ID (or None if no/ambiguous match)
        - matches is list of matching rooms (for error reporting)
    """
    rooms = list_joined_rooms(config)
    search_lower = search_term.lower()

    # Try exact alias match (most specific)
    for room in rooms:
        if room.get("alias") and room["alias"].lower() == search_lower:
            return room["room_id"], [room]

    # Try exact alias name match (without server part)
    for room in rooms:
        if room.get("alias"):
            alias_name = room["alias"].split(":")[0].lstrip("#")
            if alias_name.lower() == search_lower:
                return room["room_id"], [room]

    # Try exact name match
    name_matches = [r for r in rooms if r["name"].lower() == search_lower]
    if len(name_matches) == 1:
        room = name_matches[0]
        if room.get("alias"):
            # Room has an alias — well-identified, return directly
            return room["room_id"], name_matches
        # Room has no alias — check if other rooms have names containing
        # the search term, which suggests the user may want a different room
        alternatives = [
            r
            for r in rooms
            if r not in name_matches and search_lower in r["name"].lower()
        ]
        if alternatives:
            return None, name_matches + alternatives
        return room["room_id"], name_matches
    if len(name_matches) > 1:
        return None, name_matches

    # Try partial match
    matches = []
    for room in rooms:
        if search_lower in room["name"].lower():
            matches.append(room)
        elif room.get("alias") and search_lower in room["alias"].lower():
            if room not in matches:
                matches.append(room)

    if len(matches) == 1:
        return matches[0]["room_id"], matches

    return None, matches


def find_room_in_nio_client(client_rooms: dict, search_term: str) -> str | None:
    """Find a room by name in a matrix-nio client.rooms dict (post-sync).

    This avoids the N+1 HTTP calls of find_room_by_name() by using
    room data already loaded by client.sync().

    Args:
        client_rooms: dict from AsyncClient.rooms (room_id -> MatrixRoom)
        search_term: Room name, alias, or ID to match

    Returns:
        room_id if found, None otherwise
    """
    search_lower = search_term.lower()

    # Exact alias match
    for room_id, room in client_rooms.items():
        if room.canonical_alias and room.canonical_alias.lower() == search_lower:
            return room_id

    # Alias name match (without server part)
    for room_id, room in client_rooms.items():
        if room.canonical_alias:
            alias_name = room.canonical_alias.split(":")[0].lstrip("#")
            if alias_name.lower() == search_lower:
                return room_id

    # Exact display name match
    exact_matches = [
        room_id
        for room_id, room in client_rooms.items()
        if room.display_name and room.display_name.lower() == search_lower
    ]
    if len(exact_matches) == 1:
        return exact_matches[0]
    if len(exact_matches) > 1:
        return None  # Ambiguous exact match

    # Partial name match
    partial_matches = [
        room_id
        for room_id, room in client_rooms.items()
        if search_lower in (room.display_name or "").lower()
        or search_lower in (room.canonical_alias or "").lower()
    ]
    if len(partial_matches) == 1:
        return partial_matches[0]

    return None

"""Tests for `_lib.formatting` markdown→HTML conversion.

The skill directory contains a hyphen (`matrix-communication`) so it is
not importable as a Python package; run the file directly or use unittest
discovery:

    python3 skills/matrix-communication/scripts/_lib/test_formatting.py
    python3 -m unittest discover \\
        -s skills/matrix-communication/scripts/_lib -p 'test_formatting.py'

Stdlib only.
"""

import os
import sys
import unittest

sys.path.insert(0, os.path.dirname(__file__))

from formatting import markdown_to_html, shorten_service_urls  # noqa: E402


class ShortenServiceUrlsTests(unittest.TestCase):
    def test_jira_bare_url_is_shortened(self):
        out = shorten_service_urls("see https://jira.example.com/browse/PROJ-42")
        self.assertIn("[PROJ-42](https://jira.example.com/browse/PROJ-42)", out)

    def test_gitlab_bare_url_is_shortened(self):
        out = shorten_service_urls(
            "see https://gitlab.example.com/grp/proj/-/merge_requests/7"
        )
        self.assertIn(
            "[grp/proj#7](https://gitlab.example.com/grp/proj/-/merge_requests/7)",
            out,
        )

    def test_existing_gitlab_markdown_link_is_preserved(self):
        """Regression: pre-wrapped markdown links must NOT be re-wrapped.

        Before the fix this produced `[mytext]([grp/proj#7](url))` which
        confused the link parser and leaked literal `)` into the rendered
        message.
        """
        src = "[mytext](https://gitlab.example.com/grp/proj/-/merge_requests/7) — note"
        out = shorten_service_urls(src)
        self.assertEqual(out, src)

    def test_existing_jira_markdown_link_is_preserved(self):
        src = "[NRS-1](https://jira.example.com/browse/NRS-1) — note"
        out = shorten_service_urls(src)
        self.assertEqual(out, src)

    def test_existing_github_markdown_link_is_preserved(self):
        src = "[my pr](https://github.com/owner/repo/pull/9) — note"
        out = shorten_service_urls(src)
        self.assertEqual(out, src)

    def test_mixed_protected_and_bare_urls(self):
        src = (
            "see [my mr](https://gitlab.example.com/grp/proj/-/merge_requests/7) "
            "and https://gitlab.example.com/grp/proj/-/issues/12"
        )
        out = shorten_service_urls(src)
        # First link untouched
        self.assertIn(
            "[my mr](https://gitlab.example.com/grp/proj/-/merge_requests/7)", out
        )
        # Second link auto-shortened
        self.assertIn(
            "[grp/proj#12](https://gitlab.example.com/grp/proj/-/issues/12)", out
        )

    def test_forged_placeholder_does_not_raise(self):
        """A user-supplied `\\x00MDLINK<n>\\x00` sequence must not crash the
        restore step with IndexError. Unknown placeholders are left as-is."""
        # No real markdown link in the input, but a forged placeholder is.
        src = "literal placeholder \x00MDLINK99\x00 should pass through"
        out = shorten_service_urls(src)
        self.assertIn("\x00MDLINK99\x00", out)


class MarkdownToHtmlListTests(unittest.TestCase):
    def test_dash_bullets_render_as_ul(self):
        html = markdown_to_html("- one\n- two")
        self.assertIn("<ul>", html)
        self.assertIn("<li>one</li>", html)
        self.assertIn("<li>two</li>", html)

    def test_asterisk_bullets_render_as_ul(self):
        """Regression: `* ` bullets must render as a list, not as italic."""
        html = markdown_to_html("* one\n* two")
        self.assertIn("<ul>", html)
        self.assertIn("<li>one</li>", html)
        self.assertIn("<li>two</li>", html)
        self.assertNotIn("<em>", html)

    def test_plus_bullets_render_as_ul(self):
        html = markdown_to_html("+ one\n+ two")
        self.assertIn("<ul>", html)
        self.assertIn("<li>one</li>", html)
        self.assertIn("<li>two</li>", html)

    def test_asterisk_bullet_with_inline_italic(self):
        """A bullet line may still contain an italic span elsewhere."""
        html = markdown_to_html("* item *emph* tail")
        self.assertIn("<li>item <em>emph</em> tail</li>", html)

    def test_asterisk_inside_word_still_italicises(self):
        """Mid-line `*pairs*` still emit <em>."""
        html = markdown_to_html("hello *world*")
        self.assertIn("<em>world</em>", html)


class EmphasisFlankingTests(unittest.TestCase):
    """CommonMark left/right-flanking rules for `*…*`, `**…**`, `~~…~~`.

    Opening delimiter must NOT be followed by whitespace; closing delimiter
    must NOT be preceded by whitespace. Without these guards the italic
    regex used to swallow bullet `*` markers and produce broken output.
    """

    def test_italic_open_followed_by_space_does_not_match(self):
        # `*` immediately followed by space is not a valid italic opener.
        html = markdown_to_html("* one and *two*")
        self.assertNotIn("<em> one and </em>", html)
        self.assertIn("<em>two</em>", html)

    def test_italic_close_preceded_by_space_does_not_match(self):
        # `*` immediately preceded by space is not a valid italic closer.
        html = markdown_to_html("a *foo *bar")
        self.assertNotIn("<em>foo </em>", html)

    def test_bold_open_followed_by_space_does_not_match(self):
        html = markdown_to_html("** not bold ** but **bold**")
        self.assertNotIn("<strong> not bold </strong>", html)
        self.assertIn("<strong>bold</strong>", html)

    def test_strikethrough_open_followed_by_space_does_not_match(self):
        html = markdown_to_html("~~ not strike ~~ but ~~strike~~")
        self.assertNotIn("<del> not strike </del>", html)
        self.assertIn("<del>strike</del>", html)

    def test_italic_single_char(self):
        # `*x*` is a valid italic: opener followed by non-ws, closer
        # preceded by non-ws.
        html = markdown_to_html("*x*")
        self.assertIn("<em>x</em>", html)


class MarkdownToHtmlLinkTests(unittest.TestCase):
    def test_pre_wrapped_gitlab_link_renders_clean_anchor(self):
        """Regression: no stray `)` after the anchor."""
        src = "[my mr](https://gitlab.example.com/grp/proj/-/merge_requests/7)"
        html = markdown_to_html(src)
        self.assertIn(
            '<a href="https://gitlab.example.com/grp/proj/-/merge_requests/7">'
            "my mr</a>",
            html,
        )
        # Critical: no leaked closing paren after the anchor
        self.assertNotIn("</a>)", html)

    def test_pre_wrapped_jira_link_renders_clean_anchor(self):
        src = "[NRS-1](https://jira.example.com/browse/NRS-1)"
        html = markdown_to_html(src)
        self.assertIn('<a href="https://jira.example.com/browse/NRS-1">NRS-1</a>', html)
        self.assertNotIn("</a>)", html)

    def test_bullet_list_with_pre_wrapped_links(self):
        """End-to-end regression: a bullet list of pre-wrapped GitLab + Jira
        links must produce a clean `<ul><li><a>…</a></li>…</ul>` tree, with
        the pre-wrapped link text preserved (not replaced by the auto-shortener).
        """
        src = (
            "- [proj#7](https://gitlab.example.com/grp/proj/-/merge_requests/7) "
            "— note ([NRS-1](https://jira.example.com/browse/NRS-1))\n"
            "- [proj#8](https://gitlab.example.com/grp/proj/-/merge_requests/8)"
        )
        html = markdown_to_html(src)
        self.assertIn("<ul>", html)
        # Caller-provided link text preserved (would have been clobbered to
        # `grp/proj#7` by the broken auto-shortener)
        self.assertIn(
            '<a href="https://gitlab.example.com/grp/proj/-/merge_requests/7">'
            "proj#7</a>",
            html,
        )
        self.assertIn('<a href="https://jira.example.com/browse/NRS-1">NRS-1</a>', html)
        # No nested `[…](…)` artefacts (the symptom of broken re-wrapping)
        self.assertNotIn("[", html)
        self.assertNotIn("](", html)
        # Sanity: no leftover bullet asterisks (would mean inline ate the bullet)
        self.assertNotIn("<li>*", html)


if __name__ == "__main__":
    unittest.main()

"""Utility functions for Matrix scripts.

All functions use ONLY stdlib.
"""

import logging
import socket
from datetime import datetime


def clean_message(message: str) -> str:
    """Clean message from bash escaping artifacts.

    Bash history expansion in interactive shells can escape ! to \\!
    when using double quotes. This removes those artifacts.
    """
    # Remove backslash before ! (bash history expansion artifact)
    return message.replace("\\!", "!")


def prefer_ipv4():
    """Monkey-patch socket.getaddrinfo to prefer IPv4 results.

    Workaround for WSL2 environments where IPv6 routes are often
    unreachable while IPv4 works fine.  Call once at script startup.
    """
    _orig = socket.getaddrinfo
    socket.getaddrinfo = lambda *a, **kw: sorted(
        _orig(*a, **kw), key=lambda r: r[0] != socket.AF_INET
    )


def suppress_nio_logging():
    """Suppress noisy matrix-nio crypto/sync warnings.

    Sets nio and peewee loggers to ERROR level to hide megolm session
    warnings that clutter output during normal operation.
    """
    for name in ("nio", "nio.crypto", "nio.responses", "peewee"):
        logging.getLogger(name).setLevel(logging.ERROR)


def format_timestamp(ts: int) -> str:
    """Format Matrix timestamp to readable string.

    Args:
        ts: Unix timestamp in milliseconds

    Returns:
        Formatted string like "2024-01-15 14:30"
    """
    if ts == 0:
        return "unknown"
    dt = datetime.fromtimestamp(ts / 1000)
    return dt.strftime("%Y-%m-%d %H:%M")

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///
"""Matrix Skill health check and dependency installer.

Checks all dependencies and configuration, installs missing packages,
and reports on E2EE setup status.

Usage:
    matrix-doctor.py [--install] [--json] [--quiet]
    matrix-doctor.py --help

Options:
    --install   Automatically install missing dependencies
    --json      Output as JSON
    --quiet     Only show errors
    --help      Show this help
"""

import json
import os
import shutil
import subprocess
import sys

# Add script directory to path for _lib imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib.config import get_config_path
from _lib.e2ee import get_store_path

sys.stdout.reconfigure(line_buffering=True)
sys.stderr.reconfigure(line_buffering=True)


def check_command_exists(cmd: str) -> bool:
    """Check if a command exists in PATH."""
    return shutil.which(cmd) is not None


def get_pip_command() -> str | None:
    """Get the best available pip command. Priority: uvx, pip, pip3."""
    # Check uvx first (preferred)
    if check_command_exists("uvx"):
        return "uvx pip"
    if check_command_exists("uv"):
        return "uv pip"
    if check_command_exists("pip"):
        return "pip"
    if check_command_exists("pip3"):
        return "pip3"
    return None


def run_pip_command(pip_cmd: str, args: list[str]) -> tuple[bool, str]:
    """Run a pip command and return success status and output."""
    if pip_cmd.startswith("uvx"):
        full_cmd = ["uvx", "pip"] + args
    elif pip_cmd.startswith("uv"):
        full_cmd = ["uv", "pip"] + args
    else:
        full_cmd = [pip_cmd] + args

    try:
        result = subprocess.run(full_cmd, capture_output=True, text=True, timeout=120)
        return result.returncode == 0, result.stdout + result.stderr
    except subprocess.TimeoutExpired:
        return False, "Command timed out"
    except Exception as e:
        return False, str(e)


def check_matrix_nio_e2ee() -> tuple[bool, str]:
    """Check if matrix-nio with E2EE support is installed."""
    try:
        import nio  # noqa: F401

        # Try to get version
        try:
            from importlib.metadata import version

            nio_version = version("matrix-nio")
        except Exception:
            nio_version = "unknown"

        # Check for E2EE support by trying to import olm
        try:
            from nio.crypto import Olm  # noqa: F401

            return True, f"matrix-nio {nio_version} with E2EE support"
        except ImportError:
            return False, f"matrix-nio {nio_version} installed but E2EE deps missing"
    except ImportError:
        return False, "matrix-nio not installed"


def check_libolm() -> tuple[bool, str]:
    """Check if libolm system library is installed."""
    try:
        import _libolm  # noqa: F401

        return True, "libolm available"
    except ImportError:
        pass

    # Try loading the shared library
    import ctypes.util

    lib = ctypes.util.find_library("olm")
    if lib:
        return True, f"libolm found: {lib}"
    return False, "libolm not found (required for E2EE)"


def check_config() -> tuple[bool, str, dict]:
    """Check Matrix configuration file."""
    config_path = get_config_path()
    if not config_path.exists():
        return False, f"Config not found at {config_path}", {}

    try:
        with open(config_path) as f:
            config = json.load(f)

        required = ["homeserver", "user_id"]
        missing = [k for k in required if k not in config]
        if missing:
            return (
                False,
                f"Config missing required fields: {', '.join(missing)}",
                config,
            )

        return True, f"Config OK: {config.get('user_id')}", config
    except json.JSONDecodeError as e:
        return False, f"Invalid JSON in config: {e}", {}
    except Exception as e:
        return False, f"Error reading config: {e}", {}


def check_e2ee_setup() -> tuple[bool, str]:
    """Check E2EE device setup status."""
    store_dir = get_store_path()
    creds_file = store_dir / "credentials.json"

    if not store_dir.exists():
        return False, "E2EE not set up (no store directory)"

    if not creds_file.exists():
        return False, "E2EE not set up (no credentials)"

    try:
        with open(creds_file) as f:
            creds = json.load(f)
        device_id = creds.get("device_id", "unknown")
        return True, f"E2EE device configured: {device_id}"
    except Exception as e:
        return False, f"Error reading E2EE credentials: {e}"


def install_dependencies(pip_cmd: str, quiet: bool = False) -> tuple[bool, list[str]]:
    """Install missing dependencies."""
    messages = []

    # Install matrix-nio with E2EE support
    if not quiet:
        messages.append("Installing matrix-nio[e2e]...")

    success, output = run_pip_command(pip_cmd, ["install", "matrix-nio[e2e]"])
    if success:
        messages.append("matrix-nio[e2e] installed successfully")
    else:
        messages.append(f"Failed to install matrix-nio[e2e]: {output}")
        return False, messages

    return True, messages


def main():
    import argparse

    parser = argparse.ArgumentParser(description="Matrix Skill health check and setup")
    parser.add_argument(
        "--install",
        action="store_true",
        help="Automatically install missing dependencies",
    )
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--quiet", "-q", action="store_true", help="Only show errors")

    args = parser.parse_args()

    checks = {
        "pip_available": {"ok": False, "message": "", "critical": True},
        "matrix_nio": {"ok": False, "message": "", "critical": True},
        "libolm": {"ok": False, "message": "", "critical": False},
        "config": {"ok": False, "message": "", "critical": True},
        "e2ee_setup": {"ok": False, "message": "", "critical": False},
    }

    # Check pip availability
    pip_cmd = get_pip_command()
    if pip_cmd:
        checks["pip_available"]["ok"] = True
        checks["pip_available"]["message"] = f"Using: {pip_cmd}"
    else:
        checks["pip_available"]["message"] = (
            "No pip command found (tried: uvx, uv pip, pip, pip3)"
        )

    # Check matrix-nio
    nio_ok, nio_msg = check_matrix_nio_e2ee()
    checks["matrix_nio"]["ok"] = nio_ok
    checks["matrix_nio"]["message"] = nio_msg

    # Check libolm
    olm_ok, olm_msg = check_libolm()
    checks["libolm"]["ok"] = olm_ok
    checks["libolm"]["message"] = olm_msg

    # Check config
    config_ok, config_msg, config_data = check_config()
    checks["config"]["ok"] = config_ok
    checks["config"]["message"] = config_msg

    # Check E2EE setup
    e2ee_ok, e2ee_msg = check_e2ee_setup()
    checks["e2ee_setup"]["ok"] = e2ee_ok
    checks["e2ee_setup"]["message"] = e2ee_msg

    # Auto-install if requested
    if args.install and pip_cmd:
        if not checks["matrix_nio"]["ok"]:
            success, messages = install_dependencies(pip_cmd, args.quiet)
            if success:
                # Re-check after install
                nio_ok, nio_msg = check_matrix_nio_e2ee()
                checks["matrix_nio"]["ok"] = nio_ok
                checks["matrix_nio"]["message"] = nio_msg
                checks["install_messages"] = messages

    # Output
    if args.json:
        print(json.dumps(checks, indent=2))
        sys.exit(0 if all(c["ok"] for c in checks.values() if c.get("critical")) else 1)

    # Pretty output
    all_ok = True
    critical_ok = True

    if not args.quiet:
        print("=" * 60)
        print("Matrix Skill Health Check")
        print("=" * 60)
        print()

    for name, check in checks.items():
        if name == "install_messages":
            continue

        icon = "OK" if check["ok"] else "FAIL"
        critical = " (required)" if check.get("critical") else ""

        if not check["ok"]:
            all_ok = False
            if check.get("critical"):
                critical_ok = False

        if not args.quiet or not check["ok"]:
            print(f"[{icon}] {name}{critical}")
            print(f"     {check['message']}")
            print()

    # Summary
    if not args.quiet:
        print("=" * 60)

    if all_ok:
        print("All checks passed! Matrix Skill is ready to use.")
    elif critical_ok:
        print("Core functionality OK. Some optional features may be limited.")
    else:
        print("Some required checks failed. See above for details.")
        print()
        print("Quick fix:")
        if not checks["pip_available"]["ok"]:
            print("  - Install uv: pip install uv")
        if not checks["matrix_nio"]["ok"]:
            print("  - Run: matrix-doctor.py --install")
        if not checks["config"]["ok"]:
            print("  - Set up Matrix: see SKILL.md Setup Guide")
        if not checks["e2ee_setup"]["ok"] and checks["config"]["ok"]:
            print("  - Run: matrix-e2ee-setup.py")

    sys.exit(0 if critical_ok else 1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = ["matrix-nio[e2e]"]
# ///
"""Download media from a Matrix room message.

Usage:
    matrix-download-e2ee.py ROOM EVENT_ID [--output DIR] [--filename NAME]
    matrix-download-e2ee.py --help

Arguments:
    ROOM        Room alias (#room:server), room ID (!id:server), or room name
    EVENT_ID    Event ID of the media message ($xxx:server)

Options:
    --output DIR     Output directory [default: .]
    --filename NAME  Override filename (default: from message body)
    --debug          Show debug information
    --help           Show this help
"""

import argparse
import asyncio
import sys
import os
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from _lib import (
    check_e2ee_dependencies,
    load_config,
    get_store_path,
    load_credentials,
    find_room_in_nio_client,
    prefer_ipv4,
    suppress_nio_logging,
)

check_e2ee_dependencies()

from nio import (
    AsyncClient,
    AsyncClientConfig,
    RoomResolveAliasResponse,
    RoomGetEventError,
    DownloadError,
    MemoryDownloadResponse,
)


async def download_media(
    config: dict,
    room: str,
    event_id: str,
    output_dir: str = ".",
    filename: str | None = None,
    debug: bool = False,
) -> str:
    """Download media from a Matrix message event."""
    store_path = get_store_path()
    stored_creds = load_credentials()

    if stored_creds and stored_creds.get("user_id") == config["user_id"]:
        device_id = stored_creds["device_id"]
        access_token = stored_creds["access_token"]
    elif "access_token" in config:
        access_token = config["access_token"]
        from nio import WhoamiResponse

        temp_client = AsyncClient(config["homeserver"], config["user_id"])
        temp_client.access_token = access_token
        whoami = await temp_client.whoami()
        await temp_client.close()
        if isinstance(whoami, WhoamiResponse):
            device_id = whoami.device_id
        else:
            raise RuntimeError(f"Failed to get device info: {whoami}")
    else:
        raise RuntimeError("No credentials found. Run matrix-e2ee-setup.py first.")

    client_config = AsyncClientConfig(store_sync_tokens=True, encryption_enabled=True)
    client = AsyncClient(
        homeserver=config["homeserver"],
        user=config["user_id"],
        device_id=device_id,
        store_path=str(store_path),
        config=client_config,
    )

    try:
        client.restore_login(
            user_id=config["user_id"], device_id=device_id, access_token=access_token
        )

        if client.store:
            client.load_store()

        if client.should_upload_keys:
            await client.keys_upload()

        # Sync to get room keys
        if debug:
            print("Syncing...", file=sys.stderr)
        await client.sync(timeout=0, full_state=True)

        # Resolve room
        room_id = room
        if room.startswith("#"):
            response = await client.room_resolve_alias(room)
            if isinstance(response, RoomResolveAliasResponse):
                room_id = response.room_id
            else:
                raise RuntimeError(f"Could not resolve room alias: {response}")
        elif not room.startswith("!"):
            found = find_room_in_nio_client(client.rooms, room)
            if found:
                room_id = found
            else:
                raise RuntimeError(
                    f"Could not find room '{room}'. Use 'matrix-rooms.py' to list rooms."
                )

        # Fetch the event
        if debug:
            print(f"Fetching event {event_id}...", file=sys.stderr)

        resp = await client.room_get_event(room_id, event_id)
        if isinstance(resp, RoomGetEventError):
            raise RuntimeError(f"Failed to get event: {resp.message}")

        event = resp.event
        source = event.source if hasattr(event, "source") else {}
        content = source.get("content", {})
        msgtype = content.get("msgtype", "")

        if msgtype not in ("m.image", "m.file", "m.video", "m.audio"):
            raise RuntimeError(f"Event is not a media message (msgtype: {msgtype})")

        # Get mxc URL
        if "file" in content:
            mxc_url = content["file"]["url"]
        elif "url" in content:
            mxc_url = content["url"]
        else:
            raise RuntimeError("No media URL found in event")

        if debug:
            print(f"Downloading {mxc_url}...", file=sys.stderr)

        # Determine filename — sanitize to prevent path traversal
        if not filename:
            raw_name = content.get("body", "media_download")
            filename = Path(raw_name).name  # Strip directory components
        else:
            filename = Path(filename).name

        # Download media into memory (don't pass filename to avoid unnecessary disk write)
        resp = await client.download(mxc=mxc_url)

        if isinstance(resp, DownloadError):
            raise RuntimeError(f"Download failed: {resp.message}")

        # Get raw bytes
        if isinstance(resp, MemoryDownloadResponse):
            data = resp.body
        else:
            data = Path(resp.filename).read_bytes()

        # Decrypt E2EE media if encrypted
        if "file" in content:
            from nio.crypto import decrypt_attachment

            file_info = content["file"]
            data = decrypt_attachment(
                ciphertext=data,
                key=file_info["key"]["k"],
                hash=file_info["hashes"]["sha256"],
                iv=file_info["iv"],
            )

        # Save to file
        out_path = Path(output_dir) / filename
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_bytes(data)

        if debug:
            print(
                f"Saved to {out_path} ({out_path.stat().st_size:,} bytes)",
                file=sys.stderr,
            )

        return str(out_path)

    finally:
        await client.close()


def main():
    parser = argparse.ArgumentParser(description="Download media from Matrix room")
    parser.add_argument("room", help="Room alias, ID, or name")
    parser.add_argument("event_id", help="Event ID of the media message")
    parser.add_argument("--output", default=".", help="Output directory [default: .]")
    parser.add_argument("--filename", help="Override filename")
    parser.add_argument("--debug", action="store_true", help="Debug output")
    args = parser.parse_args()

    suppress_nio_logging()
    prefer_ipv4()

    config = load_config()

    try:
        path = asyncio.run(
            download_media(
                config,
                args.room,
                args.event_id,
                output_dir=args.output,
                filename=args.filename,
                debug=args.debug,
            )
        )
        print(path)
    except Exception as e:
        if args.debug:
            raise
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()