feat(host/hooks): operator hooks — exec + webhooks on lifecycle events (M2a)

hooks.json (RFC §6): commands and webhooks fired on host lifecycle events,
managed over GET|PUT /api/v1/hooks (validated, applied immediately) and
dispatched fire-and-forget by a bus-subscriber runner — hooks observe,
never veto, and no operator code sits in any streaming path.

- exec: detached sh -c with the event JSON on stdin + flat PF_EVENT_* env
  (the PF_STREAM_* vocabulary's sibling), per-hook timeout (default 30 s)
  with process-group kill, off-thread reap, per-hook debounce, bounded
  concurrency (8 in flight, excess dropped loudly). Windows runs hooks in
  the interactive user session (temp-file JSON argument; console-mode dev
  hosts get env + stdin like Unix).
- webhook: POST the event JSON, TLS-verified, redirects never followed, no
  punktfunk credentials outbound; optional per-hook secret file yields
  X-Punktfunk-Signature: sha256=<hex HMAC> (fails closed if unreadable).
- filters: exact-match client/fingerprint/plane/app + the same kind
  patterns as the SSE ?kinds= filter (shared crate::events::kind_matches).
- hardening (RFC §9.1): hooks.json via the private-dir/secret-file
  helpers; a hook script path must be operator/root-owned and not
  group/world-writable or it is refused loudly (the sshd rule).
- env mirrors PUNKTFUNK_ON_CONNECT_CMD / PUNKTFUNK_ON_DISCONNECT_CMD for
  the zero-config cases, beside PUNKTFUNK_RECOVER_SESSION_CMD.

Live-verified on Linux: PUT config via API → library.changed fired a real
script (env + stdin observed) and an HMAC webhook (receiver-verified
signature); a chmod-777 script was refused. 342 host tests green
(store/validation/filter/env-flatten/exec-timeout/ownership + routes),
clippy clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Cargo.lock

@@ -3106,6 +3106,7 @@ dependencies = [
  "ffmpeg-next",
  "futures-util",
  "hex",
+ "hmac",
  "http-body-util",
  "hyper",
  "hyper-util",

api/openapi.json

@@ -784,6 +784,98 @@
         ]
       }
     },
+    "/api/v1/hooks": {
+      "get": {
+        "tags": [
+          "hooks"
+        ],
+        "summary": "Get the hook configuration",
+        "description": "The operator's `hooks.json`: commands and webhooks fired on host lifecycle events. Empty\nwhen unconfigured.",
+        "operationId": "getHooks",
+        "responses": {
+          "200": {
+            "description": "The stored hook configuration",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HooksConfig"
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "Missing or invalid bearer token",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ApiError"
+                }
+              }
+            }
+          }
+        }
+      },
+      "put": {
+        "tags": [
+          "hooks"
+        ],
+        "summary": "Replace the hook configuration",
+        "description": "Validates and persists a full `hooks.json` document (this is a whole-document PUT, not a\npatch). Applies from the next event — no restart. Hook commands run as the host user\n(interactive user session on Windows): treat this configuration as operator-privileged.",
+        "operationId": "setHooks",
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/HooksConfig"
+              }
+            }
+          },
+          "required": true
+        },
+        "responses": {
+          "200": {
+            "description": "Configuration stored; the new state",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/HooksConfig"
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Structurally invalid configuration",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ApiError"
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "Missing or invalid bearer token",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ApiError"
+                }
+              }
+            }
+          },
+          "500": {
+            "description": "Configuration could not be persisted",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ApiError"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/api/v1/host": {
       "get": {
         "tags": [
@@ -3205,6 +3297,113 @@
           }
         }
       },
+      "HookEntry": {
+        "type": "object",
+        "description": "One hook: fire `run` and/or `webhook` when an event matching `on` (+ `filter`) occurs.",
+        "required": [
+          "on"
+        ],
+        "properties": {
+          "debounce_ms": {
+            "type": "integer",
+            "format": "int64",
+            "description": "Minimum interval between firings of this hook, in milliseconds. 0 = fire every time.",
+            "minimum": 0
+          },
+          "filter": {
+            "oneOf": [
+              {
+                "type": "null"
+              },
+              {
+                "$ref": "#/components/schemas/HookFilter",
+                "description": "Exact-match constraints on the event's fields; every present field must match."
+              }
+            ]
+          },
+          "hmac_secret_file": {
+            "type": [
+              "string",
+              "null"
+            ],
+            "description": "File holding the webhook HMAC secret (`X-Punktfunk-Signature: sha256=<hex>`). The file\nshould be operator-owned and private; a world-readable secret is warned about."
+          },
+          "on": {
+            "type": "string",
+            "description": "Which events fire this hook: an exact kind (`stream.started`) or a `domain.*` prefix\n(`pairing.*`) — the same vocabulary as the SSE `?kinds=` filter."
+          },
+          "run": {
+            "type": [
+              "string",
+              "null"
+            ],
+            "description": "Shell command to execute (detached, event JSON on stdin + `PF_EVENT_*` env)."
+          },
+          "timeout_s": {
+            "type": "integer",
+            "format": "int32",
+            "description": "Exec timeout in seconds (1–600, default 30); the process group is killed on expiry.",
+            "minimum": 0
+          },
+          "webhook": {
+            "type": [
+              "string",
+              "null"
+            ],
+            "description": "URL to POST the event JSON to."
+          }
+        }
+      },
+      "HookFilter": {
+        "type": "object",
+        "description": "Exact-match filters against an event's identity fields (RFC open-question 3: exact match\nonly — anything richer is what the SDK is for). Absent fields don't constrain; a filter\nfield set on an event kind that doesn't carry it (e.g. `client` on `host.started`) never\nmatches.",
+        "properties": {
+          "app": {
+            "type": [
+              "string",
+              "null"
+            ],
+            "description": "Launched app id/title (`stream.*` events)."
+          },
+          "client": {
+            "type": [
+              "string",
+              "null"
+            ],
+            "description": "Client/device name (for `session.*`: the short client label the Dashboard shows)."
+          },
+          "fingerprint": {
+            "type": [
+              "string",
+              "null"
+            ],
+            "description": "Certificate fingerprint (hex, case-insensitive)."
+          },
+          "plane": {
+            "oneOf": [
+              {
+                "type": "null"
+              },
+              {
+                "$ref": "#/components/schemas/Plane",
+                "description": "Protocol plane (`native` / `gamestream`)."
+              }
+            ]
+          }
+        }
+      },
+      "HooksConfig": {
+        "type": "object",
+        "description": "The operator's hook configuration — the `hooks.json` document and the `/api/v1/hooks` body.",
+        "properties": {
+          "hooks": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/HookEntry"
+            }
+          }
+        }
+      },
       "HostEvent": {
         "allOf": [
           {
@@ -4294,6 +4493,10 @@
     {
       "name": "events",
       "description": "Host lifecycle events: an SSE stream (client/session/stream lifecycle, pairing, displays, library, host) with Last-Event-ID resume and server-side kind filters"
+    },
+    {
+      "name": "hooks",
+      "description": "Operator hooks: commands and webhooks fired on lifecycle events (fire-and-forget — hooks observe, never veto)"
     }
   ]
 }

crates/punktfunk-host/Cargo.toml

@@ -59,6 +59,9 @@ tower = { version = "0.5", features = ["util"] }
 # Stream combinators for the mgmt API's SSE event feed (`GET /api/v1/events`) — already in the
 # tree transitively (axum/hyper) and as a Linux target dep; control plane only.
 futures-util = "0.3"
+# Webhook signing (X-Punktfunk-Signature: sha256=<hex HMAC>) for operator hooks; pairs with
+# the existing sha2. Already in the lockfile transitively.
+hmac = "0.12"
 rusty_enet = "0.4"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

crates/punktfunk-host/src/config.rs

@@ -87,6 +87,14 @@ pub struct HostConfig {
     /// `systemctl restart display-manager` under a polkit rule — with auto-login enabled the restart brings
     /// the desktop back and the client's retry lands in it. Unset/empty = disabled (the default).
     pub recover_session_cmd: Option<String>,
+    /// `PUNKTFUNK_ON_CONNECT_CMD` — zero-config mirror of a `client.connected` hook
+    /// (`crate::hooks`): fired detached with the event JSON on stdin + `PF_EVENT_*` env when a
+    /// client connects, on either plane. The full hook surface (filters, webhooks, debounce)
+    /// lives in `hooks.json`. Unset/empty = disabled (the default).
+    pub on_connect_cmd: Option<String>,
+    /// `PUNKTFUNK_ON_DISCONNECT_CMD` — the `client.disconnected` sibling of
+    /// [`Self::on_connect_cmd`].
+    pub on_disconnect_cmd: Option<String>,
 }
 
 impl HostConfig {
@@ -146,6 +154,8 @@ impl HostConfig {
             }),
             recover_session_cmd: val("PUNKTFUNK_RECOVER_SESSION_CMD")
                 .filter(|s| !s.trim().is_empty()),
+            on_connect_cmd: val("PUNKTFUNK_ON_CONNECT_CMD").filter(|s| !s.trim().is_empty()),
+            on_disconnect_cmd: val("PUNKTFUNK_ON_DISCONNECT_CMD").filter(|s| !s.trim().is_empty()),
         }
     }
 }

