feat(core): global incident freeze kill-switch

SECRETS_ROTATION.md

@@ -4,17 +4,17 @@ Single source of truth for rotating the credentials that keep the GadgetLab
 autonomous agent pipeline alive. If any of these lapse, the pipeline degrades or
 goes fully offline.
 
-Last updated: 2026-07-09
+Last updated: 2026-08-04
 
 ## Inventory
 
 | Secret / Token | Where it lives | Expiry | Alert mechanism |
 |----------------|---------------|--------|-----------------|
-| GitHub PAT (`GITHUB_PAT`) | GitHub repo secret | **2027-05-07** | `.github/workflows/pat-rotation-alert.yml` (monthly, Slack, fails ≤7d) |
+| GitHub PAT (repo secret name is **`GH_PAT`** — NOT `GITHUB_PAT`) | GitHub repo secret | **2027-05-07** | `.github/workflows/pat-rotation-alert.yml` (monthly, Slack, fails ≤7d) + Hermes cron `c4a45ae6e6d0` |
 | `ANTHROPIC_API_KEY` | GitHub repo secret | — (no hard expiry; rotated 2026-04-27) | none — review every ~6 months |
 | `OPENAI_API_KEY` | GitHub repo secret | — (no hard expiry; rotated 2026-03-25) | none — review every ~6 months |
-| DRC `x-gadgetlab-token` | **n8n only** (embedded literal in 2 nodes) | Last rotated **2026-07-06** | **none** — see below |
-| `SLACK_WEBHOOK_URL` | GitHub repo secret | — | none |
+| DRC `x-gadgetlab-token` | **n8n only** (embedded literal in 2 nodes) | Last rotated **2026-07-06** | `.github/workflows/drc-token-rotation-alert.yml` (monthly, Slack, 180d max-age) + Hermes crons `56f0f862d9a1` (monthly) & `425e85478d7b` (75-day nudge) |
+| `SLACK_WEBHOOK_URL` / `SLACK_SIGNING_SECRET` | GitHub repo secret | — | none |
 
 ---
 
@@ -32,14 +32,14 @@ miss it without ignoring a failing workflow.
 1. Go to https://github.com/settings/tokens (or org token admin).
 2. Create a new fine-grained or classic PAT with `repo` + `workflow` scopes
    (matches what the workflows need).
-3. In the repo → Settings → Secrets and variables → Actions → update `GITHUB_PAT`.
+3. In the repo → Settings → Secrets and variables → Actions → update `GH_PAT` (the actual repo secret name; the runbook previously said `GITHUB_PAT` in error).
 4. Update the `EXPIRY_DATE` constant in `.github/workflows/pat-rotation-alert.yml`
    (line ~16) so the alert counter starts from the new date.
 5. Confirm the next scheduled run reports OK.
 
 ---
 
-## 2. DRC `x-gadgetlab-token` — NO automated alert (gap to close)
+## 2. DRC `x-gadgetlab-token` — alerting exists (workflow + Hermes cron)
 
 **What it is:** a static shared secret the GitHub Event Router presents in the
 `x-gadgetlab-token` header when forwarding to the DRC Agent Loop. The DRC loop's
@@ -62,12 +62,9 @@ inactivity and autosave silently fails with 401; Claude cannot re-auth):**
    result, not `Unauthorized`. The old value must now be rejected.
 6. Update "last rotated" date wherever it's tracked (CLAUDE.md + this file).
 
-**Recommended remediation (TODO, pending your call):**
-- Add a `drc-token-rotation-alert.yml` mirroring `pat-rotation-alert.yml` with a
-  hardcoded `LAST_ROTATED` date and a 90/30/7-day reminder cadence → Slack.
-- Better: move the token out of node literals into an n8n credential / env var so
-  rotation is a one-place edit and can be referenced by both nodes. This also
-  removes the "embedded literal in two places that must stay in sync" footgun.
+**Recommended remediation (still open):**
+- The `drc-token-rotation-alert.yml` workflow already exists (mirrors `pat-rotation-alert.yml`, 180-day max-age, Slack). The remaining gap is resilience: it alerts only via Slack, so if Slack delivery fails the pipeline has no other signal. The Hermes crons (`56f0f862d9a1` monthly + `425e85478d7b` 75-day nudge) provide a second channel, but both depend on this host being up. Consider also posting to Telegram in the workflow, or rely on the Hermes crons as the independent belt-and-suspenders.
+- Better: move the token out of node literals into an n8n credential / env var so rotation is a one-place edit and can be referenced by both nodes. This also removes the "embedded literal in two places that must stay in sync" footgun.
 
 ---
 

