From bc9dbcbca5ca85c7561bfb21e1e01082c19e4b0c Mon Sep 17 00:00:00 2001 From: Casey West Date: Wed, 1 Jul 2026 18:05:25 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(kanban):=20resume=20orchestr?= =?UTF-8?q?ator=20as=20a=20distinct=20turn=20on=20a=20busy-session=20wake?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An origin-routed kanban-transition wake arrives at the platform adapter as a plain TEXT MessageEvent. When the origin session is mid-turn, handle_message takes the busy branch and either text-debounces the event or newline-merges it into the in-progress/queued turn via merge_pending_message_event(merge_text=True) — so the wake is absorbed silently instead of producing its own identifiable turn (the busy-session swallow). Tag the origin-routed wake on the webhook side (metadata["kanban_transition_wake"] = True), keyed on the emitter-stamped origin_* fields so ordinary webhook routes are unaffected. A pure predicate is_transition_wake_event reads the flag; a wake-precedence branch in merge_pending_message_event keeps the wake in the single pending slot un-merged and un-clobbered (a pending wake is never overwritten by a later non-wake, and an incoming wake replaces a pending non-wake intact — a dropped autonomy wake is worse than a user follow-up the user can resend); and the busy branch queues the wake without debounce/merge so the existing in-band drain cascade runs it as a distinct turn. Cache- and alternation-safe: no system-prompt or message-history mutation, no new drain site, no synthetic mid-loop user message, _active_sessions lifecycle untouched — the wake is delivered as an ordinary next-turn user message via the same cascade every follow-up uses. --- PATCHES.md | 1 + ...tion-wake-distinct-self-announcing-turn.md | 71 +++++++++ gateway/platforms/base.py | 53 +++++++ gateway/platforms/webhook.py | 25 +++ .../test_transition_wake_busy_session.py | 143 ++++++++++++++++++ tests/gateway/test_webhook_origin_routing.py | 30 ++++ 6 files changed, 323 insertions(+) create mode 100644 docs/plans/2026-07-01-transition-wake-distinct-self-announcing-turn.md create mode 100644 tests/gateway/test_transition_wake_busy_session.py diff --git a/PATCHES.md b/PATCHES.md index 63f4de4f368a..ff030137a6f4 100644 --- a/PATCHES.md +++ b/PATCHES.md @@ -80,3 +80,4 @@ default; do **not** retire such a row on a PR-merge signal. See the #44338 row. | fork PR (TBD — no upstream PR; fork-internal review-lane semantics) | Auto-route a reviewer's `review-changes-requested` block back to the original author from the housekeeping tick, closing the reviewer→author hop the GitHub `pull_request_review` webhook cannot close when reviewer and author share one GitHub identity. The reviewer (lamport) runs as the `cwest` identity that OWNS the team PRs, so GitHub rejects his `REQUEST_CHANGES` with HTTP 422 and he falls back to a `COMMENT` event — which is NOT `changes_requested`, so the webhook router never bounces and the card sits `blocked` until a human hand-routes it. The fix is board-internal and lives entirely in the dispatcher, NOT the reviewer (the reviewer's terminal action stays a clean `kanban_block`, preserving the lane-corruption-safe design): `auto_route_review_bounce` scans `blocked` cards on each `dispatch_once` tick (before `recompute_ready`), and for any card whose most-recent sticky `blocked` event carries the `review-changes-requested` reason prefix, reassigns it to the original author (resolved from the `assigned` event history — the `from` of the move whose `to` is the current reviewer — never a literal profile name) and unblocks it via the existing `unblock_task`, which clears the `active_pr`/`recent_success` respawn guards exactly like a manual block→unblock cutoff, plus a `dispatcher`-authored `[audit]` comment naming the PR and verdict gist. Idempotent (the route flips the card off `blocked`, so a later tick won't re-fire; two ticks → one route, one comment). The `awaiting-casey-signoff` PASS/acceptance block is excluded by the prefix match (must stay `blocked`+casey), as are non-review and circuit-breaker (`gave_up`) blocks. Two `check_respawn_guard` carve-outs support the route: the `recent_success` guard now honors a trailing `unblocked` event (the bounced build run is the work being reworked — it must not veto respawn), and the dup-PR scan excludes the dispatcher's own same-second audit comment. Toggle `kanban.auto_route_review_bounce` (default ON) gates the whole path; wired through `dispatch_once`, the `hermes kanban dispatch` CLI, and the gateway dispatcher watcher. Changes in `hermes_cli/kanban_db.py`, `hermes_cli/kanban.py`, `gateway/kanban_watchers.py` + 8 regression tests in `tests/hermes_cli/test_kanban_auto_route_review_bounce.py` (core route; route via real `dispatch_once`; acceptance block does NOT route; idempotency; routed card dispatchable; non-review block; circuit-breaker block; toggle off). Upstream has no equivalent (no `review-changes-requested`/`auto_route`/`bounce` concept in upstream `hermes_cli/*.py`) — cwest-team review-loop tooling. **Retire trigger:** never auto-retires (permanent-local); remove only if the fork's review-lane dispatch is retired. | permanent-local | main@9be292f1e | | fork PR (TBD — upstream PR not yet opened) | Never spawn a tool-less kanban worker, and bound the per-tick spawn burst. A dispatcher-spawned worker came up with ONLY the base `kanban_*` coordination tools (no web/shell/git/file) despite its profile declaring a full toolset, then self-blocked; root cause was a stuck→mass-spawn recovery tick that launched all ready cards at once, under which `_resolve_worker_cli_toolsets` came up degenerate (None/empty) and `_default_spawn` silently launched the worker WITHOUT a `--toolsets` pin (the `if worker_toolsets:` guard let it fall back to a kanban-only surface). Two layers, smallest-footprint first, covering the whole class via the single `_default_spawn` helper (shared by the ready and review dispatch paths): (1) `_default_spawn` now REQUIRES a non-empty resolved CLI toolset and raises `RuntimeError` when resolution is degenerate, so `dispatch_once`'s existing spawn-failure handler records the failure with `release_claim=True` and the card is reclaimed to `ready` for a clean retry instead of running crippled (`_resolve_worker_cli_toolsets` always recovers at least the kanban lifecycle surface for a real profile home, so None/empty is a genuine failure, not a legitimately tool-less profile); (2) a new `kanban.max_spawn_per_tick` config knob caps how many workers a single tick may launch (ready + review combined), distinct from `max_spawn` (a live concurrency cap), wired through the gateway dispatcher (`gateway/kanban_watchers.py`) and the CLI dispatch path (`hermes_cli/kanban.py`); unset (None) preserves historical unbounded behavior, invalid/<1 values normalize to None. NO new user-facing env var: the knob lives in `config.yaml` under `kanban.max_spawn_per_tick` (`hermes_cli/config.py` DEFAULT_CONFIG), per the `.env`-is-secrets-only rule. Behavior-contract tests against the real `_default_spawn`/`dispatch_once` with a temp `HERMES_HOME` (not mocks of the unit under test): a spawned worker's resolved toolset is pinned; a degenerate (None) and an empty-list resolution both raise and never `Popen`; N>cap ready cards spawn at most cap per tick; the per-tick cap also counts review spawns; and the gateway forwards `kanban.max_spawn_per_tick` to `dispatch_once`. Recovered from stranded fork PR #10 (origin commit `8fb892f1c`, authored Casey West) by cherry-pick onto current integration; the spawn-env block reconciled cleanly on the current base. **Retire trigger:** open the upstream PR, then auto-retire when it merges in a tagged release ≥ base; until then carry as upstream-pending. | upstream-pending | main@9be292f1e | | fork PR (TBD — no upstream PR; fork-internal event-driven orchestration) | Optional, default-OFF bridge that lets a kanban lifecycle transition wake the orchestrator as an agent RUN (not merely a chat ping), by POSTing the transition to a loopback webhook route — mirroring how a GitHub `pull_request` event triggers a review run. The native notifier loop (`GatewayKanbanWatchersMixin._kanban_notifier_watcher`) already delivers a terminal kanban event as a *message* to subscribed sources (ping a chat when a card reaches acceptance/done); it does NOT trigger reasoning. New module `gateway/kanban_transition_emit.py` adds pure, unit-testable decision logic (`should_emit_transition`, `build_transition_payload` with a stable `(board,task_id,kind,event_id)` idempotency key) plus a tiny fail-safe coroutine `emit_transition` that GitHub-style HMAC-signs (`X-Hub-Signature-256: sha256=`, matching the webhook adapter's validator) and POSTs to `http://127.0.0.1:8644/webhooks/kanban-transition`; it NEVER raises (a bridge failure must not break the notifier tick — the chat ping already happened). `gateway/kanban_watchers.py` gains a minimal additive call AFTER the existing chat-ping delivery, guarded by `kanban.transition_emit.enabled` (default OFF) and a bridged internal secret `HERMES_KANBAN_TRANSITION_SECRET` (read at gateway cold boot): when disabled the notifier path is byte-for-byte unchanged. NO new core tool, NO new model surface (reuses the existing webhook adapter + a route + a skill), NO user-facing `HERMES_*` config var (the flag lives in `config.yaml` under `kanban.transition_emit`; the env var is a secret bridge only). The route + its `kanban-transition-orchestrate` skill + the `config.yaml` flag live in `cwest/hermes-config` (the homestead config repo); this fork carries only the gateway-side emitter + the flag read. Tests: `tests/gateway/test_kanban_transition_emit.py` (decision logic, payload/idempotency, HMAC signature, fail-safe POST) + `tests/gateway/test_kanban_transition_emit_http.py` (real aiohttp loopback round-trip, 2xx/non-2xx, signature header). Restart-gated (the flag is read at cold boot). Cache-/alternation-safe (no system-prompt or message-history mutation; the agent run is a fresh scoped session, exactly like the github-prs route). **Retire trigger:** never auto-retires on an upstream PR-merge signal (no upstream PR — fork-internal orchestration machinery); remove only if the homestead event-driven orchestration model is retired. | permanent-local | main@9be292f1e | +| fork PR (TBD — no upstream PR; fork-internal event-driven orchestration) | Make a kanban-transition WAKE resume the orchestrator as a DISTINCT turn even when the origin session is mid-turn (companion to the transition-emit bridge + origin-routing rows above). Root cause (verified live 2026-07-01): the origin-routed wake arrives at the adapter as a plain TEXT `MessageEvent`, so when the origin session is active `handle_message` (`gateway/platforms/base.py`) drops it into the busy branch, where it is either text-debounced (`_queue_text_debounce`) or newline-merged into the in-progress/queued turn via `merge_pending_message_event(merge_text=True)` — the wake is absorbed silently instead of producing its own identifiable turn (the busy-session swallow). Fix: the webhook route detects the wake by the emitter-stamped `origin_*` fields (`_is_transition_wake_payload`) and tags the inbound `MessageEvent` with `metadata["kanban_transition_wake"]=True`. A pure predicate `is_transition_wake_event` reads that flag; a wake-precedence branch in `merge_pending_message_event` keeps the wake in the single pending slot UN-merged and UN-clobbered (a pending wake is never overwritten by a later non-wake, and an incoming wake replaces a pending non-wake intact — a dropped autonomy wake is worse than a user follow-up the user can resend), and the busy branch in `handle_message` queues the wake without debounce/merge so the existing in-band drain cascade runs it as a DISTINCT turn. Cache-/alternation-safe: no system-prompt or message-history mutation, no new drain site, no synthetic mid-loop user message, `_active_sessions` lifecycle untouched — the wake is delivered as an ordinary next-turn user message via the same cascade every follow-up uses. SCOPE: distinct-turn delivery only; the self-announcing wake banner and latency tightening are separate follow-up work. Files: `gateway/platforms/base.py` (`is_transition_wake_event`, `merge_pending_message_event` wake-precedence, busy-branch bypass), `gateway/platforms/webhook.py` (`_is_transition_wake_payload` + wake tagging) + tests (`tests/gateway/test_transition_wake_busy_session.py`, one added case in `tests/gateway/test_webhook_origin_routing.py`). **Retire trigger:** never auto-retires on an upstream PR-merge signal (no upstream PR — fork-internal orchestration machinery); remove only if the homestead event-driven orchestration model is retired. | permanent-local | main@9be292f1e | diff --git a/docs/plans/2026-07-01-transition-wake-distinct-self-announcing-turn.md b/docs/plans/2026-07-01-transition-wake-distinct-self-announcing-turn.md new file mode 100644 index 000000000000..a0ca22eb46be --- /dev/null +++ b/docs/plans/2026-07-01-transition-wake-distinct-self-announcing-turn.md @@ -0,0 +1,71 @@ +# Transition wake: a distinct turn even when the origin session is busy + +Card: t_376048fd. Base: cwest/integration @ a2705180d (#28). + +SCOPE: this card is ONLY the distinct-turn delivery (the busy-session swallow). +The self-announcing wake banner is a sibling card (t_dfafdefd); latency tightening +is another (t_940f9205). Neither is implemented here. + +## Problem — busy-session swallow (root-caused from source) + +The event-driven autonomy loop already fires and routes: a lane MOVE emits a +transition, the `kanban-transition` webhook accepts it (202), origin-routing (#26) +targets the ORIGIN session, and `_build_origin_source` (webhook.py) shapes the +inbound so it lands in the origin thread's own session. + +The remaining gap: the wake POST becomes a plain TEXT `MessageEvent` and is +dispatched via the adapter's `handle_message` (base.py). When the origin session +is mid-turn (`session_key in self._active_sessions`), the event falls through to +the busy branch (base.py ~4733-4770) and is either: + +- text-debounced (`_queue_text_debounce`), or +- merged into any existing pending TEXT via + `merge_pending_message_event(..., merge_text=True)`. + +Both destroy the wake's identity: it is silently folded into an unrelated +in-flight/queued turn instead of producing its own identifiable turn. Live +2026-07-01: the wake hit a session already running a turn → no distinct wake +turn; Casey had to paste the notice manually. + +## Design — tag the wake; never merge/debounce it; drain it as its own turn + +The busy path uses a **single-slot** `_pending_messages[session_key]`, drained as +a distinct `_process_message_background` turn by the existing in-band cascade +(base.py ~5164). The fix keeps the wake in that slot but forbids it from ever +losing its identity: + +1. **Tag** the wake `MessageEvent` on the webhook side: + `event.metadata["kanban_transition_wake"] = True`, set only on the + origin-routed transition path (detected via the emitter-stamped `origin_*` + fields — the same discriminator `_build_origin_source` uses, so ordinary + webhook routes like `github-prs` are never tagged). A pure predicate + `is_transition_wake_event(event)` reads the flag. +2. **Bypass debounce + text-merge** in `handle_message`'s busy branch: when the + event is a transition wake, queue it un-merged (never appended into pending + text, never debounced) so the in-band drain runs it as a **distinct turn**. +3. **Precedence in `merge_pending_message_event`**: a transition wake and a + non-wake must never merge into one another. The single slot can hold only one + event; the wake WINS (a dropped autonomy wake is worse than a user follow-up + the user can resend). Concretely: refuse to overwrite a pending wake with a + non-wake; a wake replaces a pending non-wake intact. + +This preserves every race-guard invariant (no parallel queue, no new drain site, +no synthetic mid-loop user message, no change to `_active_sessions` lifecycle). +Prompt-cache + role-alternation are untouched: the wake is delivered as an +ordinary next-turn user message via the same cascade every follow-up uses. + +## Files + +- `gateway/platforms/base.py` — `is_transition_wake_event`, busy-branch bypass, + `merge_pending_message_event` wake-precedence. +- `gateway/platforms/webhook.py` — `_is_transition_wake_payload` + tag the wake + event. +- tests — `tests/gateway/test_transition_wake_busy_session.py` (predicate, merge + precedence, busy-branch queue) + one added case in + `tests/gateway/test_webhook_origin_routing.py` (wake event is tagged). +- PATCHES.md row. + +## Out of scope (sibling cards) + +- Self-announcing wake banner (marker + task_id + from→to lane) — t_dfafdefd. +- Latency tightening (notifier interval lag) — t_940f9205. diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 1025964dc43b..43d1eae79b3c 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2061,6 +2061,19 @@ def text(self) -> str: return str.__str__(self) +def is_transition_wake_event(event: "MessageEvent") -> bool: + """True for a kanban-transition wake event. + + The webhook route tags an origin-routed transition wake with + ``metadata["kanban_transition_wake"] = True``. The busy-session path uses + this to keep the wake a DISTINCT turn — never text-merging or debouncing it + into an unrelated in-progress/queued turn (defect 1: the busy-session + swallow). + """ + md = getattr(event, "metadata", None) + return bool(md and md.get("kanban_transition_wake")) + + def merge_pending_message_event( pending_messages: Dict[str, MessageEvent], session_key: str, @@ -2078,8 +2091,30 @@ def merge_pending_message_event( instead of replacing the pending turn. This is used for Telegram bursty follow-ups so a multi-part user thought is not silently truncated to only the last queued fragment. + + A kanban-transition WAKE event never merges with a non-wake, in either + direction (defect 1): the single pending slot can hold one event, and the + wake WINS — a dropped autonomy wake (folded into an unrelated turn, or + clobbered outright) is worse than a user follow-up the user can resend. So: a + pending wake is never overwritten by a later non-wake, and an incoming wake + replaces a pending non-wake intact (never appended). """ existing = pending_messages.get(session_key) + + # Wake-precedence guard (defect 1): keep a transition wake a distinct, + # intact turn. Handle it before any media/text merge so it is never folded + # into — or clobbered by — an unrelated message. + incoming_is_wake = is_transition_wake_event(event) + existing_is_wake = is_transition_wake_event(existing) if existing else False + if existing_is_wake and not incoming_is_wake: + # A pending wake is higher-priority; drop the later non-wake rather than + # bury the wake. (The user follow-up is recoverable; the wake is not.) + return + if incoming_is_wake: + # The wake takes the slot intact — never appended into existing text. + pending_messages[session_key] = event + return + if existing: existing_is_photo = getattr(existing, "message_type", None) == MessageType.PHOTO incoming_is_photo = event.message_type == MessageType.PHOTO @@ -4745,6 +4780,24 @@ async def handle_message(self, event: MessageEvent) -> None: merge_pending_message_event(self._pending_messages, session_key, event) return # Don't interrupt now - will run after current task completes + # Kanban-transition WAKE: never text-merge or debounce it into an + # unrelated in-progress/queued turn (defect 1, the busy-session + # swallow). Queue it as its OWN un-merged pending event so the + # in-band drain runs it as a DISTINCT turn even when the origin + # session was busy at emit time. merge_pending_... gives the wake + # precedence in the single pending slot. + if is_transition_wake_event(event): + logger.info( + "[%s] Kanban-transition wake while session %s is active — " + "queuing as a distinct un-merged turn (no debounce, no merge)", + self.name, + session_key, + ) + merge_pending_message_event( + self._pending_messages, session_key, event + ) + return # Distinct turn — will cascade after the current turn. + if self._is_queue_text_debounce_candidate(event): logger.debug( "[%s] New text message while session %s is active — " diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py index c1da6fbfd6d9..6d3931378d54 100644 --- a/gateway/platforms/webhook.py +++ b/gateway/platforms/webhook.py @@ -605,6 +605,12 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response": prompt_template, payload, event_type, route_name ) + # A kanban-transition wake is stamped with origin_* fields by the + # emitter; detect it here so the inbound event can be tagged for the + # adapter's busy path (defect 1: the busy-session swallow). The tag is + # applied to the MessageEvent below. + is_transition_wake = self._is_transition_wake_payload(payload) + # Inject skill content if configured. # We call build_skill_invocation_message() directly rather than # using /skill-name slash commands — the gateway's command parser @@ -766,6 +772,11 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response": raw_message=payload, message_id=delivery_id, ) + # Tag a kanban-transition wake so the adapter's busy path keeps it a + # DISTINCT turn (defect 1): never text-merge/debounce it into an + # unrelated in-progress/queued turn. + if is_transition_wake: + event.metadata["kanban_transition_wake"] = True logger.info( "[webhook] %s event=%s route=%s prompt_len=%d delivery=%s", @@ -791,6 +802,20 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response": status=202, ) + def _is_transition_wake_payload(self, payload: dict) -> bool: + """True when a payload is a kanban-transition wake. + + The transition emitter is the only source that stamps ``origin_*`` fields + on a webhook payload (see ``build_transition_payload``); ordinary webhook + events (GitHub PR, monitoring, cron) never carry them. Keying on the + origin coordinates is the same discriminator ``_build_origin_source`` + uses, so the busy-bypass tag (defect 1) fires exactly on the + origin-routed transition wake and nothing else. + """ + if not isinstance(payload, dict): + return False + return bool(payload.get("origin_platform") and payload.get("origin_chat_id")) + def _build_origin_source( self, origin_platform: Optional[str], diff --git a/tests/gateway/test_transition_wake_busy_session.py b/tests/gateway/test_transition_wake_busy_session.py new file mode 100644 index 000000000000..089a34047c65 --- /dev/null +++ b/tests/gateway/test_transition_wake_busy_session.py @@ -0,0 +1,143 @@ +"""Defect 1 (busy-session swallow): a kanban-transition wake must produce a +DISTINCT turn even when the origin session is mid-turn. + +Proven live 2026-07-01: the wake POST targets the origin session correctly, but +when that session is busy, ``handle_message`` queue-merges the inbound wake into +the in-progress/queued turn (text debounce + ``merge_pending_message_event`` +with ``merge_text=True``). A wake landing during an active turn is absorbed +silently instead of producing its own identifiable turn. + +The fix tags a transition-wake ``MessageEvent`` (metadata flag) so the busy path +never text-merges/debounces it — it is queued un-merged and drained by the +existing in-band cascade as a distinct turn, and it takes precedence over an +unrelated pending non-wake message in the single pending slot (a dropped autonomy +wake is worse than a user follow-up the user can resend). +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + is_transition_wake_event, + merge_pending_message_event, +) +from gateway.session import SessionSource, build_session_key + + +class _StubAdapter(BasePlatformAdapter): + async def connect(self, *, is_reconnect: bool = False): + pass + + async def disconnect(self): + pass + + async def send(self, chat_id, text, **kwargs): + return None + + async def get_chat_info(self, chat_id): + return {} + + +def _make_adapter(): + adapter = _StubAdapter(PlatformConfig(enabled=True, token="t"), Platform.TELEGRAM) + adapter._send_with_retry = AsyncMock(return_value=None) + return adapter + + +def _src(chat_id="42"): + return SessionSource(platform=Platform.TELEGRAM, chat_id=chat_id, chat_type="dm") + + +def _text_event(text="hi", chat_id="42"): + return MessageEvent(text=text, message_type=MessageType.TEXT, source=_src(chat_id)) + + +def _wake_event(text="⟪AUTONOMOUS-WAKE⟫ t_x", chat_id="42"): + ev = MessageEvent(text=text, message_type=MessageType.TEXT, source=_src(chat_id)) + ev.metadata["kanban_transition_wake"] = True + return ev + + +def _sk(chat_id="42"): + return build_session_key(_src(chat_id)) + + +# --- the predicate -------------------------------------------------------------- + +def test_is_transition_wake_event_reads_metadata_flag(): + assert is_transition_wake_event(_wake_event()) is True + assert is_transition_wake_event(_text_event()) is False + + +# --- merge precedence: a wake never loses its identity --------------------------- + +def test_wake_is_not_text_merged_into_pending_user_text(): + """A pending user TEXT + an incoming wake must NOT newline-merge — the wake + stays a distinct event (its banner would otherwise be buried in an unrelated + turn's text).""" + pending = {} + sk = _sk() + merge_pending_message_event(pending, sk, _text_event("user question"), merge_text=True) + merge_pending_message_event(pending, sk, _wake_event("⟪AUTONOMOUS-WAKE⟫ t_x"), merge_text=True) + slot = pending[sk] + # The wake WINS the single slot and is preserved intact (never appended). + assert is_transition_wake_event(slot) + assert slot.text == "⟪AUTONOMOUS-WAKE⟫ t_x" + assert "user question" not in slot.text + + +def test_pending_wake_is_not_overwritten_by_a_later_user_text(): + """A pending wake must not be clobbered by a subsequent non-wake message — + the autonomy wake is higher-priority and rarer.""" + pending = {} + sk = _sk() + merge_pending_message_event(pending, sk, _wake_event("⟪AUTONOMOUS-WAKE⟫ t_x"), merge_text=True) + merge_pending_message_event(pending, sk, _text_event("later user text"), merge_text=True) + slot = pending[sk] + assert is_transition_wake_event(slot) + assert slot.text == "⟪AUTONOMOUS-WAKE⟫ t_x" + + +def test_non_wake_text_still_merges_normally(): + """Regression guard: ordinary text-into-text merge is unchanged.""" + pending = {} + sk = _sk() + merge_pending_message_event(pending, sk, _text_event("part one"), merge_text=True) + merge_pending_message_event(pending, sk, _text_event("part two"), merge_text=True) + assert pending[sk].text == "part one\npart two" + assert not is_transition_wake_event(pending[sk]) + + +# --- the busy branch queues a wake un-merged as a distinct turn ------------------ + +@pytest.mark.asyncio +async def test_busy_session_queues_wake_un_merged_without_debounce(): + """When the session is active, a transition wake is queued as its own + un-merged pending turn (never debounced, never text-merged), so the in-band + drain runs it as a DISTINCT turn.""" + import asyncio + + adapter = _make_adapter() + adapter._message_handler = AsyncMock(return_value="ok") + # Force the queue/debounce path that would otherwise absorb text. + adapter._busy_text_mode = "queue" + sk = _sk() + + # Mark the session active so handle_message takes the busy branch. + adapter._active_sessions[sk] = asyncio.Event() + + await adapter.handle_message(_wake_event("⟪AUTONOMOUS-WAKE⟫ t_x")) + + # The wake must be queued as a distinct, un-merged pending event — + # NOT buffered in a debounce timer, NOT merged into other text. + slot = adapter._pending_messages.get(sk) + assert slot is not None, "wake was swallowed (debounced or dropped)" + assert is_transition_wake_event(slot) + assert slot.text == "⟪AUTONOMOUS-WAKE⟫ t_x" diff --git a/tests/gateway/test_webhook_origin_routing.py b/tests/gateway/test_webhook_origin_routing.py index 39bd542c94ae..ac23951be921 100644 --- a/tests/gateway/test_webhook_origin_routing.py +++ b/tests/gateway/test_webhook_origin_routing.py @@ -121,3 +121,33 @@ def test_origin_source_none_when_fields_absent(): def test_origin_source_unknown_platform_falls_back(): adapter = _make_adapter(_kt_route()) assert adapter._build_origin_source("nope", "123", "123") is None + + +@pytest.mark.asyncio +async def test_wake_event_is_tagged_for_busy_bypass(): + """Defect 1: the origin-routed wake event must be TAGGED so the adapter's + busy path never text-merges/debounces it (it produces its own turn).""" + 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_platform": "discord", + "origin_chat_id": "1520255822704152666", + "origin_thread_id": "1520255822704152666", + }, + ) + assert resp.status == 202 + + ev = captured["event"] + assert ev.metadata.get("kanban_transition_wake") is True