stage-auto-a pulse: respond config (threshold/quorum/local-only)

Persisted-key/value helpers for the respond-pipeline auto-fire gates:
- respond_auto_threshold / set_… (Severity, default HIGH)
- respond_require_quorum / set_… (bool, default True)
- respond_local_only / set_…     (bool, default False)
Plus a _severity_rank helper for threshold comparison.

Backing store is the existing pulse_settings table; this commit also adds
generic pulse_setting_get / pulse_setting_set helpers in db.py so future
pulse settings don't each need their own column-pair helper.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

src/psyc/db.py

@@ -313,6 +313,25 @@ def kill_switch_set(armed: bool, db_path: Path = DB_PATH) -> None:
         conn.execute(stmt)
 
 
+def pulse_setting_get(key: str, db_path: Path = DB_PATH) -> Optional[str]:
+    """Read one pulse_settings row by key. None if unset."""
+    stmt = select(pulse_settings.c.value).where(pulse_settings.c.key == key)
+    with engine(db_path).connect() as conn:
+        row = conn.execute(stmt).fetchone()
+    return None if row is None else str(row.value)
+
+
+def pulse_setting_set(key: str, value: str, db_path: Path = DB_PATH) -> None:
+    """Upsert one pulse_settings row by key. Both args are strings."""
+    stmt = sqlite_insert(pulse_settings).values(key=key, value=value)
+    stmt = stmt.on_conflict_do_update(
+        index_elements=[pulse_settings.c.key],
+        set_=dict(value=stmt.excluded.value),
+    )
+    with engine(db_path).begin() as conn:
+        conn.execute(stmt)
+
+
 # ---------- federation: peers + signal buffer ----------------------------
 
 def upsert_peer(row: dict, db_path: Path = DB_PATH) -> None:

src/psyc/lines/pulse.py

@@ -22,6 +22,7 @@ from typing import Callable, Dict, List, Optional, Tuple
 from pydantic import BaseModel, Field
 
 from psyc import db, log
+from psyc.models import Severity
 
 
 _log = log.get(__name__)
@@ -33,6 +34,73 @@ class PulseMode(str, Enum):
     AUTO_EXECUTE = "auto-execute"
 
 
+# ---------- respond auto-fire gates -----------------------------------------
+# Persisted as rows in pulse_settings (key/value pairs). All defaults are
+# "safe" — quorum required, HIGH threshold, federation cases permitted only
+# when quorum-met.
+
+_KEY_RESPOND_THRESHOLD = "respond_auto_threshold"
+_KEY_RESPOND_REQUIRE_QUORUM = "respond_require_quorum"
+_KEY_RESPOND_LOCAL_ONLY = "respond_local_only"
+
+_DEFAULT_THRESHOLD = Severity.HIGH
+_DEFAULT_REQUIRE_QUORUM = True
+_DEFAULT_LOCAL_ONLY = False
+
+
+def _severity_rank(sev: Optional[Severity]) -> int:
+    """Rank order for severity threshold comparison. Unknown / None → -1."""
+    if sev is None:
+        return -1
+    return {
+        Severity.LOW: 0,
+        Severity.MEDIUM: 1,
+        Severity.HIGH: 2,
+        Severity.CRITICAL: 3,
+    }.get(sev, -1)
+
+
+def respond_auto_threshold() -> Severity:
+    raw = db.pulse_setting_get(_KEY_RESPOND_THRESHOLD)
+    if raw is None:
+        return _DEFAULT_THRESHOLD
+    try:
+        return Severity(raw)
+    except ValueError:
+        return _DEFAULT_THRESHOLD
+
+
+def set_respond_auto_threshold(sev: Severity) -> None:
+    if not isinstance(sev, Severity):
+        raise ValueError(f"not a Severity: {sev!r}")
+    db.pulse_setting_set(_KEY_RESPOND_THRESHOLD, sev.value)
+    _log.info("pulse.respond.threshold.changed", severity=sev.value)
+
+
+def respond_require_quorum() -> bool:
+    raw = db.pulse_setting_get(_KEY_RESPOND_REQUIRE_QUORUM)
+    if raw is None:
+        return _DEFAULT_REQUIRE_QUORUM
+    return raw == "1"
+
+
+def set_respond_require_quorum(state: bool) -> None:
+    db.pulse_setting_set(_KEY_RESPOND_REQUIRE_QUORUM, "1" if state else "0")
+    _log.info("pulse.respond.quorum.changed", required=bool(state))
+
+
+def respond_local_only() -> bool:
+    raw = db.pulse_setting_get(_KEY_RESPOND_LOCAL_ONLY)
+    if raw is None:
+        return _DEFAULT_LOCAL_ONLY
+    return raw == "1"
+
+
+def set_respond_local_only(state: bool) -> None:
+    db.pulse_setting_set(_KEY_RESPOND_LOCAL_ONLY, "1" if state else "0")
+    _log.info("pulse.respond.local-only.changed", local_only=bool(state))
+
+
 class Pipeline(BaseModel):
     name: str
     title: str