core/agent_config.py

@@ -139,12 +139,18 @@ def __post_init__(self) -> None:
             elif fname in flags:
                 object.__setattr__(self, fname, str(flags[fname]))
 
-    def is_action_allowed(self, action: str) -> bool:
+    def is_action_allowed(self, action: str, scope: str = "agent-core") -> bool:
         """Return True if the named action may proceed given current flags.
 
-        Checks dry_run and specific action toggles.
+        Checks (in order): global incident freeze -> dry_run -> specific action
+        toggles. The freeze is the highest-priority gate: when frozen for the
+        scope (or globally), NO write is permitted regardless of other flags.
         Actions: 'merge', 'close_issue', 'delete_branch', 'label', 'auto_fix'.
         """
+        from core.incident_freeze import freeze as _freeze
+
+        if _freeze.is_frozen(scope):
+            return False
         if self.dry_run:
             return False
         mapping = {

core/incident_freeze.py

@@ -0,0 +1,200 @@
+"""Global incident freeze — the Phase 0 "stop everything" kill switch.
+
+Per the 2026-08-04 handover (Workstream B), this is the single control that
+defaults write-capable execution to DENY while preserving ingestion, read-only
+analysis, and audit visibility. It is checked by every write path (Python
+executors, n8n write nodes/callbacks, GitHub Actions, retries, scheduled jobs,
+delayed callbacks, manual dispatches).
+
+Design rules (from the handover):
+  * Defaults to NOT frozen — normal operation is governed by dry_run / approval.
+    The freeze is an explicit operator override that escalates the posture.
+  * When frozen, ALL write actions are denied for the affected scope; read-only
+    and audit paths remain alive.
+  * Enabling/disabling records who, why, when, scope, and (for enable) expiry.
+  * Safe under repeated activation: re-enabling while already frozen is
+    idempotent (state unchanged, no error); disabling while not frozen is a
+    no-op (not an error).
+  * Does NOT silently discard work — callers must defer + retain status; this
+    module only answers "may I write?".
+  * Verifiable with a no-op test (see tests/unit/test_incident_freeze.py).
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+
+DEFAULT_STATE_PATH = Path(__file__).resolve().parent.parent / "state" / "freeze.json"
+
+# Scopes: "global" covers everything; specific scopes can be frozen independently
+# (e.g. "n8n", "github-actions", "agent-core"). "global" implies all scopes.
+GLOBAL_SCOPE = "global"
+
+
+@dataclass
+class FreezeState:
+    frozen: bool = False
+    scope: str = GLOBAL_SCOPE
+    enabled_by: str | None = None
+    enabled_at: str | None = None
+    reason: str | None = None
+    expires_at: str | None = None
+    last_disabled_by: str | None = None
+    last_disabled_at: str | None = None
+    audit: list = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        return {
+            "frozen": self.frozen,
+            "scope": self.scope,
+            "enabled_by": self.enabled_by,
+            "enabled_at": self.enabled_at,
+            "reason": self.reason,
+            "expires_at": self.expires_at,
+            "last_disabled_by": self.last_disabled_by,
+            "last_disabled_at": self.last_disabled_at,
+            "audit": self.audit,
+        }
+
+
+class FreezeControl:
+    """Operational wrapper around the persisted FreezeState."""
+
+    def __init__(self, state_path: Path = DEFAULT_STATE_PATH):
+        self.state_path = Path(state_path)
+        self._state = self._load()
+
+    # --- persistence ---
+    def _load(self) -> FreezeState:
+        if self.state_path.exists():
+            try:
+                raw = json.loads(self.state_path.read_text())
+                return FreezeState(
+                    **{k: raw.get(k, v) for k, v in FreezeState().__dict__.items()}
+                )
+            except Exception:
+                # Corrupt state file -> fail safe to NOT frozen (we never want a
+                # broken file to wedge the agent; the watchdog will flag it).
+                return FreezeState()
+        return FreezeState()
+
+    def _save(self) -> None:
+        self.state_path.parent.mkdir(parents=True, exist_ok=True)
+        # write atomically
+        tmp = self.state_path.with_suffix(".tmp")
+        tmp.write_text(json.dumps(self._state.to_dict(), indent=2))
+        tmp.replace(self.state_path)
+
+    # --- queries ---
+    def is_frozen(self, scope: str = GLOBAL_SCOPE) -> bool:
+        s = self._state
+        if not s.frozen:
+            return False
+        # expired freeze auto-clears on read
+        if s.expires_at and self._now_iso() >= s.expires_at:
+            self._clear_if_expired()
+            return False
+        if s.scope == GLOBAL_SCOPE:
+            return True
+        return scope == s.scope or scope == GLOBAL_SCOPE
+
+    def is_write_allowed(self, action: str = "any", scope: str = GLOBAL_SCOPE) -> bool:
+        """The contract every write path consults.
+
+        Returns True only if the freeze is NOT active for the scope. Dry-run and
+        approval gates are handled separately by agent_config.
+        """
+        if self.is_frozen(scope):
+            return False
+        return True
+
+    def status(self) -> dict:
+        return self._state.to_dict()
+
+    # --- mutations (operator-only) ---
+    def freeze(
+        self,
+        by: str,
+        reason: str,
+        scope: str = GLOBAL_SCOPE,
+        expires_at: str | None = None,
+    ) -> dict:
+        """Enable the freeze. Idempotent if already frozen with same scope."""
+        s = self._state
+        # If already frozen for an equal-or-broader scope, just refresh metadata
+        # without clobbering the original enable timestamp (preserve first cause).
+        if s.frozen and (
+            s.scope == GLOBAL_SCOPE or scope == s.scope or scope == GLOBAL_SCOPE
+        ):
+            entry = {
+                "event": "freeze-refreshed",
+                "by": by,
+                "at": self._now_iso(),
+                "scope": scope,
+                "reason": reason,
+            }
+            s.audit.append(entry)
+            if expires_at:
+                s.expires_at = expires_at
+            self._save()
+            return s.to_dict()
+        # New freeze (broadening scope or first enable)
+        s.frozen = True
+        s.scope = scope
+        s.enabled_by = by
+        s.enabled_at = self._now_iso()
+        s.reason = reason
+        s.expires_at = expires_at
+        s.audit.append(
+            {
+                "event": "freeze-enabled",
+                "by": by,
+                "at": s.enabled_at,
+                "scope": scope,
+                "reason": reason,
+                "expires_at": expires_at,
+            }
+        )
+        self._save()
+        return s.to_dict()
+
+    def unfreeze(self, by: str, reason: str = "operator cleared") -> dict:
+        """Disable the freeze. Idempotent if not frozen."""
+        s = self._state
+        if not s.frozen:
+            s.audit.append({"event": "unfreeze-noop", "by": by, "at": self._now_iso()})
+            self._save()
+            return s.to_dict()
+        s.frozen = False
+        s.last_disabled_by = by
+        s.last_disabled_at = self._now_iso()
+        s.audit.append(
+            {
+                "event": "freeze-disabled",
+                "by": by,
+                "at": s.last_disabled_at,
+                "reason": reason,
+            }
+        )
+        self._save()
+        return s.to_dict()
+
+    # --- helpers ---
+    def _now_iso(self) -> str:
+        return datetime.now(timezone.utc).isoformat()
+
+    def _clear_if_expired(self) -> None:
+        s = self._state
+        if s.frozen and s.expires_at and self._now_iso() >= s.expires_at:
+            s.frozen = False
+            s.audit.append({"event": "freeze-expired", "at": self._now_iso()})
+            self._save()
+
+
+# Module-level singleton — import and use directly:
+#   from core.incident_freeze import freeze
+#   if not freeze.is_write_allowed(scope="n8n"): defer()
+freeze = FreezeControl()

tests/unit/test_incident_freeze.py

@@ -0,0 +1,85 @@
+"""No-op verification tests for the global incident freeze (Phase 0 Workstream B).
+
+Run: PYTHONPATH=. /usr/bin/python3 -m pytest tests/unit/test_incident_freeze.py -q --no-cov
+"""
+
+import pytest
+
+
+@pytest.fixture
+def freeze_ctl(tmp_path, monkeypatch):
+    # Point the module at a temp state file and reload a fresh singleton
+    import core.incident_freeze as mod
+
+    mod.DEFAULT_STATE_PATH = tmp_path / "freeze.json"
+    # rebuild singleton against tmp path
+    ctl = mod.FreezeControl(state_path=tmp_path / "freeze.json")
+    monkeypatch.setattr(mod, "freeze", ctl)
+    return ctl
+
+
+def test_defaults_not_frozen(freeze_ctl):
+    assert freeze_ctl.is_frozen() is False
+    assert freeze_ctl.is_write_allowed("merge") is True
+
+
+def test_freeze_blocks_writes(freeze_ctl):
+    freeze_ctl.freeze(by="hermes", reason="drill", scope="n8n")
+    assert freeze_ctl.is_frozen("n8n") is True
+    assert freeze_ctl.is_write_allowed("merge", scope="n8n") is False
+    # read-only scope unaffected by a scoped freeze
+    assert freeze_ctl.is_write_allowed("merge", scope="github-actions") is True
+
+
+def test_global_freeze_blocks_all_scopes(freeze_ctl):
+    freeze_ctl.freeze(by="hermes", reason="incident", scope="global")
+    assert freeze_ctl.is_frozen("agent-core") is True
+    assert freeze_ctl.is_write_allowed("merge", scope="agent-core") is False
+    assert freeze_ctl.is_write_allowed("merge", scope="n8n") is False
+
+
+def test_idempotent_refreeze(freeze_ctl):
+    r1 = freeze_ctl.freeze(by="hermes", reason="first", scope="global")
+    first_at = r1["enabled_at"]
+    r2 = freeze_ctl.freeze(by="hermes", reason="dup", scope="global")
+    # enabled_at preserved (first cause retained), not overwritten
+    assert r2["enabled_at"] == first_at
+    assert r2["frozen"] is True
+
+
+def test_idempotent_unfreeze_when_not_frozen(freeze_ctl):
+    # disabling while not frozen must not raise
+    r = freeze_ctl.unfreeze(by="hermes")
+    assert r["frozen"] is False
+
+
+def test_unfreeze_restores_writes(freeze_ctl):
+    freeze_ctl.freeze(by="hermes", reason="x", scope="global")
+    assert freeze_ctl.is_write_allowed("merge") is False
+    freeze_ctl.unfreeze(by="hermes", reason="cleared")
+    assert freeze_ctl.is_write_allowed("merge") is True
+
+
+def test_expiry_auto_clears(freeze_ctl):
+    past = "2000-01-01T00:00:00+00:00"
+    freeze_ctl.freeze(by="hermes", reason="temp", scope="global", expires_at=past)
+    assert freeze_ctl.is_frozen() is False  # expired on read
+
+
+def test_audit_records_events(freeze_ctl):
+    freeze_ctl.freeze(by="op1", reason="incident", scope="global")
+    freeze_ctl.unfreeze(by="op2", reason="resolved")
+    events = [a["event"] for a in freeze_ctl.status()["audit"]]
+    assert "freeze-enabled" in events
+    assert "freeze-disabled" in events
+
+
+def test_state_persists(tmp_path):
+    import core.incident_freeze as mod
+
+    p = tmp_path / "freeze.json"
+    c1 = mod.FreezeControl(state_path=p)
+    c1.freeze(by="hermes", reason="persist", scope="global")
+    # new instance reads the persisted file
+    c2 = mod.FreezeControl(state_path=p)
+    assert c2.is_frozen() is True