✨ feat(kanban): wake origin thread session on every transition

gateway/kanban_transition_emit.py

@@ -48,7 +48,29 @@
 # decision), and it used to fire NOTHING at all. `completed`/done is deliberately
 # NOT here — done is a close-loop chat ping, not a reasoning task. Override via
 # config `kanban.transition_emit.emit_kinds`.
-DEFAULT_EMIT_KINDS: tuple[str, ...] = ("blocked", "block_loop_detected")
+DEFAULT_EMIT_KINDS: tuple[str, ...] = (
+    # Terminal / escalation kinds (always woke the orchestrator).
+    "blocked",
+    "block_loop_detected",
+    # Terminal-FAILURE escalation kinds. A worker that gives up, crashes, or
+    # times out is the failure-path twin of ``blocked``: the notifier's
+    # NOTIFY_KINDS pings chat for these (via TERMINAL_KINDS), so the wake must
+    # cover them too or the orchestrator is woken on a clean handoff but stays
+    # asleep when a worker actually fails — the same silent-escalation gap this
+    # fix closes, on the failure side.
+    "gave_up",
+    "crashed",
+    "timed_out",
+    # Actionable lane-move kinds. A card moving to review fires
+    # ``status_changed`` + ``assigned``; ``unblocked`` returns a card to the
+    # ready lane. These are handoffs the orchestrator must act on — they were
+    # silent pre-fix (live miss: card t_278408f5 review handoff woke nothing).
+    # ``completed``/done stays OUT deliberately: done is a close-loop chat
+    # ping, not a reasoning task.
+    "status_changed",
+    "assigned",
+    "unblocked",
+)
 
 DEFAULT_ROUTE = "kanban-transition"
 # The webhook adapter binds loopback by default; the bridge POSTs to itself.

gateway/platforms/webhook.py

@@ -48,6 +48,7 @@
     web = None  # type: ignore[assignment]
 
 from gateway.config import Platform, PlatformConfig
