diff --git a/gateway/kanban_transition_emit.py b/gateway/kanban_transition_emit.py index 8d3183ac23f5..62fe63e007c4 100644 --- a/gateway/kanban_transition_emit.py +++ b/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. diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py index 1aee25cc7f57..c1da6fbfd6d9 100644 --- a/gateway/platforms/webhook.py +++ b/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::`` 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) # ------------------------------------------------------------------ diff --git a/tests/gateway/test_wake_origin_on_transition.py b/tests/gateway/test_wake_origin_on_transition.py new file mode 100644 index 000000000000..55aaf9167417 --- /dev/null +++ b/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::`` 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 diff --git a/tests/gateway/test_webhook_origin_routing.py b/tests/gateway/test_webhook_origin_routing.py new file mode 100644 index 000000000000..39bd542c94ae --- /dev/null +++ b/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::`` 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:: + 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