diff --git a/src/toad/__init__.py b/src/toad/__init__.py index fef41e57c..15efc0f2f 100644 --- a/src/toad/__init__.py +++ b/src/toad/__init__.py @@ -1,33 +1,8 @@ -from typing import Literal, Mapping -import platform +from __future__ import annotations NAME = "toad" TITLE = "Toad" -type OS = Literal["linux", "macos", "windows", "*"] - -_system = platform.system() -_OS_map: dict[str, OS] = { - "Linux": "linux", - "Darwin": "macos", - "Windows": "windows", -} -os: OS = _OS_map.get(_system, "linux") - - -def get_os_matrix(matrix: Mapping[OS, str]) -> str | None: - """Get a value from a mapping where the key is an OS, falling back to a wildcard ("*"). - - Args: - matrix: A mapping where an OS literal is the key. - - Returns: - The value, if one is found, or `None`. - """ - if (result := matrix.get(os)) is None: - result = matrix.get("*") - return result - def get_version() -> str: """Get the current version of Toad. diff --git a/src/toad/acp/agent.py b/src/toad/acp/agent.py index a5f7c3125..c38a0fe14 100644 --- a/src/toad/acp/agent.py +++ b/src/toad/acp/agent.py @@ -18,7 +18,6 @@ from agent_comms.runtime_requests import GoalHistoryReply from agent_comms.ui_model.observation import ReadOwnerGoal -import toad from toad import paths from toad.acp.agent_configuration import ( ModelConfigurationSetting, @@ -48,6 +47,7 @@ from toad.agent_presentation import ACPAgentPresentation from toad.agent_schema import AgentDefinition from toad.core import events as core_events +from toad.platform import host def generate_datetime_filename( @@ -124,8 +124,7 @@ def __init__( @property def command(self) -> str | None: """The command used to launch the agent, or `None` if there isn't one.""" - acp_command = toad.get_os_matrix(self.definition.run_command) - return acp_command + return host.select(self.definition.run_command) def __rich_repr__(self) -> rich.repr.Result: yield self.project_root_path diff --git a/src/toad/agent_schema.py b/src/toad/agent_schema.py index a4de5f167..3bc114459 100644 --- a/src/toad/agent_schema.py +++ b/src/toad/agent_schema.py @@ -4,8 +4,7 @@ from typing import Annotated, Literal from agent_comms.declared_family import DeclaredFamily from agent_comms.field_codec import FieldCodec, TextRepresentation, WireName - -type OS = Literal['macos', 'linux', 'windows', '*'] +from toad.platform import Platform class AgentKind(DeclaredFamily, affix='AgentKind'): @@ -74,9 +73,8 @@ class AgentDefinition: active: bool = True recommended: bool = False - def commands_for(self, platform: OS) -> dict[str, Command]: - """Select the catalog's exact platform or its declared wildcard.""" - return self.actions.get(platform, self.actions.get('*', {})) + def commands_for(self, platform: type[Platform]) -> dict[str, Command]: + return platform.select(self.actions) or {} @classmethod def decode(cls, value: object) -> AgentDefinition: diff --git a/src/toad/platform.py b/src/toad/platform.py new file mode 100644 index 000000000..fa8ec47f4 --- /dev/null +++ b/src/toad/platform.py @@ -0,0 +1,129 @@ +"""The operating system Toad runs on, and the desktop notifier it offers.""" + +from __future__ import annotations + +import platform +from typing import ClassVar, Literal, Mapping + +type OS = Literal["linux", "macos", "windows", "*"] + + +class DesktopService: + """The platform's notification service, as notifypy detects it.""" + + def notification(self): + from notifypy import Notify + + return Notify() + + def send(self, app, notification) -> DesktopService: + notification.send() + return self + + +class DBusDesktop(DesktopService): + """notifypy's notify-send path inherits Toad's terminal as stderr; its D-Bus + notifier starts no child process that could write over the TUI.""" + + def notification(self): + from notifypy import Notify + from notifypy.os_notifiers.linux import LinuxNotifier + + return Notify(use_custom_notifier=LinuxNotifier) + + +class UncheckedDBusDesktop(DBusDesktop): + """Asks the session bus once whether anything serves notifications, because + notifypy's D-Bus notifier reports an error reply as a delivery.""" + + name: ClassVar[str] = "org.freedesktop.Notifications" + + def send(self, app, notification) -> DesktopService: + return self.served(app).send(app, notification) + + def served(self, app) -> DesktopService: + from jeepney import DBusAddress, new_method_call + from jeepney.io.blocking import open_dbus_connection + + bus = DBusAddress("/org/freedesktop/DBus", bus_name="org.freedesktop.DBus", + interface="org.freedesktop.DBus") + try: + with open_dbus_connection(bus="SESSION") as connection: + owned = connection.send_and_get_reply( + new_method_call(bus, "NameHasOwner", "s", (self.name,)), timeout=2).body[0] + activatable = self.name in connection.send_and_get_reply( + new_method_call(bus, "ListActivatableNames"), timeout=2).body[0] + except OSError: + owned = activatable = False + if owned or activatable: + return DBusDesktop() + app.call_from_thread( + app.notify, + "Desktop notifications are off for this session: no notification daemon is running.", + severity="warning", + ) + return AbsentDesktop() + + +class AbsentDesktop(DesktopService): + """Nothing serves notifications; asking again per message only fails again.""" + + def send(self, app, notification) -> DesktopService: + return self + + +class Platform: + """The operating system Toad runs on; each member owns what differs there.""" + + system: ClassVar[str] + key: ClassVar[OS] + application_name: ClassVar[str] = "Toad" + members: ClassVar[dict[str, type[Platform]]] = {} + + def __init_subclass__(cls, **kwargs) -> None: + super().__init_subclass__(**kwargs) + Platform.members[cls.system] = cls + + @classmethod + def select[V](cls, matrix: Mapping[OS, V]) -> V | None: + """The catalog's exact platform entry, or its declared wildcard.""" + return matrix.get(cls.key, matrix.get("*")) + + @classmethod + def desktop(cls) -> DesktopService: + return DesktopService() + + @classmethod + def decode(cls, system: str) -> type[Platform]: + """`platform.system()` names the member; anything undeclared is unknown.""" + return cls.members.get(system, UnknownPlatform) + + +class LinuxPlatform(Platform): + system = "Linux" + key = "linux" + + @classmethod + def desktop(cls) -> DesktopService: + return UncheckedDBusDesktop() + + +class MacPlatform(Platform): + system = "Darwin" + key = "macos" + application_name = "🐸 Toad" + + +class WindowsPlatform(Platform): + system = "Windows" + key = "windows" + + +class UnknownPlatform(Platform): + """A system Toad has no declaration for: the base defaults and the catalog wildcard.""" + + system = "" + key = "*" + + +host: type[Platform] = Platform.decode(platform.system()) diff --git a/src/toad/screens/agent_modal.py b/src/toad/screens/agent_modal.py index b26e47f77..5d0f5d87b 100644 --- a/src/toad/screens/agent_modal.py +++ b/src/toad/screens/agent_modal.py @@ -1,4 +1,3 @@ -from typing import cast from textual import on from textual import getters @@ -11,13 +10,13 @@ from textual import widgets from toad.widgets.footer import ToadFooter -import toad from textual.binding import Binding -from toad.agent_schema import AgentDefinition, OS +from toad.agent_schema import AgentDefinition from toad.catalog_actions import CatalogCommandAction, LaunchAction from toad.core.session_requests import LaunchAgent from toad.app import ToadApp from toad.preferences import SETTINGS +from toad.platform import host class DescriptionContainer(containers.VerticalScroll): @@ -47,7 +46,7 @@ def compose(self) -> ComposeResult: launcher_set = frozenset(app.settings.launcher.agents.splitlines()) agent = self.agent operations = [CatalogCommandAction.bind(name, command) - for name, command in agent.commands_for(cast(OS, toad.os)).items()] + for name, command in agent.commands_for(host).items()] choices = [*operations, LaunchAction(agent)] with containers.Vertical(id="container"): diff --git a/src/toad/setting_choices.py b/src/toad/setting_choices.py index 0d12bd17e..b65606673 100644 --- a/src/toad/setting_choices.py +++ b/src/toad/setting_choices.py @@ -2,7 +2,7 @@ from __future__ import annotations -from abc import ABC, abstractmethod +from abc import abstractmethod from dataclasses import dataclass from typing import ClassVar from random import choice @@ -10,9 +10,9 @@ from textual.content import Content from toad.widgets.future_text import FutureText -import toad from agent_comms.declared_family import DeclaredFamily from toad.setting_declarations import Choice +from toad.platform import DesktopService, host @dataclass(frozen=True) @@ -98,72 +98,8 @@ def shown(cls, count: int) -> bool: return False -class DesktopService: - """The platform's notification service, as notifypy detects it.""" - - def notification(self): - from notifypy import Notify - - return Notify() - - def send(self, app, notification) -> DesktopService: - notification.send() - return self - - -class DBusDesktop(DesktopService): - """notifypy's notify-send path inherits Toad's terminal as stderr; its D-Bus - notifier starts no child process that could write over the TUI.""" - - def notification(self): - from notifypy import Notify - from notifypy.os_notifiers.linux import LinuxNotifier - - return Notify(use_custom_notifier=LinuxNotifier) - - -class UncheckedDBusDesktop(DBusDesktop): - """Asks the session bus once whether anything serves notifications, because - notifypy's D-Bus notifier reports an error reply as a delivery.""" - - name: ClassVar[str] = "org.freedesktop.Notifications" - - def send(self, app, notification) -> DesktopService: - return self.served(app).send(app, notification) - - def served(self, app) -> DesktopService: - from jeepney import DBusAddress, new_method_call - from jeepney.io.blocking import open_dbus_connection - - bus = DBusAddress("/org/freedesktop/DBus", bus_name="org.freedesktop.DBus", - interface="org.freedesktop.DBus") - try: - with open_dbus_connection(bus="SESSION") as connection: - owned = connection.send_and_get_reply( - new_method_call(bus, "NameHasOwner", "s", (self.name,)), timeout=2).body[0] - activatable = self.name in connection.send_and_get_reply( - new_method_call(bus, "ListActivatableNames"), timeout=2).body[0] - except OSError: - owned = activatable = False - if owned or activatable: - return DBusDesktop() - app.call_from_thread( - app.notify, - "Desktop notifications are off for this session: no notification daemon is running.", - severity="warning", - ) - return AbsentDesktop() - - -class AbsentDesktop(DesktopService): - """Nothing serves notifications; asking again per message only fails again.""" - - def send(self, app, notification) -> DesktopService: - return self - - class NotificationPolicy(Choice, DeclaredFamily, affix="Notification"): - desktop: ClassVar[DesktopService] = UncheckedDBusDesktop() if toad.os == "linux" else DesktopService() + desktop: ClassVar[DesktopService] = host.desktop() sending: ClassVar[Lock] = Lock() @classmethod @@ -176,7 +112,7 @@ def deliver(cls, app, message: str, *, title: str = "", sound: str | None = None notification = cls.desktop.notification() notification.message = message notification.title = title - notification.application_name = "🐸 Toad" if toad.os == "macos" else "Toad" + notification.application_name = host.application_name if sound and app.settings.notifications.enable_sounds: notification.audio = str(files("toad.data").joinpath(f"sounds/{sound}.wav")) notification.icon = str(files("toad.data").joinpath("images/frog.png")) diff --git a/src/toad/terminal_attention.py b/src/toad/terminal_attention.py index efe1d9ed9..8499be983 100644 --- a/src/toad/terminal_attention.py +++ b/src/toad/terminal_attention.py @@ -146,7 +146,7 @@ def close(self) -> None: def notify(self, message: str, *, title: str = "", sound: str | None = None) -> None: """Dispatch the existing declared desktop policy on Textual's worker.""" policy = self.app.settings.notifications.system - # notifypy reports platform delivery failure as its own False result. + # The policy's desktop owns delivery failure: notifypy's D-Bus path reports an error reply as sent. self.app.run_worker(partial(policy.deliver, self.app, message, title=title, sound=sound), thread=True) diff --git a/tests/support/catalog.py b/tests/support/catalog.py index 07abfa1a9..6d3dd1012 100644 --- a/tests/support/catalog.py +++ b/tests/support/catalog.py @@ -13,6 +13,7 @@ import sys import tomllib from unittest.mock import patch +from toad.platform import host IDENTITY = "native-fixture" @@ -33,7 +34,6 @@ def catalog_text() -> str: A wildcard ``install`` action is declared too; this platform's own action must win over it. """ - import toad definition = { "identity": IDENTITY, "name": "A native catalog control", "short_name": "native-control", @@ -46,7 +46,7 @@ def catalog_text() -> str: text += '\n[run_command]\n"*" = ' + json.dumps(shlex.join([sys.executable, "-m", "agent_comms.acp"])) + "\n" text += '\n[actions."*".install]\ndescription = "Wrong platform"\ncommand = "exit 91"\n' for name, (label, command) in COMMANDS.items(): - text += f"\n[actions.{json.dumps(toad.os)}.{json.dumps(name)}]\n" + text += f"\n[actions.{json.dumps(host.key)}.{json.dumps(name)}]\n" text += f"description = {json.dumps(label)}\ncommand = {json.dumps(command)}\n" if name == "bootstrap": text += "bootstrap_uv = true\n" diff --git a/tests/support/catalog_steps.py b/tests/support/catalog_steps.py index 4cf5c12f2..48ab80653 100644 --- a/tests/support/catalog_steps.py +++ b/tests/support/catalog_steps.py @@ -3,6 +3,7 @@ from __future__ import annotations from textual import widgets +from toad.platform import host async def open_agent(ui, item) -> None: @@ -22,17 +23,16 @@ async def open_agent(ui, item) -> None: async def choose_action(ui, modal, name: str | None) -> None: """Pick action `name` in the agent dialog (None: the launch entry after the actions) and run it.""" - import toad _app, pilot = ui.textual - order = list(modal.agent.commands_for(toad.os)) + order = list(modal.agent.commands_for(host)) index = len(order) if name is None else order.index(name) modal.action_select.focus() await ui.press("enter", "home", *(["down"] * (index + 1)), "enter") if name is not None: chosen = modal.action_select.value assert chosen.name == name, (name, chosen) - assert chosen.command is modal.agent.commands_for(toad.os)[name] + assert chosen.command is modal.agent.commands_for(host)[name] assert await pilot.click(modal.query_one("#run-action", widgets.Button)) diff --git a/tests/textual/test_store_agent_actions.py b/tests/textual/test_store_agent_actions.py index ce00a1d50..1d449edab 100644 --- a/tests/textual/test_store_agent_actions.py +++ b/tests/textual/test_store_agent_actions.py @@ -16,6 +16,7 @@ from toad.screens.command_edit_modal import CommandEditModal from toad.screens.store import AgentItem, StoreScreen from support.ui import TextualUi +from toad.platform import host class AuditAction(CatalogCommandAction): @@ -56,9 +57,8 @@ async def test_agent_actions_run_edit_fail_cancel_and_take_input(workspace): await open_agent(ui, item) modal = app.screen - import toad - install = agent.commands_for(toad.os)["install"] + install = agent.commands_for(host)["install"] # Cancelling the editor runs nothing. await choose_action(ui, modal, "install") await ui.until(lambda: isinstance(app.screen, CommandEditModal), why="the command editor")