Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 1 addition & 26 deletions src/toad/__init__.py
Original file line number Diff line number Diff line change
@@ -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.
Expand Down
5 changes: 2 additions & 3 deletions src/toad/acp/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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
Expand Down
8 changes: 3 additions & 5 deletions src/toad/agent_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -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'):
Expand Down Expand Up @@ -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:
Expand Down
129 changes: 129 additions & 0 deletions src/toad/platform.py
Original file line number Diff line number Diff line change
@@ -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())
7 changes: 3 additions & 4 deletions src/toad/screens/agent_modal.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from typing import cast

from textual import on
from textual import getters
Expand All @@ -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):
Expand Down Expand Up @@ -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"):
Expand Down
72 changes: 4 additions & 68 deletions src/toad/setting_choices.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,17 +2,17 @@

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
from threading import Lock
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)
Expand Down Expand Up @@ -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
Expand All @@ -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"))
Expand Down
2 changes: 1 addition & 1 deletion src/toad/terminal_attention.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
4 changes: 2 additions & 2 deletions tests/support/catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
import sys
import tomllib
from unittest.mock import patch
from toad.platform import host

IDENTITY = "native-fixture"

Expand All @@ -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",
Expand All @@ -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"
Expand Down
Loading
Loading