fix(telegram): gate the spawn-approval prompt on the channels ceiling

docs/system-specs/modules/governance.md

@@ -1427,6 +1427,15 @@ disposition:
   (no reply), matching how an unauthorized user is ignored; `PlatformCompositionError`
   propagates. Default OSS build (no `channels` policy) permits, so inbound handling
   is byte-identical to today.
+  One OUTBOUND send consults this same inbound ceiling:
+  `messaging/spawn_approval_delivery.py` checks it before invoking any channel's
+  spawn-approval delivery hook, because the press that would answer that prompt is
+  inbound and is dropped on a denied channel (only an explicit reject is exempt on
+  a channel's callback path). A prompt posted under a deny is unanswerable, so its
+  deny-by-default wait elapses and the spawn gate reads the elapsed wait as a
+  refusal the operator never made; a deny therefore falls through and leaves the
+  spawn answerable on Slack or the dashboard instead. The check sits at that seam,
+  not inside each dispatcher, so a channel hook written without it is gated too.
   **Audit disposition:** a GOVERNED allow is audit-or-deny (`critical=True` — a SEL
   persistence failure denies the inbound, so a governed channel never receives
   unaudited); every DENY is recorded best-effort. The **ungoverned default-permit

docs/system-specs/modules/messaging.md

@@ -373,6 +373,19 @@ by channel namespace:
   contained and read as `None`: a channel-delivery bug must degrade to the existing
   Slack/dashboard fallback, never turn a spawn the operator could still answer into
   a hard failure.
+- The operator's `channels` governance ceiling (`channel_inbound_permitted`) is
+  consulted HERE, once, for every hook rather than inside each of them. Every hook
+  posts an interactive prompt whose answering press arrives INBOUND on the same
+  channel, and a denied channel drops that press (only an explicit reject is exempt
+  on a channel's callback path), so a prompt posted under a deny is unanswerable:
+  its deny-by-default wait elapses and the host gate reads the elapsed wait as a
+  refusal the operator never made. A deny therefore answers `None` and the spawn
+  stays answerable on Slack/dashboard. One check at this layer — which already
+  resolves the channel — gates every present and future hook, where a copy inside
+  each dispatcher would be the same authority duplicated per implementation. It
+  runs AFTER hook resolution, so a non-channel namespace or a `unified` DM bucket,
+  neither of which names a governed channel, falls through without asking the
+  profile store about a channel type that does not exist.
 
 The hook signature `async def(request_id, description, parent_session_key) -> bool | None`
 is the SAME three arguments the host `SpawnApprovalCallback` receives, so a channel

src/kiro_crew/messaging/spawn_approval_delivery.py

@@ -45,6 +45,7 @@
 import logging
 from typing import Awaitable, Callable
 
+from kiro_crew.messaging.identity import channel_inbound_permitted
 from kiro_crew.messaging.link import channel_namespace_of
 
 logger = logging.getLogger(__name__)
@@ -129,15 +130,41 @@ async def deliver_spawn_approval(
 
     Consulted FIRST by the host spawn-approval callback. Returns the channel's
     ``True``/``False`` decision when a hook answered, or ``None`` when no hook is
-    registered for the session's channel or the hook itself returned ``None`` (it
-    could not surface the prompt here). A hook that RAISES is contained and read as
+    registered for the session's channel, the operator's ``channels`` governance
+    ceiling denies that channel, or the hook itself returned ``None`` (it could
+    not surface the prompt here). A hook that RAISES is contained and read as
     ``None``: a channel-delivery bug must degrade to the existing fallback, never
     turn a spawn the operator could still answer on Slack/dashboard into a hard
     failure.
+
+    The ceiling is checked HERE rather than inside each hook, because every hook
+    posts an interactive prompt whose answering press arrives INBOUND on the same
+    channel, and a denied channel drops that press (only an explicit reject is
+    exempt on a channel's callback path). A prompt posted under a deny is
+    unanswerable, so its deny-by-default wait elapses and the host gate reads the
+    elapsed wait as a refusal the operator never made. One check at the routing
+    layer that already resolves the channel gates every present and future hook;
+    a copy inside each dispatcher would be the same authority duplicated per
+    implementation, and the next hook written without it reopens the hole.
+
+    It runs AFTER hook resolution, so a key in a non-channel namespace
+    (``dashboard:``, ``cron:``) or a ``unified`` DM bucket — neither of which
+    names a governed channel — falls through without asking the profile store
+    about a channel type that does not exist.
     """
     hook = resolve_channel_delivery(parent_session_key)
     if hook is None:
         return None
+    channel = channel_namespace_of(parent_session_key)
+    if not await channel_inbound_permitted(channel):
+        logger.info(
+            "Spawn-approval channel delivery skipped on %s for %s; the channel is "
+            "denied by channels governance policy, so the prompt would be "
+            "unanswerable there",
+            channel,
+            request_id,
+        )
+        return None
     try:
         return await hook(request_id, description, parent_session_key)
     except Exception:

src/kiro_crew/telegram/transport_dispatch.py

@@ -3209,7 +3209,9 @@ async def deliver_spawn_approval(
         (``True``/``False``), or ``None`` to tell the gate "not surfaced here,
         fall through to Slack/dashboard" — for a key this dispatcher cannot turn
         back into a chat (``unified`` dm_scope drops the peer, a non-``telegram``
-        key, an unparseable one) or when the client is not up.
+        key, an unparseable one) or when the client is not up. The operator's
+        ``channels`` governance ceiling is not consulted here: the seam checks it
+        before invoking any hook, so a denied channel never reaches this method.
 
         The wait is the SAME deny-by-default one a tool prompt uses
         (:class:`TelegramApprovalDecider`, ``APPROVAL_TIMEOUT_S``): the press

test/test_spawn_approval_channel_governance_13491.py

@@ -0,0 +1,188 @@
+"""The channels governance ceiling gates channel-side spawn-approval delivery.
+
+A spawn parented on a channel conversation is offered its Approve/Deny prompt
+there by that channel's registered delivery hook. The operator's `channels`
+ceiling has to be consulted before any hook is invoked, because the press that
+would answer such a prompt arrives INBOUND on the same channel, and a denied
+channel drops that press: a channel's callback path refuses everything except an
+explicit reject. A prompt posted under a deny is therefore unanswerable, its
+deny-by-default wait elapses, and the seam hands the host gate a ``False`` nobody
+pressed -- which the gate reads as the user's decision and refuses the spawn on,
+instead of re-offering it on the still-permitted Slack DM or dashboard surface.
+
+Four properties are pinned:
+
+* a denied channel invokes no hook, posts nothing, and answers ``None``
+  (fall through) rather than ``False`` (a denial nobody made);
+* the check sits at the seam, so a channel whose hook never spells it is gated
+  too, and the Telegram hook arms no approval nonce under a deny;
+* hook resolution runs first, so a key in a non-channel namespace never asks the
+  profile store about a channel type that does not exist;
+* a permitted channel is unchanged -- it still prompts and still resolves its
+  press.
+
+All Telegram client I/O is faked; nothing touches the network, and the ceiling
+predicate is substituted rather than driven through a real profile store.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+# Reuse the doubles the Telegram suite already uses (fake client, fake sessions,
+# dispatcher factory) so this suite exercises a REAL registered hook.
+from test_telegram import _dispatcher  # noqa: E402
+
+from kiro_crew.messaging import spawn_approval_delivery as seam
+from kiro_crew.telegram.renderer import TelegramApprovalDecider
+
+pytestmark = pytest.mark.usefixtures("healthy_host_memory")
+
+_RID = "spawn:abc"
+_TASK = "spawn_run(build the widget)"
+
+
+@pytest.fixture(autouse=True)
+def _clean_registries():
+    """Start and end with an empty seam registry and no armed prompts."""
+    seam.clear_channel_delivery_hooks()
+    TelegramApprovalDecider._REGISTRY.clear()
+    TelegramApprovalDecider._NONCES.clear()
+    yield
+    seam.clear_channel_delivery_hooks()
+    TelegramApprovalDecider._REGISTRY.clear()
+    TelegramApprovalDecider._NONCES.clear()
+
+
+@pytest.fixture(autouse=True)
+def _short_prompt_wait(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Shrink the deny-by-default wait so a REGRESSION fails fast.
+
+    A refusal returns before any prompt is awaited, so this does not touch the
+    path under test. It bounds the OTHER outcome: code that posts the prompt
+    under a deny would park every assertion below on the real approval timeout
+    (minutes) and surface as a suite-wide hang instead of a failed assertion.
+    """
+    import kiro_crew.telegram.renderer as renderer_mod
+
+    monkeypatch.setattr(renderer_mod, "_APPROVAL_TIMEOUT_S", 0.2)
+
+
+@pytest.fixture
+def _ceiling(monkeypatch: pytest.MonkeyPatch):
+    """Substitute the ceiling predicate, defaulting to PERMIT.
+
+    The real predicate hops to a thread pool and walks the profile store, so
+    leaving it in place would couple these tests to an executor round-trip
+    instead of to the delivery behaviour they are about. The premise is stated
+    rather than implied: each test either takes the permit default or flips it.
+    """
+
+    calls: list[str] = []
+    state = {"permitted": True}
+
+    async def _permitted(channel_type: str) -> bool:
+        calls.append(channel_type)
+        return bool(state["permitted"])
+
+    monkeypatch.setattr(seam, "channel_inbound_permitted", _permitted)
+    return SimpleNamespace(calls=calls, state=state)
+
+
+async def _press(dispatcher, session_key: str, flag: str) -> None:
+    """Press an inline button in the DM, exactly as the real callback path does."""
+    key = TelegramApprovalDecider.key(session_key, _RID)
+    for _ in range(50):
+        if key in TelegramApprovalDecider._REGISTRY:
+            break
+        await asyncio.sleep(0.01)
+    nonce = TelegramApprovalDecider._NONCES[key]
+    await dispatcher.on_callback(
+        SimpleNamespace(
+            callback_query_id="q1",
+            user_id=7,
+            chat_id=7,
+            message_id=100,
+            data=f"a:{_RID}:{nonce}:{flag}",
+            label="",
+            chat_type="private",
+        )
+    )
+
+
+class TestTheGovernanceCeilingGatesChannelDelivery:
+    """A denied channel reaches no hook, and the spawn stays answerable elsewhere."""
+
+    def test_a_denied_channel_invokes_no_hook_and_falls_through(self, _ceiling) -> None:
+        invoked: list[str] = []
+
+        async def _hook(rid: str, _desc: str, _parent: str) -> bool:
+            invoked.append(rid)
+            return True
+
+        seam.register_channel_delivery("telegram", _hook)
+        _ceiling.state["permitted"] = False
+
+        result = asyncio.run(seam.deliver_spawn_approval(_RID, _TASK, "telegram:k:direct:7"))
+
+        # None, not False: nothing was surfaced, so there is no decision to report
+        # and the gate re-offers the prompt on Slack/dashboard.
+        assert result is None
+        assert result is not False
+        assert invoked == []
+        assert _ceiling.calls == ["telegram"]
+
+    def test_the_ceiling_is_asked_about_the_channel_that_owns_the_session(self, _ceiling) -> None:
+        # The channel the prompt would be posted to is the one whose permission
+        # decides it, so the question carries that channel's own name.
+        async def _hook(_rid: str, _desc: str, _parent: str) -> bool:
+            return True
+
+        seam.register_channel_delivery("discord", _hook)
+        _ceiling.state["permitted"] = False
+
+        assert asyncio.run(seam.deliver_spawn_approval(_RID, _TASK, "discord:k:direct:9")) is None
+        assert _ceiling.calls == ["discord"]
+
+    def test_a_key_with_no_hook_never_asks_about_its_channel_type(self, _ceiling) -> None:
+        # Hook resolution runs first, so a non-channel namespace and a channel
+        # nobody registered both fall through without a governance question about
+        # a channel type the policy has no opinion on.
+        assert asyncio.run(seam.deliver_spawn_approval(_RID, _TASK, "dashboard:chat-1-2")) is None
+        assert asyncio.run(seam.deliver_spawn_approval(_RID, _TASK, "unified:kirocrew")) is None
+        assert asyncio.run(seam.deliver_spawn_approval(_RID, _TASK, "telegram:k:direct:7")) is None
+        assert _ceiling.calls == []
+
+    def test_a_denied_channel_leaves_the_telegram_hook_arming_nothing(self, _ceiling) -> None:
+        # Through the REAL dispatcher hook: a refused delivery arms no approval
+        # nonce and opens no wait, because the hook is never entered.
+        d, cli, _sess = _dispatcher({7})
+        session_key = d._session_key(("direct", "7"))
+        seam.register_channel_delivery("telegram", d.deliver_spawn_approval)
+        _ceiling.state["permitted"] = False
+
+        result = asyncio.run(seam.deliver_spawn_approval(_RID, _TASK, session_key))
+
+        assert result is None
+        assert cli.sent == []
+        key = TelegramApprovalDecider.key(session_key, _RID)
+        assert key not in TelegramApprovalDecider._NONCES
+        assert key not in TelegramApprovalDecider._REGISTRY
+
+    def test_a_permitted_channel_still_prompts_and_resolves_the_press(self, _ceiling) -> None:
+        d, cli, _sess = _dispatcher({7})
+        session_key = d._session_key(("direct", "7"))
+        seam.register_channel_delivery("telegram", d.deliver_spawn_approval)
+
+        async def _go() -> bool | None:
+            task = asyncio.ensure_future(seam.deliver_spawn_approval(_RID, _TASK, session_key))
+            await asyncio.sleep(0)
+            await _press(d, session_key, "1")
+            return await task
+
+        assert asyncio.run(_go()) is True
+        assert len(cli.sent) == 1
+        assert _ceiling.calls == ["telegram"]