Desktop notifications protocol: Add support for querying if the terminal emulator supports the protocol

Fixes #7658

docs/changelog.rst

@@ -108,6 +108,8 @@ Detailed list of changes
 
 - Allow controlling the easing curves used for :opt:`visual_bell_duration`
 
+- Desktop notifications protocol: Add support for querying if the terminal emulator supports the protocol (:iss:`7658`)
+
 0.35.2 [2024-06-22]
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

docs/desktop-notifications.rst

@@ -93,6 +93,38 @@ to display it based on what it does understand.
    Similarly, features such as scheduled notifications could be added in future
    revisions.
 
+Querying for support
+-------------------------
+
+An application can query the terminal emulator for support of this protocol, by
+sending the following escape code::
+
+    <OSC> 99 ; ? ; <terminator>
+
+A conforming terminal must respond with an escape code of the form::
+
+    <OSC> 99 ; ? ; key=value : key=value <terminator>
+
+Here, the ``key=value`` parts specify details about what the terminal
+implementation supports. Currently, the following keys are defined:
+
+=======  ================================================================================
+Key      Value
+=======  ================================================================================
+``a``    Comma separated list of actions from the ``a`` key that the terminal
+         implements. If no actions are supported, the a key must be absent form the
+         query response.
+
+``o``    Comma separated list of occassions from the ``o`` key that the
+         terminal implements. If no occassions are supported, the value
+         ``o=always`` must be sent in the query response.
+=======  ================================================================================
+
+In the future, if this protocol expands, more keys might be added. Clients must
+ignore keys they dont understand in the query response.
+
+Specification of all keys used in the protocol
+--------------------------------------------------
 
 =======  ====================  ========== =================
 Key      Value                 Default    Description

kitty/notify.py

@@ -7,7 +7,7 @@ from collections import OrderedDict
 from contextlib import suppress
 from enum import Enum
 from itertools import count
-from typing import Callable, Dict, Optional
+from typing import Callable, Dict, FrozenSet, Optional
 
 from .constants import is_macos, logo_png_file
 from .fast_data_types import current_focused_os_window_id, get_boss
@@ -85,20 +85,25 @@ class OnlyWhen(Enum):
     invisible = 'invisible'
 
 
+class Action(Enum):
+    focus = 'focus'
+    report = 'report'
+
+
 class NotificationCommand:
 
     done: bool = True
     identifier: str = '0'
     title: str = ''
     body: str = ''
-    actions: str = ''
+    actions: FrozenSet[Action] = frozenset((Action.focus,))
     only_when: OnlyWhen = OnlyWhen.unset
     urgency: Optional[Urgency] = None
 
     def __repr__(self) -> str:
         return (
             f'NotificationCommand(identifier={self.identifier!r}, title={self.title!r}, body={self.body!r},'
-            f'actions={self.actions!r}, done={self.done!r}, urgency={self.urgency})')
+            f'actions={self.actions}, done={self.done!r}, urgency={self.urgency})')
 
 
 def parse_osc_9(raw: str) -> NotificationCommand:
@@ -125,9 +130,20 @@ def sanitize_id(v: str) -> str:
     return sanitize_identifier_pat().sub('', v)
 
 
+class QueryResponse(Exception):
+
+    def __init__(self, response_string: str):
+        super().__init__('not an error')
+        self.response_string = response_string
+
+
 def parse_osc_99(raw: str) -> NotificationCommand:
     cmd = NotificationCommand()
     metadata, payload = raw.partition(';')[::2]
+    if metadata == '?':
+        actions = ','.join(x.name for x in Action)
+        when = ','.join(x.name for x in OnlyWhen if x.value)
+        raise QueryResponse(f'99;?;a={actions}:o={when}')
     payload_is_encoded = False
     payload_type = 'title'
     if metadata:
@@ -146,7 +162,18 @@ def parse_osc_99(raw: str) -> NotificationCommand:
             elif k == 'd':
                 cmd.done = v != '0'
             elif k == 'a':
-                cmd.actions += f',{v}'
+                for ax in v.split(','):
+                    if remove := ax.startswith('-'):
+                        ax = ax.lstrip('+-')
+                    try:
+                        ac = Action(ax)
+                    except ValueError:
+                        pass
+                    else:
+                        if remove:
+                            cmd.actions -= {ac}
+                        else:
+                            cmd.actions = cmd.actions.union({ac})
             elif k == 'o':
                 with suppress(ValueError):
                     cmd.only_when = OnlyWhen(v)
@@ -178,7 +205,7 @@ def limit_size(x: str) -> str:
 def merge_osc_99(prev: NotificationCommand, cmd: NotificationCommand) -> NotificationCommand:
     if prev.done or prev.identifier != cmd.identifier:
         return cmd
