fix(agent-team): supervise the Slack listener thread — recurring ALARM + respawn

sh-security-review (logic) MEDIUM: a crashed listener thread was logged once,
then the daemon ran on 'deaf' — posting clarifier questions but receiving no
answers, every gate silently parking, process never exiting so systemd
Restart=on-failure never fired. serve() now calls _supervise_slack_listener()
each pass: when the listener is enabled but its thread is dead, it emits a
recurring ERROR ALARM and respawns via the idempotent starter (self-heal).
No-op when alive or disabled. +3 tests. (authz detector: wiring clean — AUTHZ-01
fail-closed allowlist + open-status CAS intact, dead listener fails SAFE.)

agent-team/agent_team/coordinator.py

@@ -904,6 +904,7 @@ class Coordinator:
         try:
             while True:  # pragma: no cover - the infinite daemon loop
                 self.tick()
+                self._supervise_slack_listener()
                 time.sleep(interval)
         finally:
             self._stop_slack_listener()
@@ -978,6 +979,31 @@ class Coordinator:
         _LOG.info("inbound Slack listener started (Socket Mode, background thread)")
         return True
 
+    def _supervise_slack_listener(self) -> None:
+        """Detect a dead inbound listener thread, ALARM (recurring), and respawn.
+
+        Called every :meth:`serve` pass. Without this, a crashed listener thread
+        (:meth:`_run_listener` swallows its exception) leaves the daemon "deaf":
+        it keeps posting clarifier questions but receives no answers, every gate
+        silently times out to PARK, and the process stays up so systemd
+        ``Restart=on-failure`` never fires. This makes that failure LOUD on every
+        pass (not a one-shot crash log) and self-heals via the idempotent
+        :meth:`_maybe_start_slack_listener`. No-op when the listener is not
+        enabled or the thread is alive.
+        """
+        if not self._slack_listener_enabled():
+            return
+        thread = self._listener_thread
+        if thread is not None and thread.is_alive():
+            return
+        _LOG.error(
+            "ALARM: inbound Slack listener is DOWN — answers are NOT being "
+            "received; clarifier gates will park. Respawning the listener."
+        )
+        # Drop the dead handle so the idempotent starter actually respawns.
+        self._listener_thread = None
+        self._maybe_start_slack_listener()
+
     def _run_listener(self) -> None:
         """Thread target: run the listener's blocking serve, log on exit.
 

agent-team/tests/test_coordinator.py

@@ -738,6 +738,70 @@ def test_maybe_start_listener_is_idempotent(
         coord._stop_slack_listener()
 
 
+def test_supervise_respawns_dead_listener_and_alarms(
+    db_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A dead listener thread is respawned (self-heal) with a recurring ALARM, so
+    the daemon never silently goes 'deaf' to Slack answers (D1-silent-stall)."""
+    import logging
+
+    monkeypatch.setenv("SLACK_APP_TOKEN", "xapp-test")
+    listeners: list[_FakeListener] = []
+
+    def _factory() -> _FakeListener:
+        lst = _FakeListener()
+        listeners.append(lst)
+        return lst
+
+    coord = _slack_coordinator(db_path, build_listener=_factory)
+    coord.setup()
+    try:
+        assert coord._maybe_start_slack_listener() is True
+        first = coord._listener_thread
+        assert first is not None and first.is_alive()
+        # Kill the listener thread: close() fires the stop event so serve() returns.
+        listeners[0].close()
+        first.join(timeout=5.0)
+        assert not first.is_alive()
+        # The supervisor detects the death, ALARMs, and respawns a fresh listener.
+        with caplog.at_level(logging.ERROR):
+            coord._supervise_slack_listener()
+        assert any("listener is DOWN" in r.message for r in caplog.records)
+        second = coord._listener_thread
+        assert second is not None and second is not first and second.is_alive()
+        assert len(listeners) == 2  # a new listener was built on respawn
+    finally:
+        coord._stop_slack_listener()
+
+
+def test_supervise_is_noop_when_listener_alive(
+    db_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The supervisor leaves a healthy listener thread untouched (no churn)."""
+    monkeypatch.setenv("SLACK_APP_TOKEN", "xapp-test")
+    listener = _FakeListener()
+    coord = _slack_coordinator(db_path, build_listener=lambda: listener)
+    coord.setup()
+    try:
+        coord._maybe_start_slack_listener()
+        alive = coord._listener_thread
+        coord._supervise_slack_listener()
+        assert coord._listener_thread is alive
+    finally:
+        coord._stop_slack_listener()
+
+
+def test_supervise_is_noop_when_listener_disabled(
+    db_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """No app token -> listener disabled -> supervisor never spawns a thread."""
+    monkeypatch.delenv("SLACK_APP_TOKEN", raising=False)
+    coord = _slack_coordinator(db_path, build_listener=_FakeListener)
+    coord.setup()
+    coord._supervise_slack_listener()
+    assert coord._listener_thread is None
+
+
 def test_stop_listener_is_safe_when_none(db_path: Path) -> None:
     """Stopping with no listener running never raises (clean-shutdown safety)."""
     coord = _slack_coordinator(db_path)