crates/punktfunk-host/src/events.rs

@@ -196,6 +196,77 @@ impl EventKind {
     }
 }
 
+impl EventKind {
+    /// The client/device name this event carries, if any — the `filter.client` axis of hooks
+    /// and scripts. (For `session.*` this is the short client *label* the Dashboard shows —
+    /// cert-fingerprint prefix or peer IP — since that is what the event carries.)
+    pub fn client_name(&self) -> Option<&str> {
+        match self {
+            EventKind::ClientConnected { client }
+            | EventKind::ClientDisconnected { client, .. } => Some(&client.name),
+            EventKind::SessionStarted { session } | EventKind::SessionEnded { session } => {
+                Some(&session.client)
+            }
+            EventKind::StreamStarted { stream } | EventKind::StreamStopped { stream } => {
+                Some(&stream.client)
+            }
+            EventKind::PairingPending { device }
+            | EventKind::PairingCompleted { device }
+            | EventKind::PairingDenied { device } => Some(&device.name),
+            _ => None,
+        }
+    }
+
+    /// The certificate fingerprint this event carries, if any.
+    pub fn fingerprint(&self) -> Option<&str> {
+        match self {
+            EventKind::ClientConnected { client }
+            | EventKind::ClientDisconnected { client, .. } => client.fingerprint.as_deref(),
+            EventKind::PairingPending { device }
+            | EventKind::PairingCompleted { device }
+            | EventKind::PairingDenied { device } => Some(&device.fingerprint),
+            _ => None,
+        }
+    }
+
+    /// The protocol plane this event carries, if any.
+    pub fn plane(&self) -> Option<Plane> {
+        match self {
+            EventKind::ClientConnected { client }
+            | EventKind::ClientDisconnected { client, .. } => Some(client.plane),
+            EventKind::StreamStarted { stream } | EventKind::StreamStopped { stream } => {
+                Some(stream.plane)
+            }
+            EventKind::PairingPending { device }
+            | EventKind::PairingCompleted { device }
+            | EventKind::PairingDenied { device } => Some(device.plane),
+            _ => None,
+        }
+    }
+
+    /// The launched app id/title this event carries, if any.
+    pub fn app(&self) -> Option<&str> {
+        match self {
+            EventKind::StreamStarted { stream } | EventKind::StreamStopped { stream } => {
+                stream.app.as_deref()
+            }
+            _ => None,
+        }
+    }
+}
+
+/// Does `pattern` select `kind`? Exact kind names (`stream.started`) or `domain.*` prefixes
+/// matched on the dot boundary (`stream.*` matches `stream.started`, never `streamx.started`).
+/// One vocabulary for the SSE `?kinds=` filter and the hooks `on:` field.
+pub fn kind_matches(pattern: &str, kind: &str) -> bool {
+    match pattern.strip_suffix(".*") {
+        Some(prefix) => kind
+            .strip_prefix(prefix)
+            .is_some_and(|rest| rest.starts_with('.')),
+        None => pattern == kind,
+    }
+}
+
 /// Formats a mode as the wire's `WxH@Hz` string.
 pub fn mode_str(width: u32, height: u32, hz: u32) -> String {
     format!("{width}x{height}@{hz}")
@@ -262,13 +333,19 @@ impl EventBus {
         let _ = self.tx.send(ev);
     }
 
+    /// A live-tail-only subscription (no catch-up, no cursor) — for host-internal consumers
+    /// like the hook runner that only care about events from now on.
+    pub fn subscribe_live(&self) -> broadcast::Receiver<HostEvent> {
+        self.tx.subscribe()
+    }
+
     /// Subscribe with a resume cursor: events with `seq > since` come back as catch-up, the
     /// returned receiver carries everything after. `since = 0` means "from the ring start".
     pub fn subscribe(&self, since: u64) -> Subscription {
         let ring = self.inner.lock().unwrap_or_else(|e| e.into_inner());
         let rx = self.tx.subscribe();
         let first_seq = ring.events.front().map_or(ring.next_seq, |e| e.seq);
-        let dropped = since != 0 && since + 1 < first_seq;
+        let dropped = since != 0 && since.saturating_add(1) < first_seq;
         let catch_up = ring
             .events
             .iter()

crates/punktfunk-host/src/gamestream/mod.rs

@@ -242,6 +242,9 @@ pub fn serve(
         // rustls needs a process-wide crypto provider before any TLS config is built.
         let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
         let native_opts = crate::native::native_serve_opts(&native);
+        // The hook runner consumes the live event tail for the host's lifetime — spawned BEFORE
+        // `host.started` is emitted so operator hooks observe the full lifecycle (RFC §6).
+        tokio::spawn(crate::hooks::runner());
         // Lifecycle events (RFC §4): `host.started` as the serve planes come up; `host.stopping`
         // when they wind down (clean end OR error exit) — the ring holds it for a consumer that
         // reconnects, and a graceful-signal path can move the emit earlier when one exists.

crates/punktfunk-host/src/hooks.rs

@@ -0,0 +1,843 @@
+//! Operator hooks: commands and webhooks fired on host lifecycle events
+//! (scripting-and-hooks RFC §6, M2).
+//!
+//! `<config_dir>/hooks.json` holds a list of [`HookEntry`]s — *what to run on which event* —
+//! managed over `GET|PUT /api/v1/hooks` and applied immediately (the runner reads the store
+//! per event). The runner subscribes to the [`crate::events`] bus and dispatches matching
+//! entries **fire-and-forget**: a hook observes; it can never veto or delay a connection,
+//! stream, or pairing decision (decisions are made asynchronously through the API — RFC §6).
+//!
+//! Two actions:
+//! - **`run`** — a shell command, executed detached with the event JSON on stdin plus flat
+//!   `PF_EVENT_*` env vars (the [`crate::stream_marker`] `PF_STREAM_*` vocabulary's sibling).
+//!   Per-hook timeout (default 30 s) kills the whole process group on expiry; reaped
+//!   off-thread (the `try_recover_session` recipe). On a SYSTEM-service Windows host the
+//!   command runs **in the interactive user session** (never SYSTEM); that path cannot carry
+//!   per-process env/stdin, so the event JSON lands in a temp file appended as the command's
+//!   last argument (a console-mode Windows host gets env + stdin like Unix).
+//! - **`webhook`** — POST the event JSON to an operator URL. TLS-verified, redirects are not
+//!   followed, no punktfunk credentials are attached; an optional per-hook secret file yields
+//!   an `X-Punktfunk-Signature: sha256=<hex HMAC>` header so the receiver can authenticate us.
+//!
+//! Bounds (RFC §9.6): at most [`MAX_CONCURRENT_HOOKS`] hook executions in flight (excess
+//! firings are dropped with a warning, never queued unboundedly), per-hook `debounce_ms`, the
+//! exec timeout + process-group kill. Trust model (RFC §9.1): `hooks.json` is
+//! operator-privileged config in the DACL'd/0700 config dir; before executing a hook whose
+//! command is a script *path*, the host verifies the file is owned by the operator (or root)
+//! and not group/world-writable — the sshd/sudoers rule — and refuses loudly otherwise.
+
+use anyhow::Result;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::{Mutex, OnceLock};
+use std::time::{Duration, Instant};
+use utoipa::ToSchema;
+
+/// Concurrent hook executions in flight (exec + webhook combined). Excess firings are dropped
+/// with a warning — hooks are best-effort observers, and unbounded queueing is the failure
+/// mode this cap exists to prevent.
+const MAX_CONCURRENT_HOOKS: usize = 8;
+
+/// Default and ceiling for the exec timeout.
+const DEFAULT_TIMEOUT_S: u32 = 30;
+const MAX_TIMEOUT_S: u32 = 600;
+
+/// Outbound webhook timeout (connect + response).
+const WEBHOOK_TIMEOUT: Duration = Duration::from_secs(10);
+
+fn default_timeout_s() -> u32 {
+    DEFAULT_TIMEOUT_S
+}
+
+/// The operator's hook configuration — the `hooks.json` document and the `/api/v1/hooks` body.
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug, Default)]
+pub struct HooksConfig {
+    #[serde(default)]
+    pub hooks: Vec<HookEntry>,
+}
+
+/// One hook: fire `run` and/or `webhook` when an event matching `on` (+ `filter`) occurs.
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug)]
+pub struct HookEntry {
+    /// Which events fire this hook: an exact kind (`stream.started`) or a `domain.*` prefix
+    /// (`pairing.*`) — the same vocabulary as the SSE `?kinds=` filter.
+    pub on: String,
+    /// Exact-match constraints on the event's fields; every present field must match.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub filter: Option<HookFilter>,
+    /// Shell command to execute (detached, event JSON on stdin + `PF_EVENT_*` env).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub run: Option<String>,
+    /// URL to POST the event JSON to.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub webhook: Option<String>,
+    /// Exec timeout in seconds (1–600, default 30); the process group is killed on expiry.
+    #[serde(default = "default_timeout_s")]
+    pub timeout_s: u32,
+    /// Minimum interval between firings of this hook, in milliseconds. 0 = fire every time.
+    #[serde(default)]
+    pub debounce_ms: u64,
+    /// File holding the webhook HMAC secret (`X-Punktfunk-Signature: sha256=<hex>`). The file
+    /// should be operator-owned and private; a world-readable secret is warned about.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[schema(value_type = Option<String>)]
+    pub hmac_secret_file: Option<PathBuf>,
+}
+
+/// Exact-match filters against an event's identity fields (RFC open-question 3: exact match
+/// only — anything richer is what the SDK is for). Absent fields don't constrain; a filter
+/// field set on an event kind that doesn't carry it (e.g. `client` on `host.started`) never
+/// matches.
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug, Default)]
+pub struct HookFilter {
+    /// Client/device name (for `session.*`: the short client label the Dashboard shows).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub client: Option<String>,
+    /// Certificate fingerprint (hex, case-insensitive).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub fingerprint: Option<String>,
+    /// Protocol plane (`native` / `gamestream`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub plane: Option<crate::events::Plane>,
+    /// Launched app id/title (`stream.*` events).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub app: Option<String>,
+}
+
+impl HookFilter {
+    fn matches(&self, kind: &crate::events::EventKind) -> bool {
+        if let Some(want) = &self.client {
+            if kind.client_name() != Some(want.as_str()) {
+                return false;
+            }
+        }
+        if let Some(want) = &self.fingerprint {
+            match kind.fingerprint() {
+                Some(fp) if fp.eq_ignore_ascii_case(want) => {}
+                _ => return false,
+            }
+        }
+        if let Some(want) = self.plane {
+            if kind.plane() != Some(want) {
+                return false;
+            }
+        }
+        if let Some(want) = &self.app {
+            if kind.app() != Some(want.as_str()) {
+                return false;
+            }
+        }
+        true
+    }
+}
+
+impl HooksConfig {
+    /// Validate for the mgmt PUT: structural errors are rejected (the config would silently do
+    /// nothing or something surprising); unknown kinds are accepted (additive event catalog).
+    pub fn validate(&self) -> Result<(), String> {
+        for (i, h) in self.hooks.iter().enumerate() {
+            let at = |msg: &str| format!("hooks[{i}]: {msg}");
+            if h.on.trim().is_empty() {
+                return Err(at("`on` must be an event kind or `domain.*` pattern"));
+            }
+            if h.run.as_deref().is_none_or(|r| r.trim().is_empty())
+                && h.webhook.as_deref().is_none_or(|w| w.trim().is_empty())
+            {
+                return Err(at("needs `run` and/or `webhook`"));
+            }
+            if let Some(url) = h.webhook.as_deref().filter(|w| !w.trim().is_empty()) {
+                if !url.starts_with("https://") && !url.starts_with("http://") {
+                    return Err(at("`webhook` must be an http(s):// URL"));
+                }
+            }
+            if h.timeout_s == 0 || h.timeout_s > MAX_TIMEOUT_S {
+                return Err(at(&format!("`timeout_s` must be 1–{MAX_TIMEOUT_S}")));
+            }
+        }
+        Ok(())
+    }
+}
+
+// ------------------------------------------------------------------------- store
+
+/// The persisted hooks store — the [`crate::vdisplay::policy::DisplayPolicyStore`] recipe:
+/// private dir, temp-write + atomic rename, in-memory value changes only if the write succeeds.
+pub struct HooksStore {
+    path: PathBuf,
+    cur: Mutex<Option<HooksConfig>>,
+}
+
+impl HooksStore {
+    /// Load from `path`. Missing file ⇒ no hooks; corrupt file ⇒ no hooks with a warning
+    /// (never fail host startup over a settings file).
+    pub fn load_from(path: PathBuf) -> Self {
+        let cur = match std::fs::read(&path) {
+            Ok(bytes) => match serde_json::from_slice::<HooksConfig>(&bytes) {
+                Ok(c) => {
+                    if let Err(e) = c.validate() {
+                        tracing::warn!(path = %path.display(),
+                            "hooks.json invalid — hooks disabled until fixed: {e}");
+                        None
+                    } else {
+                        Some(c)
+                    }
+                }
+                Err(e) => {
+                    tracing::warn!(path = %path.display(),
+                        "hooks.json unreadable — hooks disabled until fixed: {e}");
+                    None
+                }
+            },
+            Err(_) => None,
+        };
+        HooksStore {
+            path,
+            cur: Mutex::new(cur),
+        }
+    }
+
+    /// The stored configuration (empty when unconfigured) — the mgmt GET and the dispatcher.
+    pub fn get(&self) -> HooksConfig {
+        self.cur.lock().unwrap().clone().unwrap_or_default()
+    }
+
+    /// Persist + adopt a new configuration (caller validates first). The in-memory value
+    /// changes only if the disk write succeeds.
+    pub fn set(&self, cfg: HooksConfig) -> Result<()> {
+        if let Some(dir) = self.path.parent() {
+            crate::gamestream::create_private_dir(dir)?;
+        }
+        let tmp = self.path.with_extension("json.tmp");
+        crate::gamestream::write_secret_file(&tmp, &serde_json::to_vec_pretty(&cfg)?)?;
+        std::fs::rename(&tmp, &self.path)?;
+        *self.cur.lock().unwrap() = Some(cfg);
+        Ok(())
+    }
+}
+
+/// The process-wide hooks store (`<config_dir>/hooks.json`), loaded once on first access.
+pub fn store() -> &'static HooksStore {
+    static STORE: OnceLock<HooksStore> = OnceLock::new();
+    STORE.get_or_init(|| HooksStore::load_from(crate::gamestream::config_dir().join("hooks.json")))
+}
+
+// ------------------------------------------------------------------------- runner
+
+/// The hook runner: a host-lifetime task consuming the live event tail and dispatching
+/// matching hooks. Spawned by `serve()` before `host.started` is emitted, so hooks can
+/// observe the full host lifetime. Lag (more events than the runner drained) skips the
+/// missed events with a warning — fire-and-forget, never a queue that grows unboundedly.
+pub async fn runner() {
+    let mut rx = crate::events::bus().subscribe_live();
+    let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_HOOKS));
+    let mut debounce: HashMap<u64, Instant> = HashMap::new();
+    loop {
+        match rx.recv().await {
+            Ok(ev) => dispatch(&ev, &sem, &mut debounce),
+            Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                tracing::warn!(
+                    missed = n,
+                    "hook runner lagged — skipped events fire no hooks"
+                );
+            }
+            Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
+        }
+    }
+}
+
+/// Stable identity for a hook entry across config reloads (the debounce key): the hash of its
+/// serialized form — an unchanged entry keeps its debounce window across a PUT.
+fn entry_key(h: &HookEntry) -> u64 {
+    use std::hash::{Hash, Hasher};
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    serde_json::to_string(h)
+        .unwrap_or_default()
+        .hash(&mut hasher);
+    hasher.finish()
+}
+
+fn dispatch(
+    ev: &crate::events::HostEvent,
+    sem: &std::sync::Arc<tokio::sync::Semaphore>,
+    debounce: &mut HashMap<u64, Instant>,
+) {
+    let kind = ev.kind.name();
+    let cfg = store().get();
+    for h in &cfg.hooks {
+        if !crate::events::kind_matches(&h.on, kind) {
+            continue;
+        }
+        if !h
+            .filter
+            .as_ref()
+            .unwrap_or(&HookFilter::default())
+            .matches(&ev.kind)
+        {
+            continue;
+        }
+        if h.debounce_ms > 0 {
+            let key = entry_key(h);
+            let now = Instant::now();
+            if debounce
+                .get(&key)
+                .is_some_and(|t| now.duration_since(*t) < Duration::from_millis(h.debounce_ms))
+            {
+                tracing::debug!(on = %h.on, kind, "hook debounced");
+                continue;
+            }
+            debounce.insert(key, now);
+        }
+        if let Some(cmd) = h.run.as_deref().filter(|c| !c.trim().is_empty()) {
+            fire_exec(cmd.to_string(), ev, h.timeout_s, sem);
+        }
+        if let Some(url) = h.webhook.as_deref().filter(|u| !u.trim().is_empty()) {
+            fire_webhook(url.to_string(), h.hmac_secret_file.clone(), ev, sem);
+        }
+    }
+    // The two env-var mirrors (`PUNKTFUNK_ON_CONNECT_CMD` / `PUNKTFUNK_ON_DISCONNECT_CMD`) —
+    // the zero-config siblings of `PUNKTFUNK_RECOVER_SESSION_CMD` for the simplest cases.
+    let mirror = match kind {
+        "client.connected" => crate::config::config().on_connect_cmd.clone(),
+        "client.disconnected" => crate::config::config().on_disconnect_cmd.clone(),
+        _ => None,
+    };
+    if let Some(cmd) = mirror {
+        fire_exec(cmd, ev, DEFAULT_TIMEOUT_S, sem);
+    }
+}
+
+// ------------------------------------------------------------------------- exec action
+
+fn fire_exec(
+    cmd: String,
+    ev: &crate::events::HostEvent,
+    timeout_s: u32,
+    sem: &std::sync::Arc<tokio::sync::Semaphore>,
+) {
+    let Ok(permit) = sem.clone().try_acquire_owned() else {
+        tracing::warn!(cmd = %cmd, "hook dropped — too many hook executions in flight");
+        return;
+    };
+    if let Err(e) = exec_path_check(&cmd) {
+        tracing::error!(cmd = %cmd, "REFUSING hook command — {e}");
+        return;
+    }
+    let json = serde_json::to_string(ev).unwrap_or_else(|_| "{}".to_string());
+    let env = flatten_env(ev);
+    let kind = ev.kind.name();
+    let timeout = Duration::from_secs(u64::from(timeout_s));
+    tracing::info!(cmd = %cmd, kind, "hook: running command");
+    // Detached execution + off-thread reap (the `try_recover_session` recipe): the streaming
+    // planes never wait on operator code. The permit rides along and frees on thread exit.
+    std::thread::spawn(move || {
+        run_hook_process(&cmd, &json, &env, timeout);
+        drop(permit);
+    });
+}
+
+/// The event flattened to `PF_EVENT_*` env vars: scalar leaves of the event JSON, path-joined
+/// with `_` and uppercased (`client.name` → `PF_EVENT_CLIENT_NAME`), plus `PF_EVENT_JSON` with
+/// the whole document. Values are control-char-stripped so a hostile device name can't smuggle
+/// newlines into a naive shell consumer.
+fn flatten_env(ev: &crate::events::HostEvent) -> Vec<(String, String)> {
+    fn walk(prefix: &str, v: &serde_json::Value, out: &mut Vec<(String, String)>) {
+        match v {
+            serde_json::Value::Object(map) => {
+                for (k, val) in map {
+                    let key = k
+                        .chars()
+                        .map(|c| {
+                            if c.is_ascii_alphanumeric() {
+                                c.to_ascii_uppercase()
+                            } else {
+                                '_'
+                            }
+                        })
+                        .collect::<String>();
+                    walk(&format!("{prefix}_{key}"), val, out);
+                }
+            }
+            serde_json::Value::Null => {}
+            serde_json::Value::String(s) => {
+                let clean: String = s.chars().filter(|c| !c.is_control()).collect();
+                out.push((prefix.to_string(), clean));
+            }
+            other => out.push((prefix.to_string(), other.to_string())),
+        }
+    }
+    let mut out = Vec::new();
+    if let Ok(v) = serde_json::to_value(ev) {
+        walk("PF_EVENT", &v, &mut out);
+    }
+    if let Ok(json) = serde_json::to_string(ev) {
+        out.push(("PF_EVENT_JSON".to_string(), json));
+    }
+    out
+}
+
+/// The sshd/sudoers rule (RFC §9.1): when the command's first token is a path to an existing
+/// file, refuse to run it unless it is owned by the host user (or root) and not
+/// group/world-writable — a world-writable hook script is privilege escalation bait. A bare
+/// command name (`systemctl`, `curl`) is left to PATH.
+#[cfg(unix)]
+fn exec_path_check(cmd: &str) -> Result<(), String> {
+    use std::os::unix::fs::MetadataExt;
+    let Some(first) = cmd.split_whitespace().next() else {
+        return Err("empty command".into());
+    };
+    if !first.starts_with('/') {
+        return Ok(());
+    }
+    let meta = match std::fs::metadata(first) {
+        Ok(m) => m,
+        Err(_) => return Ok(()), // not an existing file — the shell will report it
+    };
+    if !meta.is_file() {
+        return Ok(());
+    }
+    // SAFETY: geteuid has no preconditions and touches no memory.
+    let euid = unsafe { libc::geteuid() };
+    if meta.uid() != euid && meta.uid() != 0 {
+        return Err(format!(
+            "{first} is owned by uid {} (host runs as uid {euid}) — hook scripts must be \
+             owned by the operator or root",
+            meta.uid()
+        ));
+    }
+    if meta.mode() & 0o022 != 0 {
+        return Err(format!(
+            "{first} is group/world-writable (mode {:o}) — chmod go-w it first",
+            meta.mode() & 0o7777
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn exec_path_check(_cmd: &str) -> Result<(), String> {
+    // Windows: hooks.json lives in the SYSTEM/Admins-DACL'd config dir and the command runs in
+    // the interactive user session (never SYSTEM) — the config itself is the trust boundary.
+    // A per-script ACL check is a hardening follow-up.
+    Ok(())
+}
+
+/// Run one hook command to completion (or timeout), blocking the reaper thread it runs on.
+#[cfg(unix)]
+fn run_hook_process(cmd: &str, event_json: &str, env: &[(String, String)], timeout: Duration) {
+    use std::io::Write;
+    use std::os::unix::process::CommandExt;
+    let mut c = std::process::Command::new("/bin/sh");
+    c.arg("-c")
+        .arg(cmd)
+        .stdin(std::process::Stdio::piped())
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null())
+        // Its own process group, so the timeout can kill the whole tree the shell spawned.
+        .process_group(0);
+    c.envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str())));
+    let mut child = match c.spawn() {
+        Ok(ch) => ch,
+        Err(e) => {
+            tracing::error!(cmd = %cmd, error = %e, "hook command failed to launch");
+            return;
+        }
+    };
+    if let Some(mut stdin) = child.stdin.take() {
+        let _ = stdin.write_all(event_json.as_bytes());
+        // stdin drops (closes) here — a hook that never reads it is unaffected.
+    }
+    let deadline = Instant::now() + timeout;
+    loop {
+        match child.try_wait() {
+            Ok(Some(status)) => {
+                if !status.success() {
+                    tracing::warn!(cmd = %cmd, %status, "hook command exited non-zero");
+                }
+                return;
+            }
+            Ok(None) => {
+                if Instant::now() >= deadline {
+                    tracing::warn!(cmd = %cmd, timeout_s = timeout.as_secs(),
+                        "hook command timed out — killing its process group");
+                    #[cfg(target_os = "linux")]
+                    {
+                        // SAFETY: kill(2) with a negative pid signals the process group we
+                        // created via process_group(0); no memory is touched.
+                        unsafe { libc::kill(-(child.id() as i32), libc::SIGKILL) };
+                    }
+                    #[cfg(not(target_os = "linux"))]
+                    let _ = child.kill();
+                    let _ = child.wait(); // reap — never leave a zombie
+                    return;
+                }
+                std::thread::sleep(Duration::from_millis(100));
+            }
+            Err(e) => {
+                tracing::warn!(cmd = %cmd, error = %e, "hook command wait failed");
+                return;
+            }
+        }
+    }
+}
+
+/// Windows: on a SYSTEM host the command must run in the interactive user session
+/// ([`crate::interactive::spawn_in_active_session`], never SYSTEM) — that path can't carry
+/// per-process env or stdin, so the event JSON is written to a private temp file whose path is
+/// appended as the command's last argument. A console-mode host (dev) falls back to a plain
+/// spawn with the full Unix-style context (env + stdin).
+#[cfg(windows)]
+fn run_hook_process(cmd: &str, event_json: &str, env: &[(String, String)], timeout: Duration) {
+    use std::io::Write;
+    let stamp = format!(
+        "pf-hook-{}-{}.json",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_nanos())
+            .unwrap_or(0)
+    );
+    let json_path = std::env::temp_dir().join(stamp);
+    if std::fs::write(&json_path, event_json).is_err() {
+        tracing::warn!(cmd = %cmd, "hook: could not write event JSON temp file");
+    }
+    let cmdline = format!("{cmd} \"{}\"", json_path.display());
+    match crate::interactive::spawn_in_active_session(&cmdline, None) {
+        Ok(pid) => {
+            tracing::debug!(cmd = %cmd, pid, "hook command launched in the interactive session");
+            // No child handle on this path — wait out the timeout, then clean the temp file.
+            std::thread::sleep(timeout);
+            let _ = std::fs::remove_file(&json_path);
+        }
+        Err(e) => {
+            tracing::debug!(error = %format!("{e:#}"),
+                "interactive-session spawn unavailable — running hook in-console");
+            let mut c = std::process::Command::new("cmd.exe");
+            c.arg("/C")
+                .arg(cmd)
+                .stdin(std::process::Stdio::piped())
+                .stdout(std::process::Stdio::null())
+                .stderr(std::process::Stdio::null());
+            c.envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str())));
+            match c.spawn() {
+                Ok(mut child) => {
+                    if let Some(mut stdin) = child.stdin.take() {
+                        let _ = stdin.write_all(event_json.as_bytes());
+                    }
+                    let deadline = Instant::now() + timeout;
+                    while child.try_wait().ok().flatten().is_none() {
+                        if Instant::now() >= deadline {
+                            tracing::warn!(cmd = %cmd, "hook command timed out — killing it");
+                            let _ = child.kill();
+                            let _ = child.wait();
+                            break;
+                        }
+                        std::thread::sleep(Duration::from_millis(100));
+                    }
+                }
+                Err(e) => tracing::error!(cmd = %cmd, error = %e, "hook command failed to launch"),
+            }
+            let _ = std::fs::remove_file(&json_path);
+        }
+    }
+}
+
+// ------------------------------------------------------------------------- webhook action
+
+fn fire_webhook(
+    url: String,
+    secret_file: Option<PathBuf>,
+    ev: &crate::events::HostEvent,
+    sem: &std::sync::Arc<tokio::sync::Semaphore>,
+) {
+    let Ok(permit) = sem.clone().try_acquire_owned() else {
+        tracing::warn!(url = %url, "webhook dropped — too many hook executions in flight");
+        return;
+    };
+    let json = serde_json::to_string(ev).unwrap_or_else(|_| "{}".to_string());
+    let kind = ev.kind.name();
+    tracing::info!(url = %url, kind, "hook: posting webhook");
+    std::thread::spawn(move || {
+        post_webhook(&url, &json, secret_file.as_deref());
+        drop(permit);
+    });
+}
+
+fn post_webhook(url: &str, json: &str, secret_file: Option<&std::path::Path>) {
+    // TLS is verified (ureq's default rustls roots); redirects are never followed, so a
+    // compromised receiver can't bounce the POST cross-origin (RFC §9.5).
+    let agent = ureq::builder()
+        .redirects(0)
+        .timeout(WEBHOOK_TIMEOUT)
+        .build();
+    let mut req = agent.post(url).set("Content-Type", "application/json");
+    if let Some(path) = secret_file {
+        match std::fs::read(path) {
+            Ok(secret) => {
+                use hmac::{Hmac, Mac};
+                let mut mac = match Hmac::<sha2::Sha256>::new_from_slice(&secret) {
+                    Ok(m) => m,
+                    Err(_) => {
+                        tracing::error!(path = %path.display(), "webhook HMAC secret unusable");
+                        return;
+                    }
+                };
+                mac.update(json.as_bytes());
+                let sig = hex::encode(mac.finalize().into_bytes());
+                req = req.set("X-Punktfunk-Signature", &format!("sha256={sig}"));
+            }
+            Err(e) => {
+                // A configured-but-unreadable secret means the operator WANTS signing —
+                // failing open (unsigned POST) would defeat the receiver's authentication.
+                tracing::error!(path = %path.display(), error = %e,
+                    "webhook HMAC secret unreadable — NOT posting unsigned");
+                return;
+            }
+        }
+    }
+    match req.send_string(json) {
+        Ok(resp) => tracing::debug!(url, status = resp.status(), "webhook delivered"),
+        Err(ureq::Error::Status(code, _)) => {
+            tracing::warn!(url, status = code, "webhook rejected by receiver")
+        }
+        Err(e) => tracing::warn!(url, error = %e, "webhook delivery failed"),
+    }
+}
+
+// ------------------------------------------------------------------------- tests
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::events::{ClientRef, EventKind, HostEvent, Plane, StreamRef};
+
+    fn sample_event() -> HostEvent {
+        HostEvent {
+            seq: 7,
+            ts_ms: 1_700_000_000_000,
+            schema: 1,
+            kind: EventKind::StreamStarted {
+                stream: StreamRef {
+                    mode: "2560x1440@120".into(),
+                    hdr: true,
+                    client: "Living Room TV".into(),
+                    app: Some("steam:570".into()),
+                    plane: Plane::Native,
+                },
+            },
+        }
+    }
+
+    #[test]
+    fn validation_rejects_structural_errors() {
+        let ok = HooksConfig {
+            hooks: vec![HookEntry {
+                on: "stream.*".into(),
+                filter: None,
+                run: Some("echo hi".into()),
+                webhook: None,
+                timeout_s: 30,
+                debounce_ms: 0,
+                hmac_secret_file: None,
+            }],
+        };
+        assert!(ok.validate().is_ok());
+
+        let mut bad = ok.clone();
+        bad.hooks[0].on = " ".into();
+        assert!(bad.validate().is_err(), "empty `on`");
+
+        let mut bad = ok.clone();
+        bad.hooks[0].run = None;
+        assert!(bad.validate().is_err(), "no action");
+
+        let mut bad = ok.clone();
+        bad.hooks[0].webhook = Some("ftp://nope".into());
+        assert!(bad.validate().is_err(), "non-http webhook");
+
+        let mut bad = ok.clone();
+        bad.hooks[0].timeout_s = 0;
+        assert!(bad.validate().is_err(), "zero timeout");
+        bad.hooks[0].timeout_s = 601;
+        assert!(bad.validate().is_err(), "over-ceiling timeout");
+    }
+
+    #[test]
+    fn store_roundtrips_and_survives_corruption() {
+        let path = std::env::temp_dir().join(format!(
+            "pf-hooks-test-{}-{:p}.json",
+            std::process::id(),
+            &0u8 as *const u8
+        ));
+        let _ = std::fs::remove_file(&path);
+
+        let store = HooksStore::load_from(path.clone());
+        assert!(store.get().hooks.is_empty(), "unconfigured = no hooks");
+
+        let cfg = HooksConfig {
+            hooks: vec![HookEntry {
+                on: "pairing.pending".into(),
+                filter: Some(HookFilter {
+                    plane: Some(Plane::Native),
+                    ..Default::default()
+                }),
+                run: None,
+                webhook: Some("https://ha.local/api/webhook/punktfunk".into()),
+                timeout_s: 30,
+                debounce_ms: 500,
+                hmac_secret_file: None,
+            }],
+        };
+        store.set(cfg).unwrap();
+        assert_eq!(store.get().hooks.len(), 1);
+
+        // A fresh load sees the persisted value.
+        let reload = HooksStore::load_from(path.clone());
+        assert_eq!(reload.get().hooks.len(), 1);
+        assert_eq!(reload.get().hooks[0].on, "pairing.pending");
+
+        // Corruption never breaks startup — it just disables hooks loudly.
+        std::fs::write(&path, b"{ not json").unwrap();
+        let corrupt = HooksStore::load_from(path.clone());
+        assert!(corrupt.get().hooks.is_empty());
+        let _ = std::fs::remove_file(&path);
+    }
+
+    #[test]
+    fn filters_constrain_and_missing_fields_never_match() {
+        let ev = sample_event();
+        let f = HookFilter {
+            client: Some("Living Room TV".into()),
+            app: Some("steam:570".into()),
+            plane: Some(Plane::Native),
+            ..Default::default()
+        };
+        assert!(f.matches(&ev.kind));
+
+        let f = HookFilter {
+            client: Some("Bedroom".into()),
+            ..Default::default()
+        };
+        assert!(!f.matches(&ev.kind));
+
+        let f = HookFilter {
+            plane: Some(Plane::Gamestream),
+            ..Default::default()
+        };
+        assert!(!f.matches(&ev.kind));
+
+        // stream.* events carry no fingerprint — a fingerprint filter can't match them.
+        let f = HookFilter {
+            fingerprint: Some("ab12".into()),
+            ..Default::default()
+        };
+        assert!(!f.matches(&ev.kind));
+
+        // Fingerprint matching is case-insensitive where the field exists.
+        let connected = EventKind::ClientConnected {
+            client: ClientRef {
+                name: "Deck".into(),
+                fingerprint: Some("AB12CD".into()),
+                plane: Plane::Native,
+            },
+        };
+        let f = HookFilter {
+            fingerprint: Some("ab12cd".into()),
+            ..Default::default()
+        };
+        assert!(f.matches(&connected));
+    }
+
+    #[test]
+    fn env_flattening_is_shell_safe_and_complete() {
+        let ev = sample_event();
+        let env = flatten_env(&ev);
+        let get = |k: &str| {
+            env.iter()
+                .find(|(key, _)| key == k)
+                .map(|(_, v)| v.as_str())
+        };
+        assert_eq!(get("PF_EVENT_KIND"), Some("stream.started"));
+        assert_eq!(get("PF_EVENT_SEQ"), Some("7"));
+        assert_eq!(get("PF_EVENT_STREAM_MODE"), Some("2560x1440@120"));
+        assert_eq!(get("PF_EVENT_STREAM_HDR"), Some("true"));
+        assert_eq!(get("PF_EVENT_STREAM_CLIENT"), Some("Living Room TV"));
+        assert_eq!(get("PF_EVENT_STREAM_APP"), Some("steam:570"));
+        assert_eq!(get("PF_EVENT_STREAM_PLANE"), Some("native"));
+        assert!(get("PF_EVENT_JSON").unwrap().contains("\"seq\":7"));
+
+        // A hostile client name can't smuggle control chars into env consumers.
+        let mut evil = sample_event();
+        if let EventKind::StreamStarted { stream } = &mut evil.kind {
+            stream.client = "evil\nname\r\t".into();
+        }
+        let env = flatten_env(&evil);
+        let v = env
+            .iter()
+            .find(|(k, _)| k == "PF_EVENT_STREAM_CLIENT")
+            .map(|(_, v)| v.clone())
+            .unwrap();
+        assert_eq!(v, "evilname");
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn exec_runs_with_stdin_and_env_and_timeout_kills() {
+        // A hook that proves stdin + env delivery by writing both to a file.
+        let out = std::env::temp_dir().join(format!(
+            "pf-hook-exec-{}-{:p}.txt",
+            std::process::id(),
+            &0u8 as *const u8
+        ));
+        let _ = std::fs::remove_file(&out);
+        let ev = sample_event();
+        let env = flatten_env(&ev);
+        let json = serde_json::to_string(&ev).unwrap();
+        run_hook_process(
+            &format!(
+                "printf '%s|' \"$PF_EVENT_KIND\" > {p}; cat >> {p}",
+                p = out.display()
+            ),
+            &json,
+            &env,
+            Duration::from_secs(5),
+        );
+        let text = std::fs::read_to_string(&out).expect("hook wrote its file");
+        assert!(text.starts_with("stream.started|"), "env delivered: {text}");
+        assert!(text.contains("\"seq\":7"), "stdin delivered: {text}");
+        let _ = std::fs::remove_file(&out);
+
+        // Timeout: a sleeping hook is killed (process group) well before its sleep ends.
+        let started = Instant::now();
+        run_hook_process("sleep 30", &json, &env, Duration::from_secs(1));
+        assert!(
+            started.elapsed() < Duration::from_secs(5),
+            "timeout must kill the hook, not wait it out"
+        );
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn ownership_check_refuses_world_writable_scripts() {
+        use std::os::unix::fs::PermissionsExt;
+        let path = std::env::temp_dir().join(format!(
+            "pf-hook-own-{}-{:p}.sh",
+            std::process::id(),
+            &0u8 as *const u8
+        ));
+        std::fs::write(&path, "#!/bin/sh\ntrue\n").unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700)).unwrap();
+        assert!(exec_path_check(&format!("{} arg", path.display())).is_ok());
+
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o777)).unwrap();
+        assert!(
+            exec_path_check(&format!("{} arg", path.display())).is_err(),
+            "world-writable script must be refused"
+        );
+        let _ = std::fs::remove_file(&path);
+
+        // Bare command names are left to PATH; nonexistent paths are the shell's problem.
+        assert!(exec_path_check("systemctl suspend").is_ok());
+        assert!(exec_path_check("/nonexistent/definitely-not-here").is_ok());
+    }
+}