-    cmd.actions = limit_size(f'{prev.actions},{cmd.actions}')
+    cmd.actions = prev.actions.union(cmd.actions)
     cmd.title = limit_size(prev.title + cmd.title)
     cmd.body = limit_size(prev.body + cmd.body)
     if cmd.only_when is OnlyWhen.unset:
@@ -195,18 +222,16 @@ id_counter = count()
 class RegisteredNotification:
     identifier: str
     window_id: int
-    focus: bool = True
+    focus: bool = False
     report: bool = False
 
     def __init__(self, cmd: NotificationCommand, window_id: int):
         self.window_id = window_id
-        for x in cmd.actions.strip(',').split(','):
-            val = not x.startswith('-')
-            x = x.lstrip('+-')
-            if x == 'focus':
-                self.focus = val
-            elif x == 'report':
-                self.report = val
+        for x in cmd.actions:
+            if x is Action.focus:
+                self.focus = True
+            elif x is Action.report:
+                self.report = True
         self.identifier = cmd.identifier
 
 

kitty/window.py

@@ -94,6 +94,7 @@ from .notify import (
     NotificationCommand,
     NotifyImplementation,
     OnlyWhen,
+    QueryResponse,
     Urgency,
     handle_notification_cmd,
     notify_with_command,
@@ -1077,9 +1078,12 @@ class Window:
                 log_error(f'Ignoring unknown OSC 777: {raw_data}')
                 return  # unknown OSC 777
             raw_data = raw_data[len('notify;'):]
-        cmd = handle_notification_cmd(osc_code, raw_data, self.id, self.prev_osc99_cmd)
-        if cmd is not None and osc_code == 99:
-            self.prev_osc99_cmd = cmd
+        try:
+            cmd = handle_notification_cmd(osc_code, raw_data, self.id, self.prev_osc99_cmd)
+            if cmd is not None and osc_code == 99:
+                self.prev_osc99_cmd = cmd
+        except QueryResponse as err:
+            self.screen.send_escape_code_to_child(ESC_OSC, err.response_string)
 
     def on_mouse_event(self, event: Dict[str, Any]) -> bool:
         event['mods'] = event.get('mods', 0) & mod_mask
@@ -1492,7 +1496,6 @@ class Window:
             cmd.title = 'kitty'
             s = self.last_cmd_cmdline.replace('\\\n', ' ')
             cmd.body = f'Command {s} finished with status: {exit_status}.\nClick to focus.'
-            cmd.actions = 'focus'
             cmd.only_when = OnlyWhen(when)
             if action == 'notify':
                 notify_with_command(cmd, self.id)

kitty_tests/parser.py

@@ -15,7 +15,7 @@ from kitty.fast_data_types import (
     test_find_either_of_two_bytes,
     test_utf8_decode_to_sentinel,
 )
-from kitty.notify import NotificationCommand, Urgency, handle_notification_cmd, notification_activated, reset_registry
+from kitty.notify import NotificationCommand, QueryResponse, Urgency, handle_notification_cmd, notification_activated, reset_registry
 
 from . import BaseTest, parse_bytes
 
@@ -525,6 +525,7 @@ class TestParser(BaseTest):
         reset_registry()
         notifications = []
         activations = []
+        query_responses = []
         prev_cmd = NotificationCommand()
 
         def reset():
@@ -532,6 +533,7 @@ class TestParser(BaseTest):
             reset_registry()
             del notifications[:]
             del activations[:]
+            del query_responses[:]
             prev_cmd = NotificationCommand()
 
         def notify(title, body, identifier, urgency=Urgency.Normal):
@@ -539,9 +541,12 @@ class TestParser(BaseTest):
 
         def h(raw_data, osc_code=99, window_id=1):
             nonlocal prev_cmd
-            x = handle_notification_cmd(osc_code, raw_data, window_id, prev_cmd, notify)
-            if x is not None and osc_code == 99:
-                prev_cmd = x
+            try:
+                x = handle_notification_cmd(osc_code, raw_data, window_id, prev_cmd, notify)
+                if x is not None and osc_code == 99:
+                    prev_cmd = x
+            except QueryResponse as err:
+                query_responses.append(err.response_string)
 
         def activated(identifier, window_id, focus, report):
             activations.append((identifier, window_id, focus, report))
@@ -583,6 +588,9 @@ class TestParser(BaseTest):
         notification_activated(notifications[-1][-2], activated)
         self.ae(activations, [('0', 1, True, False)])
         reset()
+        h('?')
+        self.assertFalse(notifications)
+        self.ae(query_responses, ['99;?;a=focus,report:o=always,unfocused,invisible'])
 
     def test_dcs_codes(self):
         s = self.create_screen()