+from gateway.session import SessionSource
 from gateway.platforms.base import (
     BasePlatformAdapter,
     MessageEvent,
@@ -716,29 +717,46 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         # same route get independent agent runs (not queued/interrupted).
         session_chat_id = f"webhook:{route_name}:{delivery_id}"
 
-        # Store delivery info for send().  Read by every send() invocation
-        # for this chat_id (interim status messages and the final response),
-        # so we do NOT pop on send.  TTL-based cleanup keeps the dict bounded.
-        deliver_config = {
-            "deliver": route_config.get("deliver", "log"),
-            "deliver_extra": self._render_delivery_extra(
-                route_config.get("deliver_extra", {}), payload
-            ),
-            "payload": payload,
-        }
-        self._delivery_info[session_chat_id] = deliver_config
-        self._delivery_info_created[session_chat_id] = now
-        self._delivery_info_order.append((now, session_chat_id))
-        self._prune_delivery_info(now)
-
-        # Build source and event
-        source = self.build_source(
-            chat_id=session_chat_id,
-            chat_name=f"webhook/{route_name}",
-            chat_type="webhook",
-            user_id=f"webhook:{route_name}",
-            user_name=route_name,
+        # ── Origin routing (event-driven autonomy) ──────────────────
+        # When the payload carries origin_* fields (the kanban-transition
+        # emitter stamps them), route the woken run into the ORIGIN thread
+        # session it was born in — not a contextless webhook: session — so
+        # the orchestrator resumes in that thread and can act + continue
+        # there. Without origin fields, fall back to the webhook: session
+        # (backward-compatible: non-origin webhooks are unaffected).
+        origin_platform = payload.get("origin_platform")
+        origin_chat_id = payload.get("origin_chat_id")
+        origin_thread_id = payload.get("origin_thread_id")
+        origin_source = self._build_origin_source(
+            origin_platform, origin_chat_id, origin_thread_id
         )
+
+        if origin_source is not None:
+            source = origin_source
+        else:
+            # Store delivery info for send().  Read by every send() invocation
+            # for this chat_id (interim status messages and the final response),
+            # so we do NOT pop on send.  TTL-based cleanup keeps the dict bounded.
+            deliver_config = {
+                "deliver": route_config.get("deliver", "log"),
+                "deliver_extra": self._render_delivery_extra(
+                    route_config.get("deliver_extra", {}), payload
+                ),
+                "payload": payload,
+            }
+            self._delivery_info[session_chat_id] = deliver_config
+            self._delivery_info_created[session_chat_id] = now
+            self._delivery_info_order.append((now, session_chat_id))
+            self._prune_delivery_info(now)
+
+            # Build source and event
+            source = self.build_source(
+                chat_id=session_chat_id,
+                chat_name=f"webhook/{route_name}",
+                chat_type="webhook",
+                user_id=f"webhook:{route_name}",
+                user_name=route_name,
+            )
         if profile and isinstance(profile, str):
             source.profile = profile
         event = MessageEvent(
@@ -773,6 +791,48 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
             status=202,
         )
 
+    def _build_origin_source(
+        self,
+        origin_platform: Optional[str],
+        origin_chat_id: Optional[str],
+        origin_thread_id: Optional[str],
+    ) -> Optional[SessionSource]:
+        """Build a SessionSource targeting the ORIGIN thread/session, or None.
+
+        When a transition-emit payload carries the origin thread it was born in,
+        the woken run must resume THAT session — not a contextless
+        ``webhook:<route>:<delivery>`` one — so the orchestrator reports back
+        and continues in the origin thread (the event-driven autonomy contract).
+
+        Returns ``None`` when origin fields are absent/unusable, so the caller
+        falls back to the default webhook session (backward-compatible).
+
+        The source is shaped to mirror the live inbound key
+        (``build_session_key``): for a thread the platform delivers
+        ``chat_type="thread"`` with ``chat_id``/``thread_id`` both set, and the
+        key omits the user id (threads are shared). Matching that shape here is
+        what makes the wake land in the exact origin session.
+        """
+        if not origin_platform or not origin_chat_id:
+            return None
+        try:
+            platform = Platform(str(origin_platform))
+        except ValueError:
+            logger.warning(
+                "[webhook] origin routing: unknown platform %r; "
+                "falling back to webhook session",
+                origin_platform,
+            )
+            return None
+        chat_type = "thread" if origin_thread_id else "channel"
+        return SessionSource(
+            platform=platform,
+            chat_id=str(origin_chat_id),
+            chat_name="kanban-transition/origin",
+            chat_type=chat_type,
+            thread_id=str(origin_thread_id) if origin_thread_id else None,
+        )
+
     # ------------------------------------------------------------------
     # Author resolution (for the author allow-list security gate)
     # ------------------------------------------------------------------

tests/gateway/test_wake_origin_on_transition.py

@@ -0,0 +1,73 @@
+"""Tests for the wake-origin-on-transition fix.
+
+Two defects proven from a live E2E test (card t_278408f5, 2026-07-01):
+
+1. ``should_emit_transition`` / ``DEFAULT_EMIT_KINDS`` did not cover the
+   actionable lane-move kinds (``status_changed``, ``assigned``, ``unblocked``),
+   so a card moving to ``review`` fired no wake at all.
+
+2. The webhook route ignored the ``origin_*`` fields the emitter sends, so every
+   wake ran in a contextless ``webhook:<route>:<delivery>`` session instead of
+   the origin thread session it was born in.
+
+These tests assert the intended behavior and are RED against the pre-fix code.
+"""
+
+from __future__ import annotations
+
+import gateway.kanban_transition_emit as kte
+
+
+# ---------------------------------------------------------------------------
+# Defect 1 — the wake gate must cover actionable lane-move kinds
+# ---------------------------------------------------------------------------
+
+def _enabled_cfg(**over):
+    cfg = {"enabled": True}
+    cfg.update(over)
+    return cfg
+
+
+def test_should_emit_covers_status_changed_by_default():
+    # A card moving to review fires ``status_changed`` — it MUST wake the
+    # orchestrator so it can act on the handoff. Was silent pre-fix.
+    assert kte.should_emit_transition(_enabled_cfg(), "status_changed") is True
+
+
+def test_should_emit_covers_assigned_by_default():
+    assert kte.should_emit_transition(_enabled_cfg(), "assigned") is True
+
+
+def test_should_emit_covers_unblocked_by_default():
+    assert kte.should_emit_transition(_enabled_cfg(), "unblocked") is True
+
+
+def test_should_emit_still_covers_blocked_and_loop_detected():
+    # No regression to the two kinds that already worked.
+    assert kte.should_emit_transition(_enabled_cfg(), "blocked") is True
+    assert kte.should_emit_transition(
+        _enabled_cfg(), "block_loop_detected"
+    ) is True
+
+
+def test_should_emit_covers_terminal_failure_kinds_by_default():
+    # A worker that gives up / crashes / times out is a terminal-failure
+    # escalation: the notifier's NOTIFY_KINDS pings chat for these (via
+    # TERMINAL_KINDS), but the wake must ALSO fire so the orchestrator can act
+    # on the failure path. These were silent pre-fix — a crashed worker pinged
+    # chat but woke no orchestrator, the same silent-escalation gap this fix
+    # closes, on the failure side.
+    for kind in ("gave_up", "crashed", "timed_out"):
+        assert kte.should_emit_transition(_enabled_cfg(), kind) is True, kind
+
+
+def test_should_emit_respects_explicit_config_override():
+    # An explicit emit_kinds list still wins over the default set.
+    cfg = _enabled_cfg(emit_kinds=["blocked"])
+    assert kte.should_emit_transition(cfg, "blocked") is True
+    assert kte.should_emit_transition(cfg, "status_changed") is False
+
+
+def test_should_emit_disabled_when_not_enabled():
+    assert kte.should_emit_transition({"enabled": False}, "status_changed") is False
+    assert kte.should_emit_transition(None, "status_changed") is False

tests/gateway/test_webhook_origin_routing.py

@@ -0,0 +1,123 @@
+"""Tests: the kanban-transition wake routes to the ORIGIN thread session.
+
+Defect 2 (proven live, card t_278408f5): the webhook route ignored the
+``origin_*`` fields the emitter sends, so every transition wake ran in a
+contextless ``webhook:<route>:<delivery>`` session and never reached the origin
+Discord thread. These assert the route now honors origin routing, with a clean
+fallback when origin fields are absent (no regression to non-origin webhooks).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+from aiohttp.test_utils import TestClient, TestServer
+from aiohttp import web
+
+from gateway.platforms.webhook import WebhookAdapter, _INSECURE_NO_AUTH
+from gateway.config import PlatformConfig
+
+
+def _make_adapter(routes):
+    extra = {"host": "0.0.0.0", "port": 0, "routes": routes,
+             "rate_limit": 100, "max_body_bytes": 1_048_576}
+    return WebhookAdapter(PlatformConfig(enabled=True, extra=extra))
+
+
+def _app(adapter):
+    app = web.Application()
+    app.router.add_post("/webhooks/{route_name}", adapter._handle_webhook)
+    return app
+
+
+def _kt_route():
+    return {"kanban-transition": {"secret": _INSECURE_NO_AUTH, "prompt": "{title}"}}
+
+
+@pytest.mark.asyncio
+async def test_wake_targets_origin_thread_when_origin_fields_present():
+    """A POST carrying origin_* fields builds a source targeting the origin
+    thread/session — NOT the contextless webhook: session."""
+    adapter = _make_adapter(_kt_route())
+    captured = {}
+
+    async def _capture(event):
+        captured["event"] = event
+
+    adapter.handle_message = AsyncMock(side_effect=_capture)
+
+    async with TestClient(TestServer(_app(adapter))) as cli:
+        resp = await cli.post(
+            "/webhooks/kanban-transition",
+            json={
+                "event_type": "status_changed",
+                "task_id": "t_abc",
+                "title": "probe card",
+                "origin_session_id": "20260629_125345_f5fe11ba",
+                "origin_platform": "discord",
+                "origin_chat_id": "1520255822704152666",
+                "origin_thread_id": "1520255822704152666",
+            },
+        )
+        assert resp.status == 202
+
+    adapter.handle_message.assert_awaited_once()
+    src = captured["event"].source
+    # The wake must be routed to the ORIGIN, not a webhook: synthetic session.
+    assert src.platform.value == "discord"
+    assert str(src.chat_id) == "1520255822704152666"
+    assert str(getattr(src, "thread_id", "")) == "1520255822704152666"
+    assert not str(src.chat_id).startswith("webhook:")
+
+
+@pytest.mark.asyncio
+async def test_wake_falls_back_to_webhook_session_without_origin_fields():
+    """No origin fields → current behavior: a webhook:<route>:<delivery>
+    session. Backward-compatible; non-origin webhooks are unaffected."""
+    adapter = _make_adapter(_kt_route())
+    captured = {}
+
+    async def _capture(event):
+        captured["event"] = event
+
+    adapter.handle_message = AsyncMock(side_effect=_capture)
+
+    async with TestClient(TestServer(_app(adapter))) as cli:
+        resp = await cli.post(
+            "/webhooks/kanban-transition",
+            json={"event_type": "status_changed", "task_id": "t_abc",
+                  "title": "no-origin card"},
+        )
+        assert resp.status == 202
+
+    adapter.handle_message.assert_awaited_once()
+    src = captured["event"].source
+    assert str(src.chat_id).startswith("webhook:kanban-transition:")
+
+
+def test_origin_source_yields_the_live_thread_session_key():
+    """The origin source MUST produce the exact session key the live Discord
+    thread inbound produces — otherwise the wake lands in a phantom session
+    (the F2 key-mismatch trap). Locks the key-mirror invariant."""
+    from gateway.session import build_session_key
+    adapter = _make_adapter(_kt_route())
+    src = adapter._build_origin_source(
+        "discord", "1520255822704152666", "1520255822704152666"
+    )
+    assert src is not None
+    key = build_session_key(
+        src, group_sessions_per_user=True, thread_sessions_per_user=False
+    )
+    assert key == "agent:main:discord:thread:1520255822704152666:1520255822704152666"
+
+
+def test_origin_source_none_when_fields_absent():
+    adapter = _make_adapter(_kt_route())
+    assert adapter._build_origin_source(None, None, None) is None
+    assert adapter._build_origin_source("discord", None, None) is None
+
+
+def test_origin_source_unknown_platform_falls_back():
+    adapter = _make_adapter(_kt_route())
+    assert adapter._build_origin_source("nope", "123", "123") is None