crates/punktfunk-host/src/main.rs

@@ -50,6 +50,7 @@ mod gpu;
 #[path = "linux/gpuclocks.rs"]
 mod gpuclocks;
 mod hdr;
+mod hooks;
 mod inject;
 #[cfg(target_os = "windows")]
 #[path = "windows/install.rs"]

crates/punktfunk-host/src/mgmt.rs

@@ -34,6 +34,7 @@ mod clients;
 mod display;
 mod events;
 mod gpu;
+mod hooks;
 mod host;
 mod library;
 mod native;
@@ -217,7 +218,8 @@ fn api_router_parts() -> (Router<Arc<MgmtState>>, utoipa::openapi::OpenApi) {
                     stats::stats_recording_delete
                 ))
                 .routes(routes!(stats::logs_get))
-                .routes(routes!(events::stream_events)),
+                .routes(routes!(events::stream_events))
+                .routes(routes!(hooks::get_hooks, hooks::set_hooks)),
         )
         .split_for_parts()
 }
@@ -254,6 +256,7 @@ pub fn openapi_json() -> String {
         (name = "stats", description = "Streaming performance-stats capture: arm/stop a recording, read the live + saved time-series for graphing"),
         (name = "logs", description = "Host log stream: the newest in-memory log entries, cursor-paged for live following"),
         (name = "events", description = "Host lifecycle events: an SSE stream (client/session/stream lifecycle, pairing, displays, library, host) with Last-Event-ID resume and server-side kind filters"),
+        (name = "hooks", description = "Operator hooks: commands and webhooks fired on lifecycle events (fire-and-forget — hooks observe, never veto)"),
     )
 )]
 struct ApiDoc;

