feat(notifications): quiet hours notification suppression

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 "<skill-dir>/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

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 |

cargento/skills/cargento/cargento_runtime/aggregate.py

@@ -827,17 +827,18 @@ 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,
             tripwires.sources_from_sessions(out_sessions),
             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"),

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,

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:

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

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)

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`

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)

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,
         )

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()