From 71b0b57b147fc560e5659b5cf4eb18b010bba309 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Wed, 16 Sep 2026 09:40:57 +0800 Subject: [PATCH] feat(notifications): add quiet hours notification suppression Signed-off-by: Jared Scott --- HOW_TO_USE.md | 14 ++ cargento/skills/cargento/SKILL.md | 2 + .../cargento/cargento_runtime/aggregate.py | 6 +- .../skills/cargento/cargento_runtime/cli.py | 17 ++ .../cargento/cargento_runtime/config.py | 8 + .../cargento/cargento_runtime/lifecycle.py | 5 + .../cargento_runtime/notifications.py | 86 ++++++- .../cargento/cargento_runtime/unasked.py | 2 + .../cargento/tests/test_config_diagnostics.py | 11 + .../cargento/tests/test_documentation.py | 1 + .../skills/cargento/tests/test_lifecycle.py | 21 ++ .../cargento/tests/test_notifications.py | 218 +++++++++++++++++- 12 files changed, 381 insertions(+), 10 deletions(-) diff --git a/HOW_TO_USE.md b/HOW_TO_USE.md index 43403af1..e8d59a32 100644 --- a/HOW_TO_USE.md +++ b/HOW_TO_USE.md @@ -515,6 +515,20 @@ nudges are disabled by default and can be suppressed for any run with `--no-reac [SECURITY.md](SECURITY.md#off-machine-nudges-reaching-the-operator-away-from-the-desk) for the full security contract. +## Quiet hours + +When configured, Cargento suppresses non-urgent notifications (native popups, unasked checks, +tripwire alerts, and reach nudges) during a specified local time window: + +```bash +python3 "/server.py" --quiet-hours "22:00-08:00" +``` + +The window can also be set via the `CARGENTO_QUIET_HOURS` environment variable or saved in +`~/.cargento/quiet_hours`. The window format is `HH:MM-HH:MM` in 24-hour local time and can cross +midnight (e.g. `22:00-08:00`). Direct operator questions asking for input are not suppressed. +Quiet hours can be disabled for any run with `--no-quiet-hours`. + ## Stop a dashboard, and unstick a port ```bash diff --git a/cargento/skills/cargento/SKILL.md b/cargento/skills/cargento/SKILL.md index 575a7d1b..db8c99a3 100644 --- a/cargento/skills/cargento/SKILL.md +++ b/cargento/skills/cargento/SKILL.md @@ -535,6 +535,8 @@ Paths 2 and 3 are complementary and can both be installed. Keep `Notification` o | `--no-history` | For this run, keep no local history of what the server observed: nothing is written and an existing store is not read back, so the board opens with no memory of earlier sessions. | | `--no-reach` | For this run, disable outbound reach nudges: no webhook URL is resolved and no off-machine nudge is posted. | | `--reach-url URL` | Webhook URL for off-machine reach nudges when sessions need input or finish unread while away from the desk. Overrides `CARGENTO_REACH_URL` and `~/.cargento/reach_url`. | +| `--no-quiet-hours` | For this run, disable quiet hours notification suppression: notifications fire regardless of local time or configured window. | +| `--quiet-hours WINDOW` | Local time window (`HH:MM-HH:MM`) during which non-urgent notifications (popups, unasked checks, tripwires, reach nudges) are suppressed, except direct questions. Overrides `CARGENTO_QUIET_HOURS` and `~/.cargento/quiet_hours`. | | `--history-days N` | How long the local history keeps an observation, in days (default 14). Eviction is age first, so narrowing this drops what falls outside the window and widening it again brings nothing back. Zero or negative is refused. | | `--history-max-bytes N` | The size cap on the local history store, in bytes (default 1048576). It is the read cap too: a file larger than it is discarded unread rather than parsed. Zero or negative is refused. | | `http://127.0.0.1:4553/?all=1` | Show all sessions ever, including idle ones | diff --git a/cargento/skills/cargento/cargento_runtime/aggregate.py b/cargento/skills/cargento/cargento_runtime/aggregate.py index bbc90f18..6c3e3034 100644 --- a/cargento/skills/cargento/cargento_runtime/aggregate.py +++ b/cargento/skills/cargento/cargento_runtime/aggregate.py @@ -827,6 +827,7 @@ def collect(self, *, show_all: bool, notify: bool = True) -> Collection: # happened — subtracting first would punch gaps in the history of any # session the reader ever cleared. self._notify_waits(out_sessions, generations, notify=notify) + in_quiet = notifications.is_quiet_hours(config, now=now) stage_conditions = tripwires.collect( config, state, @@ -834,10 +835,10 @@ def collect(self, *, show_all: bool, notify: bool = True) -> Collection: now, self.native_notifier(config.platform_name), self.popup_notifier, - notify=notify, + notify=notify and not in_quiet, ) out_sessions, cleared = _subtract_dismissed(out_sessions, cleared_marks) - if notify: + if notify and not in_quiet: reach.maybe_reach_nudge(config, state, out_sessions, now=now) _attach_cached_goals(config, out_sessions) sessions.assign_display_ids(config, out_sessions) @@ -864,6 +865,7 @@ def collect(self, *, show_all: bool, notify: bool = True) -> Collection: # Which layer owns needs-input popups. Empty means the page should # raise its own; a backend name means the server already did. "native_notify": self.native_notifier(config.platform_name), + "in_quiet_hours": in_quiet, "harnesses": harnesses, "summary": { "needs_input": sum(1 for x in active_sessions if x["state"] == "needs_input"), diff --git a/cargento/skills/cargento/cargento_runtime/cli.py b/cargento/skills/cargento/cargento_runtime/cli.py index 1c4308bf..fd669b1f 100644 --- a/cargento/skills/cargento/cargento_runtime/cli.py +++ b/cargento/skills/cargento/cargento_runtime/cli.py @@ -312,6 +312,21 @@ def build_parser() -> argparse.ArgumentParser: "--reach-url", help="webhook endpoint for off-machine nudges when a session needs human attention", ) + parser.add_argument( + "--quiet-hours", + help=( + "time window (HH:MM-HH:MM in 24-hour local time) during which non-ask " + "notifications and nudges are suppressed (e.g. 22:00-08:00 or 13:00-14:00)" + ), + ) + parser.add_argument( + "--no-quiet-hours", + action="store_true", + help=( + "do not suppress notifications for quiet hours for this run, " + "regardless of the stored setting or environment variable" + ), + ) parser.add_argument( "--no-irreversible", action="store_true", @@ -422,6 +437,8 @@ def build_runtime( ask_enabled=not args.no_ask, reach_enabled=not args.no_reach, reach_url=args.reach_url, + quiet_hours_enabled=not getattr(args, "no_quiet_hours", False), + quiet_hours=getattr(args, "quiet_hours", None), history_enabled=not args.no_history, history_retention_sec=args.history_days * runtime_config.SECONDS_PER_DAY, history_max_bytes=args.history_max_bytes, diff --git a/cargento/skills/cargento/cargento_runtime/config.py b/cargento/skills/cargento/cargento_runtime/config.py index 5e0f95a6..5853a163 100644 --- a/cargento/skills/cargento/cargento_runtime/config.py +++ b/cargento/skills/cargento/cargento_runtime/config.py @@ -109,6 +109,10 @@ class RuntimeConfig: reach_enabled: bool reach_url: str | None reach_cooldown_sec: float + # Quiet hours notification suppression (DRC-4032). + # `--no-quiet-hours` is the off switch for this run. + quiet_hours_enabled: bool + quiet_hours: str | None # The trailing window every published token rate is averaged over. What a # row carries is therefore a MEAN and not an instantaneous reading, and at # ten minutes it lags a burst by minutes. `sessions.rate_from` divides by it, @@ -596,6 +600,8 @@ def build_runtime_config( reach_enabled: bool = True, reach_url: str | None = None, reach_cooldown_sec: float = 60.0, + quiet_hours_enabled: bool = True, + quiet_hours: str | None = None, history_enabled: bool = True, history_retention_sec: float = HISTORY_RETENTION_DEFAULT_DAYS * SECONDS_PER_DAY, history_max_bytes: int = HISTORY_MAX_BYTES_DEFAULT, @@ -649,6 +655,8 @@ def build_runtime_config( reach_enabled=reach_enabled, reach_url=reach_url, reach_cooldown_sec=reach_cooldown_sec, + quiet_hours_enabled=quiet_hours_enabled, + quiet_hours=quiet_hours, history_enabled=history_enabled, # Ten minutes stays. The burn ordering (DRC-4011) wants the fastest # session "right now", and this window is the reason it cannot have it: diff --git a/cargento/skills/cargento/cargento_runtime/lifecycle.py b/cargento/skills/cargento/cargento_runtime/lifecycle.py index 9bbe1a48..1c783f02 100644 --- a/cargento/skills/cargento/cargento_runtime/lifecycle.py +++ b/cargento/skills/cargento/cargento_runtime/lifecycle.py @@ -635,6 +635,8 @@ def _opt_out_argv(args: argparse.Namespace) -> list[str]: if getattr(args, "no_reach", False): # SECURITY.md's off switch for off-machine reach nudges. argv.append("--no-reach") + if getattr(args, "no_quiet_hours", False): + argv.append("--no-quiet-hours") return argv @@ -666,6 +668,9 @@ def spawn_argv(config: RuntimeConfig, args: argparse.Namespace) -> list[str]: reach_url = getattr(args, "reach_url", None) if reach_url: argv.extend(["--reach-url", reach_url]) + quiet_hours = getattr(args, "quiet_hours", None) + if quiet_hours: + argv.extend(["--quiet-hours", quiet_hours]) argv.extend(_history_bound_argv(args)) # Forward the bind host only when the operator chose a non-default address, # so a Windows --daemon re-spawn keeps a --host 0.0.0.0 bind instead of diff --git a/cargento/skills/cargento/cargento_runtime/notifications.py b/cargento/skills/cargento/cargento_runtime/notifications.py index 190a5835..eedf74fc 100644 --- a/cargento/skills/cargento/cargento_runtime/notifications.py +++ b/cargento/skills/cargento/cargento_runtime/notifications.py @@ -6,7 +6,7 @@ import subprocess import time from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Final from cargento_runtime import claude_data, deliveries, dismissals, records from cargento_runtime import io as runtime_io @@ -75,6 +75,68 @@ } +QUIET_HOURS_FILENAME: Final[str] = "quiet_hours" +QUIET_HOURS_ENV: Final[str] = "CARGENTO_QUIET_HOURS" +QUIET_HOURS_MAX_BYTES: Final[int] = 128 + + +def parse_quiet_hours(spec: Any) -> tuple[int, int, int, int] | None: + """Parse a quiet hours window string (e.g. '22:00-08:00') into (start_h, start_m, end_h, end_m). + + Returns None if spec is invalid, empty, or has identical start and end times. + """ + if not isinstance(spec, str) or "-" not in spec: + return None + try: + start_str, end_str = spec.strip().split("-", 1) + start_h, start_m = (int(x) for x in start_str.split(":", 1)) + end_h, end_m = (int(x) for x in end_str.split(":", 1)) + valid_time = ( + 0 <= start_h <= 23 and 0 <= start_m <= 59 and 0 <= end_h <= 23 and 0 <= end_m <= 59 + ) + if not valid_time or (start_h, start_m) == (end_h, end_m): + return None + except (ValueError, AttributeError): + return None + else: + return (start_h, start_m, end_h, end_m) + + +def resolve_quiet_hours(config: RuntimeConfig) -> tuple[int, int, int, int] | None: + """Resolve the effective quiet hours window, or None if disabled or unconfigured.""" + if not config.quiet_hours_enabled: + return None + candidate = config.quiet_hours + if not candidate: + candidate = os.environ.get(QUIET_HOURS_ENV) + if not candidate: + quiet_file = os.path.join(config.state_home, QUIET_HOURS_FILENAME) + if os.path.isfile(quiet_file): + try: + with open(quiet_file, encoding="utf-8") as f: + candidate = f.read(QUIET_HOURS_MAX_BYTES) + except OSError: + candidate = None + return parse_quiet_hours(candidate) + + +def is_quiet_hours(config: RuntimeConfig, *, now: float | None = None) -> bool: + """Return True if the current local time falls within the configured quiet hours window.""" + window = resolve_quiet_hours(config) + if window is None: + return False + start_h, start_m, end_h, end_m = window + current_ts = now if now is not None else time.time() + local_tm = time.localtime(current_ts) + cur_m = local_tm.tm_hour * 60 + local_tm.tm_min + start_minutes = start_h * 60 + start_m + end_minutes = end_h * 60 + end_m + + if start_minutes < end_minutes: + return start_minutes <= cur_m < end_minutes + return cur_m >= start_minutes or cur_m < end_minutes + + def normalized_notification_type(value: Any) -> str: """Return a normalized structured notification type, if present.""" return value.strip().lower() if isinstance(value, str) and value.strip() else "" @@ -421,13 +483,15 @@ def maybe_popup( the comment on that branch has the reason. """ prefix, harness_label = subject.prefix, subject.label + now = time.time() + if is_quiet_hours(config, now=now): + return if dismissals.suppresses(config, state, subject.harness, prefix, subject.activity): # Returned before the last-session-state write below, deliberately. This # call is not evidence about the session, so recording a transition from # it would let a dismissal rewrite the history the popup decision after a # restore is made against. return - now = time.time() with state.hook_lock: if ( expect_generation is not None @@ -675,6 +739,14 @@ def clear_session(state: RuntimeState, config: RuntimeConfig, prefix: str) -> No ) +def _payload_response(cleared: bool, in_quiet: bool) -> dict[str, Any]: + if cleared: + return {"ok": True, "suppressed": "cleared"} + if in_quiet: + return {"ok": True, "suppressed": "quiet_hours"} + return {"ok": True} + + def handle_payload( config: RuntimeConfig, state: RuntimeState, @@ -763,10 +835,12 @@ def handle_payload( global_ready = now - state.last_popup.get("_global", 0) >= config.global_popup_cooldown_sec # Claude re-emits the same idle/permission notification for as long as # the session stays blocked; repeating the popup adds no information. - # One popup per distinct message per session within the repeat window. prev_msg, prev_ts = state.last_popup_message.get(popup_key, ("", 0.0)) repeat = message == prev_msg and now - prev_ts < config.popup_repeat_suppress_sec - fire = popup and session_ready and global_ready and not repeat and not cleared + in_quiet = is_quiet_hours(config, now=now) + fire = ( + popup and session_ready and global_ready and not repeat and not cleared and not in_quiet + ) if fire: spent_before = ( state.last_popup.get(popup_key), @@ -788,6 +862,4 @@ def handle_payload( # Claude's own hook forwarder and nothing else posts there, so the # harness is a property of the route rather than a field to trust. record_outcome(config, "claude", prefix, "hook", outcome, now) - if cleared: - return {"ok": True, "suppressed": "cleared"} - return {"ok": True} + return _payload_response(cleared, in_quiet) diff --git a/cargento/skills/cargento/cargento_runtime/unasked.py b/cargento/skills/cargento/cargento_runtime/unasked.py index 87baeeff..bb670d87 100644 --- a/cargento/skills/cargento/cargento_runtime/unasked.py +++ b/cargento/skills/cargento/cargento_runtime/unasked.py @@ -395,6 +395,8 @@ def _raise( A reading that departed on both Goal and Expected Output is one thing that happened, and two banners about it would read as two events. """ + if notifications.is_quiet_hours(self.config, now=now): + return label = self.harness_label(str(row.get("harness") or "")) first = raised[0] # The constraint first and the model's sentence after. `notify_mac` diff --git a/cargento/skills/cargento/tests/test_config_diagnostics.py b/cargento/skills/cargento/tests/test_config_diagnostics.py index 2e3508a5..ff7b61f5 100644 --- a/cargento/skills/cargento/tests/test_config_diagnostics.py +++ b/cargento/skills/cargento/tests/test_config_diagnostics.py @@ -149,6 +149,17 @@ def test_build_runtime_threads_host_into_config(self) -> None: config, _state = cli.build_runtime(args, started=1.0, launcher_path=SERVER_PATH) self.assertEqual("0.0.0.0", config.host) + def test_build_runtime_threads_quiet_hours_into_config(self) -> None: + args = cli.build_parser().parse_args(["--quiet-hours", "22:00-08:00"]) + config, _state = cli.build_runtime(args, started=1.0, launcher_path=SERVER_PATH) + self.assertEqual("22:00-08:00", config.quiet_hours) + self.assertTrue(config.quiet_hours_enabled) + + def test_build_runtime_threads_no_quiet_hours_into_config(self) -> None: + args = cli.build_parser().parse_args(["--no-quiet-hours"]) + config, _state = cli.build_runtime(args, started=1.0, launcher_path=SERVER_PATH) + self.assertFalse(config.quiet_hours_enabled) + def test_host_flag_defaults_to_loopback(self) -> None: args = cli.build_parser().parse_args([]) self.assertEqual("127.0.0.1", args.host) diff --git a/cargento/skills/cargento/tests/test_documentation.py b/cargento/skills/cargento/tests/test_documentation.py index c2946ae7..c582c61a 100644 --- a/cargento/skills/cargento/tests/test_documentation.py +++ b/cargento/skills/cargento/tests/test_documentation.py @@ -1822,6 +1822,7 @@ def test_the_two_unshipped_switches_are_the_only_no_flags_missing(self) -> None: "--no-irreversible", "--no-tripwires", "--no-reach", + "--no-quiet-hours", }, shipped, ) diff --git a/cargento/skills/cargento/tests/test_lifecycle.py b/cargento/skills/cargento/tests/test_lifecycle.py index 48901971..c6efb357 100644 --- a/cargento/skills/cargento/tests/test_lifecycle.py +++ b/cargento/skills/cargento/tests/test_lifecycle.py @@ -1739,6 +1739,27 @@ def test_reach_url_is_forwarded(self) -> None: self.assertIn("--reach-url", argv) self.assertIn("https://example.com/webhook", argv) + def test_no_quiet_hours_is_forwarded_when_requested(self) -> None: + config = cfg() + argv = lifecycle.spawn_argv(config, self._args(no_quiet_hours=True)) + self.assertIn("--no-quiet-hours", argv) + + def test_no_quiet_hours_is_absent_when_not_requested(self) -> None: + config = cfg() + argv = lifecycle.spawn_argv(config, self._args(no_quiet_hours=False)) + self.assertNotIn("--no-quiet-hours", argv) + + def test_quiet_hours_is_forwarded(self) -> None: + config = cfg() + argv = lifecycle.spawn_argv(config, self._args(quiet_hours="22:00-08:00")) + self.assertIn("--quiet-hours", argv) + self.assertIn("22:00-08:00", argv) + + def test_quiet_hours_is_absent_when_not_requested(self) -> None: + config = cfg() + argv = lifecycle.spawn_argv(config, self._args()) + self.assertNotIn("--quiet-hours", argv) + def test_daemon_is_never_forwarded(self) -> None: """Forwarding --daemon would respawn forever.""" config = cfg() diff --git a/cargento/skills/cargento/tests/test_notifications.py b/cargento/skills/cargento/tests/test_notifications.py index 662098fd..bd732be3 100644 --- a/cargento/skills/cargento/tests/test_notifications.py +++ b/cargento/skills/cargento/tests/test_notifications.py @@ -1,5 +1,6 @@ from __future__ import annotations +import dataclasses import http.client import http.server import json @@ -15,7 +16,15 @@ from typing import TYPE_CHECKING, Any, ClassVar from unittest import mock -from cargento_runtime import aggregate, claude_data, dismissals, notifications, records +from cargento_runtime import ( + aggregate, + claude_data, + departures, + dismissals, + notifications, + records, + unasked, +) from cargento_runtime import events as runtime_events from cargento_runtime import sessions as runtime_sessions @@ -2042,3 +2051,210 @@ def test_every_registry_row_is_titled_by_its_own_label(self) -> None: [(f"{spec.label} is waiting on you", "[proj] permission requested")], self.popups, ) + + +class QuietHoursTest(RuntimeTestCase): + def test_parse_quiet_hours(self) -> None: + self.assertEqual((22, 0, 8, 0), notifications.parse_quiet_hours("22:00-08:00")) + self.assertEqual((13, 30, 15, 45), notifications.parse_quiet_hours("13:30-15:45")) + self.assertEqual((9, 5, 17, 30), notifications.parse_quiet_hours("9:05-17:30")) + self.assertIsNone(notifications.parse_quiet_hours("invalid")) + self.assertIsNone(notifications.parse_quiet_hours("25:00-08:00")) + self.assertIsNone(notifications.parse_quiet_hours("12:60-14:00")) + self.assertIsNone(notifications.parse_quiet_hours("12:00-12:00")) + self.assertIsNone(notifications.parse_quiet_hours("")) + self.assertIsNone(notifications.parse_quiet_hours(None)) + + def test_is_quiet_hours_daytime_window(self) -> None: + base_config, _ = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=True, quiet_hours="13:00-15:00" + ) + + # Construct timestamps with known local hours and minutes + # 12:59 + t_1259 = time.mktime(datetime(2026, 9, 16, 12, 59, 0).timetuple()) + # 13:00 + t_1300 = time.mktime(datetime(2026, 9, 16, 13, 0, 0).timetuple()) + # 14:30 + t_1430 = time.mktime(datetime(2026, 9, 16, 14, 30, 0).timetuple()) + # 15:00 + t_1500 = time.mktime(datetime(2026, 9, 16, 15, 0, 0).timetuple()) + + self.assertFalse(notifications.is_quiet_hours(config, now=t_1259)) + self.assertTrue(notifications.is_quiet_hours(config, now=t_1300)) + self.assertTrue(notifications.is_quiet_hours(config, now=t_1430)) + self.assertFalse(notifications.is_quiet_hours(config, now=t_1500)) + + def test_is_quiet_hours_overnight_window(self) -> None: + base_config, _ = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=True, quiet_hours="22:00-08:00" + ) + + t_2159 = time.mktime(datetime(2026, 9, 16, 21, 59, 0).timetuple()) + t_2200 = time.mktime(datetime(2026, 9, 16, 22, 0, 0).timetuple()) + t_2330 = time.mktime(datetime(2026, 9, 16, 23, 30, 0).timetuple()) + t_0200 = time.mktime(datetime(2026, 9, 16, 2, 0, 0).timetuple()) + t_0759 = time.mktime(datetime(2026, 9, 16, 7, 59, 0).timetuple()) + t_0800 = time.mktime(datetime(2026, 9, 16, 8, 0, 0).timetuple()) + t_1200 = time.mktime(datetime(2026, 9, 16, 12, 0, 0).timetuple()) + + self.assertFalse(notifications.is_quiet_hours(config, now=t_2159)) + self.assertTrue(notifications.is_quiet_hours(config, now=t_2200)) + self.assertTrue(notifications.is_quiet_hours(config, now=t_2330)) + self.assertTrue(notifications.is_quiet_hours(config, now=t_0200)) + self.assertTrue(notifications.is_quiet_hours(config, now=t_0759)) + self.assertFalse(notifications.is_quiet_hours(config, now=t_0800)) + self.assertFalse(notifications.is_quiet_hours(config, now=t_1200)) + + def test_quiet_hours_disabled_flag_overrides(self) -> None: + base_config, _ = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=False, quiet_hours="22:00-08:00" + ) + + t_2330 = time.mktime(datetime(2026, 9, 16, 23, 30, 0).timetuple()) + self.assertFalse(notifications.is_quiet_hours(config, now=t_2330)) + + def test_maybe_popup_suppressed_during_quiet_hours(self) -> None: + base_config, state = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=True, quiet_hours="22:00-08:00" + ) + + t_quiet = time.mktime(datetime(2026, 9, 16, 23, 0, 0).timetuple()) + fired: list[tuple[str, str]] = [] + + with mock.patch.object(time, "time", return_value=t_quiet): + notifications.maybe_popup( + config, + state, + notifications.PopupSubject( + harness="claude", label="Claude", prefix="sess1", activity=0.0 + ), + "needs_input", + "Needs input", + popup_notifier=lambda t, m: fired.append((t, m)), + ) + self.assertEqual([], fired) + + # Outside quiet hours, it fires + t_active = time.mktime(datetime(2026, 9, 16, 10, 0, 0).timetuple()) + with mock.patch.object(time, "time", return_value=t_active): + notifications.maybe_popup( + config, + state, + notifications.PopupSubject( + harness="claude", label="Claude", prefix="sess2", activity=0.0 + ), + "needs_input", + "Needs input", + popup_notifier=lambda t, m: fired.append((t, m)), + ) + self.assertEqual([("Claude is waiting on you", "Needs input")], fired) + + def test_handle_payload_suppressed_during_quiet_hours(self) -> None: + base_config, state = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=True, quiet_hours="22:00-08:00" + ) + + t_quiet = time.mktime(datetime(2026, 9, 16, 23, 0, 0).timetuple()) + fired: list[tuple[str, str]] = [] + + payload = { + "session_id": "sess12345678", + "message": "Claude needs input", + "notification_type": "permission_prompt", + } + res = notifications.handle_payload( + config, + state, + payload, + now=t_quiet, + popup_notifier=lambda t, m: fired.append((t, m)), + ) + self.assertTrue(res.get("ok")) + self.assertEqual([], fired) + # Hook is still saved for the dashboard UI + self.assertIn("sess1234", state.hook_notifications) + + def test_maybe_ask_popup_not_suppressed_during_quiet_hours(self) -> None: + base_config, state = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=True, quiet_hours="22:00-08:00" + ) + + t_quiet = time.mktime(datetime(2026, 9, 16, 23, 0, 0).timetuple()) + fired: list[tuple[str, str]] = [] + + subject = notifications.AskSubject( + harness="claude", + sid="session-1", + label="Claude", + question="May I run command?", + project="test-project", + ) + notifications.maybe_ask_popup( + config, + state, + subject, + now=t_quiet, + popup_notifier=lambda t, m: fired.append((t, m)), + ) + # Direct questions are NOT suppressed + self.assertEqual( + [("Claude is asking you", "May I run command? \u00b7 test-project")], fired + ) + + def test_unasked_departure_suppressed_during_quiet_hours(self) -> None: + base_config, _ = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=True, quiet_hours="22:00-08:00" + ) + t_quiet = time.mktime(datetime(2026, 9, 16, 23, 0, 0).timetuple()) + fired: list[tuple[str, str]] = [] + lane = unasked.Lane( + config, + popup_notifier=lambda t, m: fired.append((t, m)), + harness_label=lambda _h: "Claude", + ) + row = {"harness": "claude", "sid": "s1"} + raised: list[departures.Check] = [ + { + "harness": "claude", + "sid": "s1", + "at": t_quiet, + "constraint": "Goal", + "clause": "Stay on topic", + "reading": "drifted", + "evidence": "irrelevant code", + "revision": 1, + "cutoff": t_quiet, + "cutoff_text": "1 message", + "withdrawn": False, + } + ] + lane._raise(row, raised, now=t_quiet) + self.assertEqual([], fired) + + def test_aggregate_collect_suppresses_reach_during_quiet_hours(self) -> None: + base_config, state = runtime() + config = dataclasses.replace( + base_config, quiet_hours_enabled=True, quiet_hours="22:00-08:00" + ) + t_quiet = time.mktime(datetime(2026, 9, 16, 23, 0, 0).timetuple()) + app = aggregate.Application( + config, + state, + (), + native_notifier=lambda _p: "", + popup_notifier=lambda _t, _m: None, + diagnostic_sink=lambda _line: None, + clock=lambda: t_quiet, + ) + with mock.patch("cargento_runtime.reach.maybe_reach_nudge") as reach_mock: + collection = app.collect(show_all=False) + reach_mock.assert_not_called() + self.assertTrue(collection["in_quiet_hours"])