crates/punktfunk-host/src/mgmt/events.rs

@@ -72,12 +72,7 @@ impl KindFilter {
     fn matches(&self, kind: &str) -> bool {
         match &self.0 {
             None => true,
-            Some(pats) => pats.iter().any(|p| match p.strip_suffix(".*") {
+            Some(pats) => pats.iter().any(|p| crate::events::kind_matches(p, kind)),
-                Some(prefix) => kind
-                    .strip_prefix(prefix)
-                    .is_some_and(|rest| rest.starts_with('.')),
-                None => p == kind,
-            }),
         }
     }
 }

crates/punktfunk-host/src/mgmt/hooks.rs

@@ -0,0 +1,57 @@
+//! Hooks management endpoints (scripting-and-hooks RFC §6): read and replace the operator's
+//! `hooks.json` — validated on write, applied immediately (the runner reads the store per
+//! event, so no restart is needed).
+
+use super::shared::*;
+
+/// Get the hook configuration
+///
+/// The operator's `hooks.json`: commands and webhooks fired on host lifecycle events. Empty
+/// when unconfigured.
+#[utoipa::path(
+    get,
+    path = "/hooks",
+    tag = "hooks",
+    operation_id = "getHooks",
+    responses(
+        (status = OK, description = "The stored hook configuration", body = crate::hooks::HooksConfig),
+        (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError),
+    )
+)]
+pub(crate) async fn get_hooks() -> Json<crate::hooks::HooksConfig> {
+    Json(crate::hooks::store().get())
+}
+
+/// Replace the hook configuration
+///
+/// Validates and persists a full `hooks.json` document (this is a whole-document PUT, not a
+/// patch). Applies from the next event — no restart. Hook commands run as the host user
+/// (interactive user session on Windows): treat this configuration as operator-privileged.
+#[utoipa::path(
+    put,
+    path = "/hooks",
+    tag = "hooks",
+    operation_id = "setHooks",
+    request_body = crate::hooks::HooksConfig,
+    responses(
+        (status = OK, description = "Configuration stored; the new state", body = crate::hooks::HooksConfig),
+        (status = BAD_REQUEST, description = "Structurally invalid configuration", body = ApiError),
+        (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError),
+        (status = INTERNAL_SERVER_ERROR, description = "Configuration could not be persisted", body = ApiError),
+    )
+)]
+pub(crate) async fn set_hooks(ApiJson(cfg): ApiJson<crate::hooks::HooksConfig>) -> Response {
+    if let Err(e) = cfg.validate() {
+        return api_error(StatusCode::BAD_REQUEST, &e);
+    }
+    match crate::hooks::store().set(cfg) {
+        Ok(()) => {
+            tracing::info!("management API: hook configuration updated");
+            Json(crate::hooks::store().get()).into_response()
+        }
+        Err(e) => api_error(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("persist hooks.json: {e:#}"),
+        ),
+    }
+}

