From 18d1a5ce49c1b245ecc7dbfbf698d81d58503b6f Mon Sep 17 00:00:00 2001 From: Casey West Date: Wed, 1 Jul 2026 14:36:22 -0400 Subject: [PATCH 1/2] =?UTF-8?q?=E2=9C=A8=20feat(kanban):=20wake=20origin?= =?UTF-8?q?=20thread=20session=20on=20every=20transition?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transition-emit bridge stamped origin_* fields on its payload, but the webhook route ignored them and always minted a contextless webhook:: session — so a woken orchestrator never resumed the thread the work was born in. And the emit gate covered only blocked/ block_loop_detected, so a card moving to review (status_changed + assigned) woke nothing at all. - webhook route: when a payload carries origin_platform/chat_id/thread_id, build a SessionSource targeting that origin thread instead of the synthetic webhook session, so the run resumes the origin session and reports back there. Falls back to the webhook session when origin fields are absent (backward-compatible). The origin source is shaped to mirror the live inbound session key (chat_type=thread, chat_id+thread_id set, no per-user suffix) so the wake lands in the exact origin session, not a phantom one. - widen DEFAULT_EMIT_KINDS to include the actionable lane-move kinds (status_changed, assigned, unblocked) alongside blocked/block_loop_detected; completed stays out (done is a close-loop ping, not a reasoning task). - tests: origin routing + key-mirror invariant + emit-kind coverage; RED before, GREEN after; full gateway suite green. --- gateway/kanban_transition_emit.py | 15 ++- gateway/platforms/webhook.py | 104 +++++++++++---- .../gateway/test_wake_origin_on_transition.py | 62 +++++++++ tests/gateway/test_webhook_origin_routing.py | 123 ++++++++++++++++++ 4 files changed, 281 insertions(+), 23 deletions(-) create mode 100644 tests/gateway/test_wake_origin_on_transition.py create mode 100644 tests/gateway/test_webhook_origin_routing.py diff --git a/gateway/kanban_transition_emit.py b/gateway/kanban_transition_emit.py index 8d3183ac23f5..a13e1d5fd7a8 100644 --- a/gateway/kanban_transition_emit.py +++ b/gateway/kanban_transition_emit.py @@ -48,7 +48,20 @@ # 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", + # 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..f13d0a979096 --- /dev/null +++ b/tests/gateway/test_wake_origin_on_transition.py @@ -0,0 +1,62 @@ +"""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_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 From 76291be89f1739e73937261d357888b1d4df966d Mon Sep 17 00:00:00 2001 From: Casey West Date: Wed, 1 Jul 2026 14:57:12 -0400 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=90=9B=20fix(kanban):=20wake=20orches?= =?UTF-8?q?trator=20on=20terminal-failure=20transitions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DEFAULT_EMIT_KINDS omitted the terminal-failure kinds gave_up, crashed, and timed_out. Those flow through the same gate as the chat-ping notifier (should_emit_transition is invoked inside the loop over NOTIFY_KINDS, whose TERMINAL_KINDS = completed, blocked, gave_up, crashed, timed_out), so a worker that gave up, crashed, or timed out pinged chat but woke no orchestrator — the same silent-escalation gap this change set closes, left open on the failure path. Add the three kinds to the default emit set so the failure path wakes the orchestrator exactly like blocked/block_loop_detected do. completed stays out by design (done is a close-loop chat ping, not a reasoning task). Add a default-coverage test so the failure-path wake cannot silently regress. --- gateway/kanban_transition_emit.py | 9 +++++++++ tests/gateway/test_wake_origin_on_transition.py | 11 +++++++++++ 2 files changed, 20 insertions(+) diff --git a/gateway/kanban_transition_emit.py b/gateway/kanban_transition_emit.py index a13e1d5fd7a8..62fe63e007c4 100644 --- a/gateway/kanban_transition_emit.py +++ b/gateway/kanban_transition_emit.py @@ -52,6 +52,15 @@ # 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 diff --git a/tests/gateway/test_wake_origin_on_transition.py b/tests/gateway/test_wake_origin_on_transition.py index f13d0a979096..55aaf9167417 100644 --- a/tests/gateway/test_wake_origin_on_transition.py +++ b/tests/gateway/test_wake_origin_on_transition.py @@ -50,6 +50,17 @@ def test_should_emit_still_covers_blocked_and_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"])