core/config: add the [notify] section

`[notify]` decides what the CLI does when a prompt is waiting on the human:
`grace_seconds` (default 5s, roughly Claude Code's permission notice) before
anything is shown, `min_interval_seconds` between two notifications,
`channels` (`auto` = desktop then terminal), `actions` for the allow/deny/review
buttons, `include_reason` for the one-line reason, and `protocol` to pin the
terminal escape protocol.

The reason stays out by default because it names commands, argv and URLs, and a
notification outlives the prompt and may be mirrored to a lock screen. Unknown
keys degrade to the defaults, the way `[mcp]` does, and `reload()` re-reads the
section.

Desktop notifications need the optional `notify` extra (`desktop-notifier`),
which is the only new dependency and stays off the default install.
This commit is contained in:
2026-10-21 19:53:19 +08:00
parent 557300ff4f
commit 10402f5be6
5 changed files with 139 additions and 0 deletions
+4
View File
@@ -22,6 +22,10 @@ requires-python = ">=3.14,<4"
readme = "README.md" readme = "README.md"
license = { text = "MIT" } license = { text = "MIT" }
[project.optional-dependencies]
# Desktop notifications while a prompt waits on the human (see `[notify]`).
notify = ["desktop-notifier>=6.2.0"]
[project.scripts] [project.scripts]
plyngent = "plyngent.cli:main" plyngent = "plyngent.cli:main"
+1
View File
@@ -17,6 +17,7 @@ from .models import McpConfig as McpConfig
from .models import McpServerConfig as McpServerConfig from .models import McpServerConfig as McpServerConfig
from .models import ModelConfig as ModelConfig from .models import ModelConfig as ModelConfig
from .models import NetworkingConfig as NetworkingConfig from .models import NetworkingConfig as NetworkingConfig
from .models import NotifyConfig as NotifyConfig
from .models import OpenAICompatibleProvider as OpenAICompatibleProvider from .models import OpenAICompatibleProvider as OpenAICompatibleProvider
from .models import OpenAIProvider as OpenAIProvider from .models import OpenAIProvider as OpenAIProvider
from .models import PluginsConfig as PluginsConfig from .models import PluginsConfig as PluginsConfig
+26
View File
@@ -278,6 +278,32 @@ class SkillsConfig(Struct, omit_defaults=True):
allow_write: bool = False allow_write: bool = False
class NotifyConfig(Struct, omit_defaults=True):
"""Desktop notifications shown while a prompt waits on the human.
A prompt is announced only after ``grace_seconds`` of waiting, so answering
at the terminal stays silent; ``min_interval_seconds`` throttles bursts.
``channels`` picks the surfaces: ``auto`` tries the desktop first and falls
back to the terminal, ``desktop`` / ``terminal`` pin one of them.
Desktop notifications come from the optional ``desktop-notifier`` extra
(``pip install plyngent[notify]``); without it the terminal channel is all
that is left. ``actions = false`` keeps the notification text-only, and
``include_reason`` lets the one-line reason through — it can name a command
or a URL, so it stays off by default (notifications outlive the prompt and
may be mirrored to a lock screen).
"""
enabled: bool = True
grace_seconds: float = 5.0
min_interval_seconds: float = 10.0
channels: list[str] = field(default_factory=lambda: ["auto"])
actions: bool = True
include_reason: bool = False
# Terminal protocol override: "" = detect, else osc777 / osc9 / bell.
protocol: str = ""
class ModelConfig(Struct, omit_defaults=True): class ModelConfig(Struct, omit_defaults=True):
"""Capability flags and optional routing overrides for a model. """Capability flags and optional routing overrides for a model.
+27
View File
@@ -12,6 +12,7 @@ from .models import (
McpConfig, McpConfig,
ModelConfig, ModelConfig,
NetworkingConfig, NetworkingConfig,
NotifyConfig,
PluginsConfig, PluginsConfig,
Provider, Provider,
SkillsConfig, SkillsConfig,
@@ -66,6 +67,22 @@ def _parse_skills(raw: dict[str, object]) -> SkillsConfig:
return SkillsConfig() return SkillsConfig()
def _parse_notify(raw: dict[str, object]) -> NotifyConfig:
"""Parse the ``[notify]`` section, falling back to defaults.
Unknown keys are rejected like ``[mcp]`` does: a typo degrades to the
defaults (notifications on, ``auto`` channel) instead of half-applying.
"""
try:
config = msgspec.convert(raw, NotifyConfig)
except msgspec.ValidationError:
return NotifyConfig()
known = set(NotifyConfig.__struct_fields__)
if set(raw) - known:
return NotifyConfig()
return config
def _parse_mcp(raw: dict[str, object]) -> McpConfig: def _parse_mcp(raw: dict[str, object]) -> McpConfig:
"""Parse the ``[mcp]`` section, falling back to defaults. """Parse the ``[mcp]`` section, falling back to defaults.
@@ -160,6 +177,7 @@ class ConfigStore:
_networking: NetworkingConfig _networking: NetworkingConfig
_mcp: McpConfig _mcp: McpConfig
_skills: SkillsConfig _skills: SkillsConfig
_notify: NotifyConfig
_providers: dict[str, Provider] _providers: dict[str, Provider]
_bad_providers: dict[str, object] _bad_providers: dict[str, object]
_recoverable_providers: dict[str, Provider] _recoverable_providers: dict[str, Provider]
@@ -174,6 +192,7 @@ class ConfigStore:
self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {}))) self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {}))) self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {})))
self._skills = _parse_skills(cast("dict[str, object]", raw.get("skills", {}))) self._skills = _parse_skills(cast("dict[str, object]", raw.get("skills", {})))
self._notify = _parse_notify(cast("dict[str, object]", raw.get("notify", {})))
self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(document) self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(document)
@property @property
@@ -238,6 +257,13 @@ class ConfigStore:
"""Typed skills section (roots, discovery sources, catalog limits).""" """Typed skills section (roots, discovery sources, catalog limits)."""
return self._skills return self._skills
# -- notify (read-only) --
@property
def notify_config(self) -> NotifyConfig:
"""Typed notifications section (grace, channels, actions)."""
return self._notify
def set_plugins_enable(self, names: Sequence[str]) -> PluginsConfig: def set_plugins_enable(self, names: Sequence[str]) -> PluginsConfig:
"""Replace the plugin enable list (in-memory; call :meth:`write` to persist).""" """Replace the plugin enable list (in-memory; call :meth:`write` to persist)."""
cleaned = [name.strip() for name in names if name.strip()] cleaned = [name.strip() for name in names if name.strip()]
@@ -442,6 +468,7 @@ class ConfigStore:
self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {}))) self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {})))
self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {}))) self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {}))) self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {})))
self._notify = _parse_notify(cast("dict[str, object]", raw.get("notify", {})))
self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(self._document) self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(self._document)
# -- internal sync helpers -- # -- internal sync helpers --
+81
View File
@@ -0,0 +1,81 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from plyngent.config import load
if TYPE_CHECKING:
from pathlib import Path
def test_notify_section_defaults(tmp_path: Path) -> None:
"""Notifications are on by default: a five-second grace, desktop first."""
path = tmp_path / "c.toml"
_ = path.write_text("", encoding="utf-8")
cfg = load(path).notify_config
assert cfg.enabled is True
assert cfg.grace_seconds == 5.0
assert cfg.min_interval_seconds == 10.0
assert cfg.channels == ["auto"]
assert cfg.actions is True
assert cfg.include_reason is False
assert cfg.protocol == ""
def test_notify_section_parse(tmp_path: Path) -> None:
path = tmp_path / "c.toml"
_ = path.write_text(
"""
[notify]
enabled = false
grace_seconds = 2.5
min_interval_seconds = 30.0
channels = ["terminal"]
actions = false
include_reason = true
protocol = "osc9"
""",
encoding="utf-8",
)
cfg = load(path).notify_config
assert cfg.enabled is False
assert cfg.grace_seconds == 2.5
assert cfg.min_interval_seconds == 30.0
assert cfg.channels == ["terminal"]
assert cfg.actions is False
assert cfg.include_reason is True
assert cfg.protocol == "osc9"
def test_notify_section_typo_falls_back(tmp_path: Path) -> None:
"""An unknown key or a bad type degrades to the defaults, like ``[mcp]``."""
path = tmp_path / "c.toml"
_ = path.write_text(
"""
[notify]
grace = 1.0
""",
encoding="utf-8",
)
assert load(path).notify_config.grace_seconds == 5.0
path2 = tmp_path / "d.toml"
_ = path2.write_text(
"""
[notify]
enabled = "yes"
""",
encoding="utf-8",
)
assert load(path2).notify_config.enabled is True
def test_notify_section_survives_reload(tmp_path: Path) -> None:
path = tmp_path / "c.toml"
_ = path.write_text("[notify]\nenabled = false\n", encoding="utf-8")
store = load(path)
assert store.notify_config.enabled is False
_ = path.write_text("[notify]\nenabled = true\ngrace_seconds = 1.0\n", encoding="utf-8")
store.reload()
assert store.notify_config.enabled is True
assert store.notify_config.grace_seconds == 1.0