crates/punktfunk-host/src/mgmt/tests.rs

@@ -1155,3 +1155,54 @@ async fn events_stream_connection_cap() {
         .expect("infallible");
     assert_eq!(resp.status(), StatusCode::OK, "cap frees with the slots");
 }
+
+// ------------------------------------------------------------------ hooks
+
+/// GET returns the (empty-when-unconfigured) config; PUT validation rejects structural errors
+/// with the reason. A *successful* PUT is deliberately not exercised through the route — it
+/// would write the developer's real config dir; persistence is unit-tested in `crate::hooks`
+/// against a temp path.
+#[tokio::test]
+async fn hooks_get_shape_and_put_validation() {
+    let app = test_app(test_state(), None);
+
+    let (s, json) = send(&app, get_req("/api/v1/hooks")).await;
+    assert_eq!(s, StatusCode::OK);
+    assert!(json["hooks"].is_array());
+
+    let put = |body: serde_json::Value| {
+        axum::http::Request::put("/api/v1/hooks")
+            .header(axum::http::header::CONTENT_TYPE, "application/json")
+            .body(Body::from(body.to_string()))
+            .unwrap()
+    };
+
+    // Structurally invalid: an entry with no action.
+    let (s, json) = send(
+        &app,
+        put(serde_json::json!({"hooks": [{"on": "stream.started"}]})),
+    )
+    .await;
+    assert_eq!(s, StatusCode::BAD_REQUEST);
+    assert!(
+        json["error"].as_str().unwrap().contains("run"),
+        "error names the problem: {json}"
+    );
+
+    // Non-http(s) webhook.
+    let (s, _) = send(
+        &app,
+        put(serde_json::json!({"hooks": [{"on": "pairing.*", "webhook": "ftp://x"}]})),
+    )
+    .await;
+    assert_eq!(s, StatusCode::BAD_REQUEST);
+
+    // Wrong bearer → 401 (the hooks surface is admin-lane).
+    let mut req = get_req("/api/v1/hooks");
+    req.headers_mut().insert(
+        axum::http::header::AUTHORIZATION,
+        axum::http::HeaderValue::from_static("Bearer wrong"),
+    );
+    let resp = app.clone().oneshot(req).await.expect("infallible");
+    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
+}