From bd130d4b36504ebc8e79f29bdd489efb1867a09b Mon Sep 17 00:00:00 2001 From: Jack Arturo Date: Fri, 17 Apr 2026 20:41:10 +0200 Subject: [PATCH 1/3] feat(profile): guard writes against the running Elgato desktop app MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Elgato Stream Deck app keeps every profile in memory and rewrites the on-disk manifests from its snapshot when it quits. Any edit made while the app is running is silently discarded the next time the app closes, which is exactly what bit a live-hardware smoke test of the previous PR. - Raise a new StreamDeckAppRunningError from write_page when the app is running, with a remediation message pointing callers at the auto_quit_app opt-in and the follow-up restart_app call. - Add auto_quit_app=False param on write_page. When True and the app is running, quit it cleanly (AppleScript quit first, killall fallback) before writing; the report is surfaced in the write_page response. - Rework restart_app to launch by explicit bundle path. The old `open -a "Stream Deck"` / `open -a "Elgato Stream Deck"` name lookup returns LSOpen error -600 on some macOS installs; launching by path bypasses LaunchServices. STREAMDECK_APP_PATH env var overrides the default `/Applications/Elgato Stream Deck.app`. - Expose reusable is_stream_deck_app_running() and stop_stream_deck_app() helpers so other future entrypoints can share the same lifecycle logic. - Update the README workflow section to spell out the quit → write → relaunch cycle and the env-var escape hatch. - New tests (5) cover the refuse/auto-quit branches, the restart-by-path path, and the missing-bundle error. An autouse fixture stubs the process check so existing tests are unaffected regardless of whether the contributor has the app running locally. Scope: unchanged tool count (6). auto_quit_app is a single additive field on the existing write_page tool. Co-Authored-By: Claude Opus 4.7 (1M context) --- README.md | 13 ++- profile_manager.py | 144 ++++++++++++++++++++++++++++------ profile_server.py | 21 ++++- tests/test_profile_manager.py | 125 ++++++++++++++++++++++++++++- 4 files changed, 277 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index c3bd7e7..44a20ec 100644 --- a/README.md +++ b/README.md @@ -77,9 +77,20 @@ uvx --from streamdeck-mcp streamdeck-mcp-usb - Generated icons are stored in `~/.streamdeck-mcp/generated-icons/`. - Generated shell scripts are stored in `~/StreamDeckScripts/`. +## Editing Workflow (Important) + +The Elgato desktop app keeps every profile in memory and rewrites the on-disk manifests from that snapshot when it quits, so any edit made while the app is running is wiped the next time it closes. The profile writer enforces a quit → write → relaunch cycle: + +1. Ensure the Elgato app is not running, or pass `auto_quit_app: true` to `streamdeck_write_page` to have it quit the app for you (AppleScript first, `killall` fallback). +2. Make as many `streamdeck_write_page` calls as you need — the app stays quit across them. +3. Call `streamdeck_restart_app` when you are done. The device re-reads the manifests on launch and your changes appear. + +`streamdeck_write_page` raises a `StreamDeckAppRunningError` when the app is running and `auto_quit_app` is not set, so you cannot accidentally write changes that will be silently discarded. + +If your Elgato app is installed somewhere other than `/Applications/Elgato Stream Deck.app`, set `STREAMDECK_APP_PATH` to the bundle path. + ## Usage Notes -- After writing profiles, the Elgato desktop app may need a restart to pick up the new manifests. - `streamdeck_create_action` is the safest way to build shell-command buttons because it writes a standalone script and returns the native Open action block for it. - The profile writer does not require exclusive USB access. diff --git a/profile_manager.py b/profile_manager.py index fac1962..2e58cf1 100644 --- a/profile_manager.py +++ b/profile_manager.py @@ -20,6 +20,7 @@ import string import subprocess import sys +import time import uuid from dataclasses import dataclass from pathlib import Path @@ -55,6 +56,13 @@ "UI Stream Deck": (4, 2), } +# The Elgato Stream Deck desktop app caches every profile in memory and rewrites the +# on-disk manifests when it quits, so any edit made while it is running gets clobbered +# the next time the user closes or restarts the app. +STREAM_DECK_APP_PROCESS_NAMES = ("Stream Deck", "Elgato Stream Deck") +DEFAULT_STREAM_DECK_APP_PATH = Path("/Applications/Elgato Stream Deck.app") +STREAM_DECK_APP_PATH_ENV = "STREAMDECK_APP_PATH" + HEX_COLOR_PATTERN = re.compile(r"^#[0-9a-fA-F]{6}$") POSITION_PATTERN = re.compile(r"^\d+,\d+$") UUID_PATTERN = re.compile( @@ -87,6 +95,16 @@ class ProfileValidationError(ProfileManagerError): """Raised when inputs for profile operations are invalid.""" +class StreamDeckAppRunningError(ProfileManagerError): + """Raised when a write is attempted while the Elgato desktop app is running. + + The app rewrites every profile manifest from its in-memory snapshot on quit, so + writes made while it is running are silently discarded. Callers must quit the + app first (pass `auto_quit_app=True` to `write_page`) and then call + `restart_app` once their edits are complete to see the changes. + """ + + @dataclass class PageRef: """Resolved page directory metadata.""" @@ -226,6 +244,79 @@ def _count_icons(page_dir: Path) -> int: return len([path for path in images_dir.iterdir() if path.is_file()]) +def _resolve_app_path() -> Path: + override = os.environ.get(STREAM_DECK_APP_PATH_ENV) + if override: + return Path(override).expanduser() + return DEFAULT_STREAM_DECK_APP_PATH + + +def is_stream_deck_app_running() -> bool: + """Return True if the Elgato Stream Deck desktop app is currently running.""" + + if sys.platform != "darwin": + return False + + for name in STREAM_DECK_APP_PROCESS_NAMES: + result = subprocess.run( + ["pgrep", "-x", name], + capture_output=True, + text=True, + check=False, + ) + if result.returncode == 0 and result.stdout.strip(): + return True + return False + + +def stop_stream_deck_app(*, graceful_timeout: float = 3.0) -> dict[str, Any]: + """Quit the Elgato Stream Deck desktop app. + + Tries an AppleScript quit first so the app can persist any unrelated state, then + falls back to `killall` if it does not exit in time. Returns a small report about + which path was taken. + """ + + if sys.platform != "darwin": + return {"stopped": False, "reason": "non-darwin platform"} + + if not is_stream_deck_app_running(): + return {"stopped": False, "reason": "not running"} + + graceful: list[str] = [] + for name in STREAM_DECK_APP_PROCESS_NAMES: + result = subprocess.run( + ["osascript", "-e", f'tell application "{name}" to quit'], + capture_output=True, + text=True, + check=False, + ) + if result.returncode == 0: + graceful.append(name) + + deadline = time.monotonic() + graceful_timeout + while time.monotonic() < deadline and is_stream_deck_app_running(): + time.sleep(0.2) + + forced: list[str] = [] + if is_stream_deck_app_running(): + for name in STREAM_DECK_APP_PROCESS_NAMES: + result = subprocess.run( + ["killall", name], + capture_output=True, + text=True, + check=False, + ) + if result.returncode == 0: + forced.append(name) + + return { + "stopped": not is_stream_deck_app_running(), + "graceful": graceful, + "forced": forced, + } + + class ProfileManager: """Read and write Elgato Stream Deck profiles.""" @@ -356,9 +447,21 @@ def write_page( clear_existing: bool = True, create_new: bool = False, make_current: bool = False, + auto_quit_app: bool = False, ) -> dict[str, Any]: """Create a page or rewrite an existing page manifest.""" + app_stop_report: dict[str, Any] | None = None + if is_stream_deck_app_running(): + if not auto_quit_app: + raise StreamDeckAppRunningError( + "The Elgato Stream Deck app is running and will overwrite this " + "edit on quit. Retry with auto_quit_app=True to quit it first, " + "then call streamdeck_restart_app once your edits are complete " + "to apply the changes." + ) + app_stop_report = stop_stream_deck_app() + profile_dir, profile_manifest = self._resolve_profile( profile_name=profile_name, profile_id=profile_id ) @@ -434,6 +537,7 @@ def write_page( "button_count": len(actions), "page_name": page_manifest.get("Name", ""), "manifest_path": str(page_dir / "manifest.json"), + "app_quit": app_stop_report, } def create_icon( @@ -515,40 +619,34 @@ def restart_app(self) -> dict[str, Any]: "streamdeck_restart_app is currently only supported on macOS." ) - killed = [] - for app_name in ("Elgato Stream Deck", "Stream Deck"): - result = subprocess.run( - ["killall", app_name], - capture_output=True, - text=True, - check=False, + stop_report = stop_stream_deck_app() + + app_path = _resolve_app_path() + if not app_path.exists(): + raise ProfileManagerError( + f"Stream Deck app not found at {app_path}. " + f"Set {STREAM_DECK_APP_PATH_ENV} to override the default install path." ) - if result.returncode == 0: - killed.append(app_name) - open_result = subprocess.run( - ["open", "-a", "Stream Deck"], + # `open -a ` relies on LaunchServices name lookup, which returns error + # -600 on some systems even when the bundle is present. Launching by explicit + # path bypasses that lookup. + result = subprocess.run( + ["open", str(app_path)], capture_output=True, text=True, check=False, ) - if open_result.returncode != 0: - fallback = subprocess.run( - ["open", "-a", "Elgato Stream Deck"], - capture_output=True, - text=True, - check=False, - ) - open_result = fallback - - if open_result.returncode != 0: + if result.returncode != 0: + message = result.stderr.strip() or result.stdout.strip() raise ProfileManagerError( - open_result.stderr.strip() or "Failed to relaunch Stream Deck app." + f"Failed to relaunch Stream Deck ({app_path}): {message or 'unknown error'}" ) return { - "killed": killed, "restarted": True, + "app_path": str(app_path), + "stop": stop_report, } def _resolve_profile( diff --git a/profile_server.py b/profile_server.py index dfeaa1e..e5be62d 100644 --- a/profile_server.py +++ b/profile_server.py @@ -23,6 +23,7 @@ ProfileManagerError, ProfileNotFoundError, ProfileValidationError, + StreamDeckAppRunningError, ) logging.basicConfig( @@ -172,7 +173,12 @@ async def list_tools() -> list[Tool]: Tool( name="streamdeck_write_page", description=( - "Create a new page or replace/update an existing Stream Deck desktop page manifest." + "Create a new page or replace/update an existing Stream Deck desktop " + "page manifest. IMPORTANT: the Elgato desktop app overwrites profile " + "manifests from its in-memory state on quit, so writes made while the " + "app is running are lost. This tool refuses to write when the app is " + "running unless auto_quit_app=True is passed. Call streamdeck_restart_app " + "once your edits are complete to make the changes visible on the device." ), inputSchema={ "type": "object", @@ -210,6 +216,16 @@ async def list_tools() -> list[Tool]: "When true, make the page the active current page after writing." ), }, + "auto_quit_app": { + "type": "boolean", + "description": ( + "If true and the Elgato Stream Deck desktop app is " + "running, quit it (graceful AppleScript first, then " + "killall) before writing. Required when the app is " + "running or the write will raise an error. Defaults to " + "false so callers must explicitly consent to quitting it." + ), + }, }, }, ), @@ -305,6 +321,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: clear_existing=arguments.get("clear_existing", True), create_new=arguments.get("create_new", False), make_current=arguments.get("make_current", False), + auto_quit_app=arguments.get("auto_quit_app", False), ) return [TextContent(type="text", text=json.dumps(result, indent=2))] @@ -333,6 +350,8 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: return [TextContent(type="text", text=f"❌ Unknown tool: {name}")] + except StreamDeckAppRunningError as exc: + return [TextContent(type="text", text=f"⚠️ {exc}")] except ( ProfileManagerError, ProfileNotFoundError, diff --git a/tests/test_profile_manager.py b/tests/test_profile_manager.py index ee5b1b8..58befaa 100644 --- a/tests/test_profile_manager.py +++ b/tests/test_profile_manager.py @@ -9,7 +9,21 @@ import pytest -from profile_manager import PageNotFoundError, ProfileManager, ProfileValidationError +from profile_manager import ( + PageNotFoundError, + ProfileManager, + ProfileManagerError, + ProfileValidationError, + StreamDeckAppRunningError, +) + + +@pytest.fixture(autouse=True) +def _stub_stream_deck_app_not_running(): + """Default: act as if the Stream Deck app is not running so tests hit the write path.""" + + with patch("profile_manager.is_stream_deck_app_running", return_value=False): + yield def _write_json(path: Path, payload: dict) -> None: @@ -364,3 +378,112 @@ def test_read_page_requires_locator(sample_profiles_v3: Path, tmp_path: Path) -> with pytest.raises(PageNotFoundError): manager.read_page(profile_name="Default Profile", directory_id="DOES-NOT-EXIST") + + +def test_write_page_refuses_while_app_running(sample_profiles_v3: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + with patch("profile_manager.is_stream_deck_app_running", return_value=True): + with pytest.raises(StreamDeckAppRunningError, match="running"): + manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + ) + + +def test_write_page_auto_quit_app_stops_then_writes( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + stop_report = {"stopped": True, "graceful": ["Stream Deck"], "forced": []} + + with ( + patch("profile_manager.is_stream_deck_app_running", return_value=True), + patch("profile_manager.stop_stream_deck_app", return_value=stop_report) as stop_mock, + ): + result = manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + auto_quit_app=True, + ) + + stop_mock.assert_called_once() + assert result["app_quit"] == stop_report + page = manager.read_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + ) + assert page["buttons"][0]["title"] == "x" + + +def test_write_page_records_no_quit_when_app_not_running( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + # autouse fixture already pins is_stream_deck_app_running to False + result = manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + ) + assert result["app_quit"] is None + + +def test_restart_app_launches_by_explicit_path(sample_profiles_v3: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + fake_app = tmp_path / "Fake Stream Deck.app" + fake_app.mkdir() + open_call = {"args": None} + + def _fake_run(cmd, **kwargs): + open_call["args"] = cmd + return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})() + + with ( + patch("profile_manager.sys.platform", "darwin"), + patch.dict("os.environ", {"STREAMDECK_APP_PATH": str(fake_app)}), + patch("profile_manager.stop_stream_deck_app", return_value={"stopped": True}), + patch("profile_manager.subprocess.run", side_effect=_fake_run), + ): + result = manager.restart_app() + + assert open_call["args"] == ["open", str(fake_app)] + assert result["app_path"] == str(fake_app) + assert result["restarted"] is True + + +def test_restart_app_errors_when_app_missing(sample_profiles_v3: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + missing_app = tmp_path / "nope.app" + + with ( + patch("profile_manager.sys.platform", "darwin"), + patch.dict("os.environ", {"STREAMDECK_APP_PATH": str(missing_app)}), + patch("profile_manager.stop_stream_deck_app", return_value={"stopped": True}), + ): + with pytest.raises(ProfileManagerError, match="not found"): + manager.restart_app() From 3bbc9e886216c0f409ff4c7469df26cef6194985 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 17 Apr 2026 18:56:24 +0000 Subject: [PATCH 2/3] fix: address review feedback - consistent stop schema, verify stop before write, validate path before stop Agent-Logs-Url: https://github.com/verygoodplugins/streamdeck-mcp/sessions/bf3b8260-ea35-455c-99cf-579444999ff4 Co-authored-by: jack-arturo <13076544+jack-arturo@users.noreply.github.com> --- profile_manager.py | 13 +++++++++---- tests/test_profile_manager.py | 26 ++++++++++++++++++++++++-- 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/profile_manager.py b/profile_manager.py index 2e58cf1..bd15be0 100644 --- a/profile_manager.py +++ b/profile_manager.py @@ -278,10 +278,10 @@ def stop_stream_deck_app(*, graceful_timeout: float = 3.0) -> dict[str, Any]: """ if sys.platform != "darwin": - return {"stopped": False, "reason": "non-darwin platform"} + return {"stopped": False, "graceful": [], "forced": [], "reason": "non-darwin platform"} if not is_stream_deck_app_running(): - return {"stopped": False, "reason": "not running"} + return {"stopped": False, "graceful": [], "forced": [], "reason": "not running"} graceful: list[str] = [] for name in STREAM_DECK_APP_PROCESS_NAMES: @@ -461,6 +461,11 @@ def write_page( "to apply the changes." ) app_stop_report = stop_stream_deck_app() + if not app_stop_report.get("stopped", False) or is_stream_deck_app_running(): + raise StreamDeckAppRunningError( + "The Elgato Stream Deck app could not be stopped. Aborting page " + "write because the running app may overwrite these edits on quit." + ) profile_dir, profile_manifest = self._resolve_profile( profile_name=profile_name, profile_id=profile_id @@ -619,8 +624,6 @@ def restart_app(self) -> dict[str, Any]: "streamdeck_restart_app is currently only supported on macOS." ) - stop_report = stop_stream_deck_app() - app_path = _resolve_app_path() if not app_path.exists(): raise ProfileManagerError( @@ -628,6 +631,8 @@ def restart_app(self) -> dict[str, Any]: f"Set {STREAM_DECK_APP_PATH_ENV} to override the default install path." ) + stop_report = stop_stream_deck_app() + # `open -a ` relies on LaunchServices name lookup, which returns error # -600 on some systems even when the bundle is present. Launching by explicit # path bypasses that lookup. diff --git a/tests/test_profile_manager.py b/tests/test_profile_manager.py index 58befaa..a7630d2 100644 --- a/tests/test_profile_manager.py +++ b/tests/test_profile_manager.py @@ -407,7 +407,7 @@ def test_write_page_auto_quit_app_stops_then_writes( stop_report = {"stopped": True, "graceful": ["Stream Deck"], "forced": []} with ( - patch("profile_manager.is_stream_deck_app_running", return_value=True), + patch("profile_manager.is_stream_deck_app_running", side_effect=[True, False]), patch("profile_manager.stop_stream_deck_app", return_value=stop_report) as stop_mock, ): result = manager.write_page( @@ -426,6 +426,29 @@ def test_write_page_auto_quit_app_stops_then_writes( assert page["buttons"][0]["title"] == "x" +def test_write_page_auto_quit_app_raises_when_stop_fails( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + stop_report = {"stopped": False, "graceful": [], "forced": [], "reason": "killall failed"} + + with ( + patch("profile_manager.is_stream_deck_app_running", return_value=True), + patch("profile_manager.stop_stream_deck_app", return_value=stop_report), + ): + with pytest.raises(StreamDeckAppRunningError, match="could not be stopped"): + manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + auto_quit_app=True, + ) + + def test_write_page_records_no_quit_when_app_not_running( sample_profiles_v3: Path, tmp_path: Path ) -> None: @@ -483,7 +506,6 @@ def test_restart_app_errors_when_app_missing(sample_profiles_v3: Path, tmp_path: with ( patch("profile_manager.sys.platform", "darwin"), patch.dict("os.environ", {"STREAMDECK_APP_PATH": str(missing_app)}), - patch("profile_manager.stop_stream_deck_app", return_value={"stopped": True}), ): with pytest.raises(ProfileManagerError, match="not found"): manager.restart_app() From 1fce5e86d79f48fb7bbdda6f65a520baeb02133b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 17 Apr 2026 18:57:42 +0000 Subject: [PATCH 3/3] fix: include stop reason in error message when app stop fails Agent-Logs-Url: https://github.com/verygoodplugins/streamdeck-mcp/sessions/bf3b8260-ea35-455c-99cf-579444999ff4 Co-authored-by: jack-arturo <13076544+jack-arturo@users.noreply.github.com> --- profile_manager.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/profile_manager.py b/profile_manager.py index bd15be0..b9d2058 100644 --- a/profile_manager.py +++ b/profile_manager.py @@ -461,10 +461,14 @@ def write_page( "to apply the changes." ) app_stop_report = stop_stream_deck_app() - if not app_stop_report.get("stopped", False) or is_stream_deck_app_running(): + stop_failed = not app_stop_report.get("stopped", False) + still_running = is_stream_deck_app_running() + if stop_failed or still_running: + reason = app_stop_report.get("reason", "") + detail = f" Reason: {reason}." if reason else "" raise StreamDeckAppRunningError( - "The Elgato Stream Deck app could not be stopped. Aborting page " - "write because the running app may overwrite these edits on quit." + f"The Elgato Stream Deck app could not be stopped.{detail} Aborting " + "page write because the running app may overwrite these edits on quit." ) profile_dir, profile_manifest = self._resolve_profile(