diff --git a/.gitignore b/.gitignore index cad0729..e97851c 100644 --- a/.gitignore +++ b/.gitignore @@ -93,6 +93,10 @@ data_engine.db-journal .artemis_ipc_port .ls_address .jetski_ls_address +.jetski_env +.artemis_env +live_screenshot*.jpg +live_screenshot*.png # Environment & Secrets .env diff --git a/artemis/interfaces/cli/commands/mcp.py b/artemis/interfaces/cli/commands/mcp.py index 03ae95a..d66c144 100644 --- a/artemis/interfaces/cli/commands/mcp.py +++ b/artemis/interfaces/cli/commands/mcp.py @@ -66,6 +66,7 @@ def _get_config_snippet(client: str, python_exe: str, project_root: str) -> dict "env": { "PYTHONUNBUFFERED": "1", "PYTHONPATH": project_root, + "ARTEMIS_DESKTOP_NOTIFY": "true", }, } config_with_cwd = {**config_body, "cwd": project_root} diff --git a/mcp_server/README.md b/mcp_server/README.md index ae1309e..77a5959 100644 --- a/mcp_server/README.md +++ b/mcp_server/README.md @@ -14,10 +14,11 @@ mcp_server/ │ └── task_runner.py # Subprocess runner for Artemis Agent ├── notifiers/ # Multi-environment notification adapters │ ├── base.py # BaseNotifier abstract class -│ ├── agentapi.py # Jetski / Antigravity AgentAPI notifier +│ ├── agentapi.py # Jetski / Antigravity AgentAPI notifier (with session sorting & caching) +│ ├── desktop.py # Native OS toast notifier (macOS/Linux/Windows, enabled by default) +│ ├── script.py # Universal Script / Command Hook notifier (ARTEMIS_NOTIFY_CMD) │ ├── webhook.py # OpenClaw / CI / Discord / Slack webhook notifier -│ ├── desktop.py # Native OS toast notifier -│ ├── file.py # File audit logger +│ ├── file.py # File audit logger (notifications.jsonl) │ └── composite.py # Multi-channel notification dispatcher ├── tools/ # Modular standard MCP tools │ ├── task_runner.py # mobile_run_task @@ -30,6 +31,18 @@ mcp_server/ └── trace_store.py # Traces directory and status manager ``` +## 🔔 Multi-Platform Notification & Wakeup Support + +When you start an asynchronous mobile test via `mobile_run_task`, Artemis MCP runs the job as a detached background process so the AI agent is not blocked. Upon completion or failure, Artemis dispatches alerts across **all available notification channels** so that both human developers and AI agents are notified immediately across different platforms: + +| Platform / Environment | Primary Notification Mechanism | How It Works & Configuration | +| :--- | :--- | :--- | +| **Antigravity / Jetski** | **Reactive Wakeup (`AgentApiNotifier`)** | Automatically detects and caches the active IDE session credentials (`ANTIGRAVITY_LS_ADDRESS`, `ANTIGRAVITY_CSRF_TOKEN`), sorting candidate processes by creation time to avoid stale PIDs, and calls `agentapi send-message` to wake up the agent directly in the conversation. | +| **Cursor / Windsurf / VS Code / Cline / Roo Code** | **Native Desktop Toast (`DesktopNotifier`) + File Audit (`FileNotifier`)** | Enabled **by default** across macOS (`osascript`), Linux (`notify-send`), and Windows (`powershell`). Pops up a system notification banner informing the developer whether the task completed or failed. The AI agent inspects `/status.json` or uses `mobile_manage_task(action="status")`. Can be silenced via `ARTEMIS_DESKTOP_NOTIFY=false`. | +| **Claude Code / Desktop** | **Native Desktop Toast + File Audit** | Alerts the user via system desktop notification banner when background execution concludes, while maintaining a complete JSONL audit log in `notifications.jsonl`. | +| **OpenClaw / Slack / Discord / CI/CD** | **HTTP Webhook (`WebhookNotifier`)** | Sends structured JSON POST payloads to custom endpoints configured via `OPENCLAW_WEBHOOK_URL`, `MCP_NOTIFICATION_WEBHOOK`, or `ARTEMIS_WEBHOOK_URL`. | +| **Universal Custom IDE / CLI Hooks** | **Custom Script Hook (`ScriptNotifier`)** | Set `ARTEMIS_NOTIFY_CMD="my-script --title '{title}' --message '{message}' --trace-id '{trace_id}'"` in your environment to execute any custom command or script upon event completion. | + ## 🧠 AI Agent Behavioral Rules (`rules.md`) When connecting ARTEMIS to AI coding assistants like **Antigravity**, **Claude Code**, **Cursor**, or **Windsurf**, providing the assistant with domain-specific testing discipline is critical for generating reliable test code. diff --git a/mcp_server/notifiers/__init__.py b/mcp_server/notifiers/__init__.py index 75626e1..ab6a040 100644 --- a/mcp_server/notifiers/__init__.py +++ b/mcp_server/notifiers/__init__.py @@ -21,6 +21,7 @@ from mcp_server.notifiers.base import BaseNotifier from mcp_server.notifiers.composite import CompositeNotifier from mcp_server.notifiers.desktop import DesktopNotifier from mcp_server.notifiers.file import FileNotifier +from mcp_server.notifiers.script import ScriptNotifier from mcp_server.notifiers.webhook import WebhookNotifier _default_notifier = CompositeNotifier() @@ -38,14 +39,20 @@ def notify( event_type: str = "completed", payload: dict[str, Any] | None = None, ) -> bool: - """Dispatches a notification across all available channels (AgentAPI, Webhook, Desktop, File).""" - return _default_notifier.notify( - conversation_id=conversation_id, - message=message, - title=title, - event_type=event_type, - payload=payload, - ) + """Dispatches a notification across all available channels (AgentAPI, Webhook, Desktop, Script, File).""" + try: + return _default_notifier.notify( + conversation_id=conversation_id, + message=message, + title=title, + event_type=event_type, + payload=payload, + ) + except Exception as e: + import logging + + logging.getLogger("mcp_server.notifiers").debug(f"Global notify error: {e}") + return False def notify_jetski(conversation_id: str, message: str) -> bool: @@ -58,6 +65,7 @@ __all__ = [ "AgentApiNotifier", "WebhookNotifier", "DesktopNotifier", + "ScriptNotifier", "FileNotifier", "CompositeNotifier", "get_default_notifier", diff --git a/mcp_server/notifiers/agentapi.py b/mcp_server/notifiers/agentapi.py index 94afef8..d0489db 100644 --- a/mcp_server/notifiers/agentapi.py +++ b/mcp_server/notifiers/agentapi.py @@ -29,84 +29,151 @@ logger = logging.getLogger("mcp_server.notifiers.agentapi") class AgentApiNotifier(BaseNotifier): """Notifier implementation for environments supporting the agentapi CLI (e.g. Jetski / Antigravity).""" + def __init__(self): + super().__init__() + try: + addr = os.environ.get("ANTIGRAVITY_LS_ADDRESS") + token = os.environ.get("ANTIGRAVITY_CSRF_TOKEN") + if addr and token: + self._save_shared_env(addr, token) + except Exception: + pass + @property def name(self) -> str: return "agentapi" def _find_agentapi_path(self) -> str | None: """Locates the agentapi executable.""" - which_path = shutil.which("agentapi") - if which_path: - return which_path + try: + which_path = shutil.which("agentapi") + if which_path: + return which_path - candidates = [ - os.path.expanduser("~/.gemini/jetski/bin/agentapi"), - os.path.expanduser("~/.artemis/bin/agentapi"), - os.path.expanduser("~/bin/agentapi"), - "/usr/local/bin/agentapi", - ] - for cand in candidates: - if os.path.exists(cand) and os.access(cand, os.X_OK): - return cand + candidates = [ + os.path.expanduser("~/.gemini/jetski/bin/agentapi"), + os.path.expanduser("~/.artemis/bin/agentapi"), + os.path.expanduser("~/bin/agentapi"), + "/usr/local/bin/agentapi", + ] + for cand in candidates: + if os.path.exists(cand) and os.access(cand, os.X_OK): + return cand + except Exception: + pass return None def is_available(self) -> bool: """Returns True if agentapi binary exists or LS address is detected.""" - if self._find_agentapi_path(): - return True - if "ANTIGRAVITY_LS_ADDRESS" in os.environ: - return True + try: + if self._find_agentapi_path(): + return True + if "ANTIGRAVITY_LS_ADDRESS" in os.environ: + return True + except Exception: + pass return False - def _load_shared_env(self, force_proc_scan: bool = False) -> None: - """Recovers and loads ANTIGRAVITY_LS_ADDRESS and ANTIGRAVITY_CSRF_TOKEN.""" + def _save_shared_env(self, addr: str, token: str) -> None: + """Persists recovered environment to shared files for faster subsequent access.""" + shared_candidates = [ + os.path.expanduser("~/.gemini/jetski/.jetski_env"), + os.path.expanduser("~/.artemis/.artemis_env"), + ] + try: + parent_dir = os.path.dirname( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + ) + shared_candidates.append(os.path.join(parent_dir, ".jetski_env")) + except Exception: + pass + + data = { + "ANTIGRAVITY_LS_ADDRESS": addr, + "ANTIGRAVITY_CSRF_TOKEN": token, + } + for shared_file in shared_candidates: + try: + os.makedirs(os.path.dirname(shared_file), exist_ok=True) + with open(shared_file, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2) + logger.debug(f"[EnvRecovery] Saved session to {shared_file}.") + except Exception as e: + logger.debug(f"[EnvRecovery] Error writing shared env file {shared_file}: {e}") + + def _get_candidate_envs(self, force_proc_scan: bool = False) -> list[tuple[str, str]]: + """Returns unique candidate (address, token) pairs ordered by priority and recency.""" + candidates: list[tuple[str, str]] = [] + seen: set[tuple[str, str]] = set() + + def _add(addr: str | None, token: str | None) -> None: + if addr and token and (addr, token) not in seen: + seen.add((addr, token)) + candidates.append((addr, token)) + + # 1. Current environment variables + _add( + os.environ.get("ANTIGRAVITY_LS_ADDRESS"), + os.environ.get("ANTIGRAVITY_CSRF_TOKEN"), + ) + + # 2. Shared env files (if not forcing fresh process scan) if not force_proc_scan: + shared_candidates = [ + os.path.expanduser("~/.gemini/jetski/.jetski_env"), + os.path.expanduser("~/.artemis/.artemis_env"), + ] try: - shared_candidates = [ - os.path.expanduser("~/.gemini/jetski/.jetski_env"), - os.path.expanduser("~/.artemis/.artemis_env"), - ] - # Project root fallback parent_dir = os.path.dirname( os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ) shared_candidates.append(os.path.join(parent_dir, ".jetski_env")) + except Exception: + pass - for shared_file in shared_candidates: - if os.path.exists(shared_file): + for shared_file in shared_candidates: + if os.path.exists(shared_file): + try: with open(shared_file, encoding="utf-8") as f: env_data = json.load(f) - for key, val in env_data.items(): - if val: - os.environ[key] = val - logger.debug(f"[EnvRecovery] Loaded {key} from {shared_file}.") - return - except Exception as e: - logger.debug(f"[EnvRecovery] Error reading shared env file: {e}") - - # Fallback to psutil process scan - logger.debug("[EnvRecovery] Scanning process table for active agent session...") - found_addr = None - found_token = None + _add( + env_data.get("ANTIGRAVITY_LS_ADDRESS"), + env_data.get("ANTIGRAVITY_CSRF_TOKEN"), + ) + except Exception as e: + logger.debug( + f"[EnvRecovery] Error reading shared env file {shared_file}: {e}" + ) + # 3. Scan process table for active agent sessions, sorted by newest creation time first + proc_matches: list[tuple[float, str, str]] = [] try: import psutil - for proc in psutil.process_iter(["environ"]): + for proc in psutil.process_iter(["create_time", "environ"]): try: env = proc.info.get("environ") - if env and "ANTIGRAVITY_LS_ADDRESS" in env and "ANTIGRAVITY_CSRF_TOKEN" in env: - found_addr = env["ANTIGRAVITY_LS_ADDRESS"] - found_token = env["ANTIGRAVITY_CSRF_TOKEN"] - break - except (psutil.AccessDenied, psutil.NoSuchProcess): + if ( + env + and "ANTIGRAVITY_LS_ADDRESS" in env + and "ANTIGRAVITY_CSRF_TOKEN" in env + ): + ctime = proc.info.get("create_time", 0.0) or 0.0 + proc_matches.append( + ( + ctime, + env["ANTIGRAVITY_LS_ADDRESS"], + env["ANTIGRAVITY_CSRF_TOKEN"], + ) + ) + except (psutil.AccessDenied, psutil.NoSuchProcess, Exception): continue except ImportError: - # Fallback on Linux /proc import glob for env_path in glob.glob("/proc/[0-9]*/environ"): try: + mtime = os.path.getmtime(env_path) with open(env_path, "rb") as f: content = f.read() env_vars = content.split(b"\x00") @@ -114,20 +181,33 @@ class AgentApiNotifier(BaseNotifier): local_token = None for var in env_vars: if var.startswith(b"ANTIGRAVITY_LS_ADDRESS="): - local_addr = var.split(b"=", 1)[1].decode("utf-8", errors="ignore") + local_addr = ( + var.split(b"=", 1)[1].decode("utf-8", errors="ignore") + ) elif var.startswith(b"ANTIGRAVITY_CSRF_TOKEN="): - local_token = var.split(b"=", 1)[1].decode("utf-8", errors="ignore") + local_token = ( + var.split(b"=", 1)[1].decode("utf-8", errors="ignore") + ) if local_addr and local_token: - found_addr = local_addr - found_token = local_token - break + proc_matches.append((mtime, local_addr, local_token)) except Exception: pass - if found_addr and found_token: - os.environ["ANTIGRAVITY_LS_ADDRESS"] = found_addr - os.environ["ANTIGRAVITY_CSRF_TOKEN"] = found_token - logger.debug(f"[EnvRecovery] Recovered session from process. Address: {found_addr}") + # Sort matches descending by creation timestamp so active/newest sessions come first + proc_matches.sort(key=lambda x: x[0], reverse=True) + for _, addr, token in proc_matches: + _add(addr, token) + + return candidates + + def _load_shared_env(self, force_proc_scan: bool = False) -> None: + """Recovers and loads the newest candidate ANTIGRAVITY_LS_ADDRESS and ANTIGRAVITY_CSRF_TOKEN.""" + candidates = self._get_candidate_envs(force_proc_scan=force_proc_scan) + if candidates: + addr, token = candidates[0] + os.environ["ANTIGRAVITY_LS_ADDRESS"] = addr + os.environ["ANTIGRAVITY_CSRF_TOKEN"] = token + logger.debug(f"[EnvRecovery] Loaded session candidate: {addr}") def notify( self, @@ -138,33 +218,74 @@ class AgentApiNotifier(BaseNotifier): payload: dict[str, Any] | None = None, ) -> bool: """Dispatches a notification via the agentapi send-message command.""" - if not conversation_id or conversation_id.lower() in ("default", "none", ""): - return False - - agentapi_path = self._find_agentapi_path() - if not agentapi_path: - return False - - if "ANTIGRAVITY_LS_ADDRESS" not in os.environ or "ANTIGRAVITY_CSRF_TOKEN" not in os.environ: - self._load_shared_env() - - cmd = [agentapi_path, "send-message"] - if title: - cmd.extend(["--title", title]) - cmd.extend([conversation_id, message]) - try: - subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=10) - logger.info("AgentAPI notification sent successfully.") - return True - except (subprocess.CalledProcessError, FileNotFoundError, Exception): - try: - self._load_shared_env(force_proc_scan=True) - if "ANTIGRAVITY_LS_ADDRESS" in os.environ: - subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=10) - logger.info("AgentAPI notification sent successfully on retry.") - return True - except Exception as retry_err: - logger.debug(f"AgentAPI notification retry failed: {retry_err}") + if not conversation_id or conversation_id.lower() in ("default", "none", ""): return False - return False + + agentapi_path = self._find_agentapi_path() + if not agentapi_path: + return False + + cmd = [agentapi_path, "send-message"] + if title: + cmd.append(f"--title={title}") + cmd.extend([conversation_id, message]) + + candidates = self._get_candidate_envs(force_proc_scan=False) + if not candidates: + candidates = self._get_candidate_envs(force_proc_scan=True) + + for attempt, (addr, token) in enumerate(candidates): + env = os.environ.copy() + env["ANTIGRAVITY_LS_ADDRESS"] = addr + env["ANTIGRAVITY_CSRF_TOKEN"] = token + try: + subprocess.run( + cmd, + capture_output=True, + text=True, + check=True, + timeout=10, + env=env, + ) + os.environ["ANTIGRAVITY_LS_ADDRESS"] = addr + os.environ["ANTIGRAVITY_CSRF_TOKEN"] = token + self._save_shared_env(addr, token) + logger.info(f"AgentAPI notification sent successfully via {addr}.") + return True + except Exception as err: + logger.debug( + f"Candidate session {addr} failed (attempt {attempt + 1}): {err}" + ) + + # If all initial candidates failed, force a fresh process scan in case a new session started + fresh_candidates = self._get_candidate_envs(force_proc_scan=True) + for addr, token in fresh_candidates: + if (addr, token) in candidates: + continue + env = os.environ.copy() + env["ANTIGRAVITY_LS_ADDRESS"] = addr + env["ANTIGRAVITY_CSRF_TOKEN"] = token + try: + subprocess.run( + cmd, + capture_output=True, + text=True, + check=True, + timeout=10, + env=env, + ) + os.environ["ANTIGRAVITY_LS_ADDRESS"] = addr + os.environ["ANTIGRAVITY_CSRF_TOKEN"] = token + self._save_shared_env(addr, token) + logger.info( + f"AgentAPI notification sent successfully on fresh scan via {addr}." + ) + return True + except Exception as err: + logger.debug(f"Fresh candidate session {addr} failed: {err}") + + return False + except Exception as e: + logger.debug(f"AgentApiNotifier encountered unhandled error: {e}") + return False diff --git a/mcp_server/notifiers/composite.py b/mcp_server/notifiers/composite.py index 7c8f63a..57ead8b 100644 --- a/mcp_server/notifiers/composite.py +++ b/mcp_server/notifiers/composite.py @@ -21,6 +21,7 @@ from mcp_server.notifiers.agentapi import AgentApiNotifier from mcp_server.notifiers.base import BaseNotifier from mcp_server.notifiers.desktop import DesktopNotifier from mcp_server.notifiers.file import FileNotifier +from mcp_server.notifiers.script import ScriptNotifier from mcp_server.notifiers.webhook import WebhookNotifier logger = logging.getLogger("mcp_server.notifiers.composite") @@ -37,6 +38,7 @@ class CompositeNotifier(BaseNotifier): FileNotifier(), AgentApiNotifier(), WebhookNotifier(), + ScriptNotifier(), DesktopNotifier(), ] @@ -45,7 +47,13 @@ class CompositeNotifier(BaseNotifier): return "composite" def is_available(self) -> bool: - return any(n.is_available() for n in self._notifiers) + for n in self._notifiers: + try: + if n.is_available(): + return True + except Exception: + continue + return False def register_notifier(self, notifier: BaseNotifier) -> None: """Registers an additional custom notifier.""" @@ -61,18 +69,19 @@ class CompositeNotifier(BaseNotifier): ) -> bool: any_success = False for notifier in self._notifiers: - if notifier.is_available(): - try: - success = notifier.notify( - conversation_id=conversation_id, - message=message, - title=title, - event_type=event_type, - payload=payload, - ) - if success: - any_success = True - except Exception as e: - logger.debug(f"Notifier '{notifier.name}' encountered error: {e}") + try: + if not notifier.is_available(): + continue + success = notifier.notify( + conversation_id=conversation_id, + message=message, + title=title, + event_type=event_type, + payload=payload, + ) + if success: + any_success = True + except Exception as e: + logger.debug(f"Notifier '{notifier.name}' encountered error: {e}") return any_success diff --git a/mcp_server/notifiers/desktop.py b/mcp_server/notifiers/desktop.py index 02b6687..7ee6799 100644 --- a/mcp_server/notifiers/desktop.py +++ b/mcp_server/notifiers/desktop.py @@ -27,14 +27,26 @@ logger = logging.getLogger("mcp_server.notifiers.desktop") class DesktopNotifier(BaseNotifier): - """Notifier that displays native desktop notifications when explicitly enabled.""" + """Notifier that displays native desktop notifications across macOS, Linux, and Windows.""" @property def name(self) -> str: return "desktop" def is_available(self) -> bool: - return os.getenv("ARTEMIS_DESKTOP_NOTIFY", "").lower() in ("1", "true", "yes") + try: + val = os.getenv("ARTEMIS_DESKTOP_NOTIFY", "").lower() + if val in ("0", "false", "no", "off"): + return False + if os.getenv("CI", "").lower() in ("1", "true", "yes") and val not in ("1", "true", "yes"): + return False + if sys.platform == "linux" and not shutil.which("notify-send"): + return False + if sys.platform == "darwin" and not shutil.which("osascript"): + return False + return True + except Exception: + return False def notify( self, diff --git a/mcp_server/notifiers/script.py b/mcp_server/notifiers/script.py new file mode 100644 index 0000000..6d88fd8 --- /dev/null +++ b/mcp_server/notifiers/script.py @@ -0,0 +1,90 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Custom Script / Command Hook Notifier for universal IDE and editor integration.""" + +import logging +import os +import subprocess +from typing import Any + +from mcp_server.notifiers.base import BaseNotifier + +logger = logging.getLogger("mcp_server.notifiers.script") + + +class ScriptNotifier(BaseNotifier): + """Notifier that executes a user-defined command or script when an event occurs. + + This adapter enables universal integration with any custom IDE, editor (Neovim/Emacs), + or automation platform by allowing users to define ARTEMIS_NOTIFY_CMD or MCP_NOTIFY_COMMAND. + Placeholders like {title}, {message}, {conversation_id}, {event_type}, and {trace_id} + are automatically replaced before execution. + """ + + ENV_VARS = [ + "ARTEMIS_NOTIFY_CMD", + "MCP_NOTIFY_COMMAND", + ] + + @property + def name(self) -> str: + return "script" + + def _get_command_template(self) -> str | None: + for var in self.ENV_VARS: + val = os.getenv(var) + if val and val.strip(): + return val.strip() + return None + + def is_available(self) -> bool: + return self._get_command_template() is not None + + def notify( + self, + conversation_id: str, + message: str, + title: str | None = None, + event_type: str = "completed", + payload: dict[str, Any] | None = None, + ) -> bool: + cmd_template = self._get_command_template() + if not cmd_template: + return False + + trace_id = (payload or {}).get("trace_id", "") + formatted_title = title or f"Artemis Task {event_type.capitalize()}" + + # Replace template placeholders safely + try: + cmd = ( + cmd_template.replace("{title}", str(formatted_title)) + .replace("{message}", str(message)) + .replace("{conversation_id}", str(conversation_id)) + .replace("{event_type}", str(event_type)) + .replace("{trace_id}", str(trace_id)) + ) + subprocess.run( + cmd, + shell=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=10, + ) + logger.info(f"Custom script notification command executed: {cmd[:60]}...") + return True + except Exception as e: + logger.warning(f"Failed to execute custom script notification: {e}") + return False diff --git a/tests/unit/mcp/test_notifiers.py b/tests/unit/mcp/test_notifiers.py index 9a02275..a43633f 100644 --- a/tests/unit/mcp/test_notifiers.py +++ b/tests/unit/mcp/test_notifiers.py @@ -94,12 +94,40 @@ def test_webhook_notifier_not_configured(monkeypatch): assert notifier.notify("conv-1", "hello") is False -def test_desktop_notifier_not_enabled(monkeypatch): - monkeypatch.delenv("ARTEMIS_DESKTOP_NOTIFY", raising=False) +def test_desktop_notifier_disabled(monkeypatch): + monkeypatch.setenv("ARTEMIS_DESKTOP_NOTIFY", "false") notifier = DesktopNotifier() assert notifier.is_available() is False +def test_desktop_notifier_enabled_by_default(monkeypatch): + monkeypatch.delenv("ARTEMIS_DESKTOP_NOTIFY", raising=False) + monkeypatch.delenv("CI", raising=False) + import shutil + import sys + + if sys.platform == "linux": + monkeypatch.setattr(shutil, "which", lambda cmd: "/usr/bin/notify-send" if cmd == "notify-send" else None) + elif sys.platform == "darwin": + monkeypatch.setattr(shutil, "which", lambda cmd: "/usr/bin/osascript" if cmd == "osascript" else None) + notifier = DesktopNotifier() + assert notifier.is_available() is True + + +def test_script_notifier(monkeypatch): + from mcp_server.notifiers.script import ScriptNotifier + + monkeypatch.delenv("ARTEMIS_NOTIFY_CMD", raising=False) + monkeypatch.delenv("MCP_NOTIFY_COMMAND", raising=False) + notifier = ScriptNotifier() + assert notifier.is_available() is False + + monkeypatch.setenv("ARTEMIS_NOTIFY_CMD", "echo 'Notify: {title} - {message}'") + assert notifier.is_available() is True + res = notifier.notify("conv-123", "Task done", title="Success", payload={"trace_id": "t-1"}) + assert res is True + + def test_composite_notifier_dispatch(): d1 = DummyNotifier(available=True, return_val=True) d2 = DummyNotifier(available=False, return_val=False) @@ -122,6 +150,29 @@ def test_composite_notifier_dispatch(): assert d3.called_with is not None +class BrokenNotifier(BaseNotifier): + @property + def name(self) -> str: + return "broken" + + def is_available(self) -> bool: + raise RuntimeError("Catastrophic error in is_available") + + def notify(self, conversation_id, message, title=None, event_type="completed", payload=None): + raise RuntimeError("Catastrophic error in notify") + + +def test_composite_notifier_fault_tolerance(): + broken = BrokenNotifier() + d1 = DummyNotifier(available=True, return_val=True) + composite = CompositeNotifier(notifiers=[broken, d1]) + + # Must never raise an exception even if a notifier crashes in is_available or notify + assert composite.is_available() is True + assert composite.notify("conv-test", "Message") is True + assert d1.called_with is not None + + def test_global_notify_function(): res = notify( conversation_id="conv-xyz", @@ -129,3 +180,52 @@ def test_global_notify_function(): payload={"trace_id": "test-trace"}, ) assert isinstance(res, bool) + + +def test_agentapi_notifier_candidate_recovery_and_retry(monkeypatch, tmp_path): + from mcp_server.notifiers.agentapi import AgentApiNotifier + import subprocess + + monkeypatch.delenv("ANTIGRAVITY_LS_ADDRESS", raising=False) + monkeypatch.delenv("ANTIGRAVITY_CSRF_TOKEN", raising=False) + + shared_file = tmp_path / ".test_jetski_env" + monkeypatch.setattr( + AgentApiNotifier, + "_get_candidate_envs", + lambda self, force_proc_scan=False: [ + ("localhost:9999", "bad-token"), + ("localhost:1234", "good-token"), + ], + ) + + saved_sessions = [] + monkeypatch.setattr( + AgentApiNotifier, + "_save_shared_env", + lambda self, addr, token: saved_sessions.append((addr, token)), + ) + monkeypatch.setattr( + AgentApiNotifier, + "_find_agentapi_path", + lambda self: "/mock/bin/agentapi", + ) + + calls = [] + + def mock_run(cmd, capture_output, text, check, timeout, env): + calls.append(env.get("ANTIGRAVITY_LS_ADDRESS")) + if env.get("ANTIGRAVITY_LS_ADDRESS") == "localhost:9999": + raise subprocess.CalledProcessError(1, cmd) + return subprocess.CompletedProcess(cmd, 0, stdout="success") + + monkeypatch.setattr(subprocess, "run", mock_run) + + notifier = AgentApiNotifier() + success = notifier.notify("conv-test-123", "Task done!") + assert success is True + assert calls == ["localhost:9999", "localhost:1234"] + assert saved_sessions == [("localhost:1234", "good-token")] + assert os.environ["ANTIGRAVITY_LS_ADDRESS"] == "localhost:1234" + assert os.environ["ANTIGRAVITY_CSRF_TOKEN"] == "good-token" +