A message that arrives mid-turn is parked in the adapter's pending slot and drained in-band by the runner's recursive _run_agent, a path that never touches on_processing_start / on_processing_complete — so queued, interrupting and steer-demoted messages never got the read-receipt reaction idle-session messages get, on every adapter implementing the hooks. Bracket the drain in TurnRunner._run_agent_queued_followup with the hooks, resolving the adapter from the follow-up's own source (multiplex-safe). Fires only for real inbound platform events (message_id or raw_message present) and only for adapters that override on_processing_start, so complete-only adapters (Google Chat, webhook) are not handed an early completion. Cancels are classified like _process_message_background does. Re-ported onto current main: the drain moved from gateway/run.py to gateway/run_turn.py::_run_agent_queued_followup (#102117 decomposition) and the helpers now live in the topical sibling gateway/run_turn_followup_ack.py. Original commits (8828a3db37c2, f3c4f7a68c8d) by Mira Solari <268252643+mira-solari@users.noreply.github.com>: --- 8828a3db37c2 fix(gateway): fire processing hooks for runner-drained queued follow-ups A message that arrives while a turn is already running never gets the processing-start acknowledgement — the 👀 read receipt on Slack, and the equivalent in-progress reaction on Discord, Telegram, Feishu, Matrix, Signal and Photon. It is not added-then-removed; the hook is never called. `_run_processing_hook("on_processing_start", …)` has exactly one call site, inside `BasePlatformAdapter._process_message_background` (base.py:5403). `handle_message` takes the busy branch at base.py:5165, parks the event and returns at base.py:5316 — above `_start_session_processing`, which is the only thing that spawns `_process_message_background`. The parked event is then drained in-band by the runner (`_dequeue_pending_event`, run.py:23277) and replayed through a recursive `_run_agent` that touches no adapter hooks. Because `get_pending_message` pops, the adapter's own drain (base.py:5823) — the one path that would fire the hook — finds an empty slot. The gap is structural, not a race, and it is shared by every mode (`queue`, `interrupt`, steer-demoted-to-queue), by `/queue`, by photo-burst and text-debounce flushes, and by voice drains. Fire the existing hook pair around the recursive call. The follow-up now gets the same lifecycle an idle-session message already gets, on every platform, through one shared site. Details that shaped the placement: - Fired after the depth-cap requeue (run.py:23372) and after every discard and early return in the block, so no path can strand an in-progress marker: from that point on, control either reaches the recursion or raises, and both close the hook. - Fired before `_refresh_agent_cache_message_count` so that re-baseline stays adjacent to the recursive call it exists to protect — inserting a reaction round-trip between them would widen the window in which the cross-process coherence guard (#45966) can trip on our own writes and rebuild the agent, destroying the prompt-cache prefix #46237 preserves. - The hook adapter is resolved from the follow-up's own source, not the completing turn's: a multiplexed gateway can route it to a different profile's adapter, and only that instance holds the per-message reaction state. - Gated on a truthy `message_id`, which is what every adapter's own hook already checks. Synthetic drains (`/goal` continuations, wake-ups, CLI hand-offs) carry no id and stay silent; `interrupt_message` and leftover `/steer` carry no event at all. - Cancellation maps to CANCELLED rather than FAILURE, matching _process_message_background — Telegram clears the marker on CANCELLED and Signal deliberately leaves it, so the distinction is load-bearing. Outcome is SUCCESS unless the recursion raises. That mirrors the existing non-queued contract, where a run returning `failed: True` still delivers a diagnostic message and reports SUCCESS; making the outcome track agent failure is a separate change that should apply to both paths at once. No new config, no new env var, no new hook, no change to message construction or role alternation. Fixes #72502 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --- f3c4f7a68c8d fix(gateway): don't hand a completion to complete-only adapters; cover raw-envelope events Self-review of the previous commit found three defects in it. All three are about the *guard*, not the mechanism. 1. Complete-only adapters were handed an unpaired completion. Google Chat (google_chat/adapter.py:2766) and webhook (webhook.py:901) implement on_processing_complete WITHOUT on_processing_start, and theirs is end-of-cycle teardown, not a reaction: Google Chat reaps the typing card, patching it to "(no reply)"/"(interrupted)", and webhook ends the per-delivery session. The drain fires before the follow-up's reply is delivered — delivery happens after the whole chain unwinds back into _process_message_background — so a Google Chat space would get a permanent "(no reply)" tombstone on every queued follow-up, and the real answer would then land as a separate message. Now gated on the adapter actually overriding on_processing_start: we bracket, so both halves must be ours. 2. The message_id gate made the fix a no-op on Signal, which the previous commit message claimed to fix. SignalAdapter never sets message_id (signal.py:749-766) — its hook keys off raw_message["sender"] and ["timestamp_ms"] via _extract_reaction_target — and Discord's start hook reads raw_message too. Gate is now message_id OR raw_message; synthetic drains still carry neither, so /goal continuations stay silent. 3. Cancellation was classified unconditionally as CANCELLED. base.py:5862-5868 only reports CANCELLED for a task in _expected_cancelled_tasks (/stop, /new, /reset, adapter cleanup) and downgrades anything else to FAILURE. Signal and Matrix deliberately LEAVE the marker in place on CANCELLED, so the previous version would strand exactly the marker this PR exists to clear. Now mirrors base.py via _followup_cancel_outcome(). Also moves _refresh_agent_cache_message_count inside the try. It awaits DB I/O and guards it with `except Exception`, which does not catch cancellation, so a /stop landing there escaped both handlers and stranded the marker. Ordering is unchanged, so the prompt-cache adjacency the previous commit describes still holds. Two new tests, both failing against the previous commit: test_complete_only_adapter_is_left_alone and test_raw_envelope_only_followup_is_acknowledged. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Mira Solari
+29
-9
@@ -21,7 +21,7 @@ from contextlib import nullcontext, suppress
|
||||
from contextvars import copy_context
|
||||
from gateway.config import Platform
|
||||
from gateway.media_repair import repair_explicit_computer_use_media_paths
|
||||
from gateway.platforms.base import BasePlatformAdapter
|
||||
from gateway.platforms.base import BasePlatformAdapter, ProcessingOutcome
|
||||
from gateway.platforms.event import MessageEvent
|
||||
from gateway.session import (
|
||||
SessionSource, _session_key_namespace, build_channel_continuity_note,
|
||||
@@ -3621,15 +3621,35 @@ class GatewayTurnMixin:
|
||||
# whole _run_agent chain unwinds — too late for the in-band follow-up. Use the same (session_key,
|
||||
# session_id) the recursive call runs under so the snapshot matches exactly what the follow-up's
|
||||
# guard will consult. Fail-safe in helper.
|
||||
await self._refresh_agent_cache_message_count(session_key, session_id)
|
||||
# Acknowledge the follow-up the way an idle-session message is: this in-band drain is the only
|
||||
# place a queued/interrupting message ever runs, so base.py's hook site is never entered for it.
|
||||
# Resolve the adapter from the follow-up's OWN source — a multiplexed gateway can route it to a
|
||||
# different profile's adapter, and only that instance holds the per-message reaction state.
|
||||
from gateway.run_turn_followup_ack import _followup_cancel_outcome, _run_followup_processing_hook
|
||||
_hook_adapter = self._adapter_for_source(next_source) if pending_event is not None else None
|
||||
await _run_followup_processing_hook(_hook_adapter, pending_event, "on_processing_start")
|
||||
# The re-baseline sits inside the try: a /stop landing on its DB await must still close the marker
|
||||
# (the helper's own ``except Exception`` does not catch cancellation).
|
||||
try:
|
||||
await self._refresh_agent_cache_message_count(session_key, session_id)
|
||||
|
||||
followup_result = await self._run_agent(
|
||||
message=next_message, context_prompt=turn_ctx.context_prompt, history=updated_history,
|
||||
source=next_source, session_id=session_id, session_key=next_session_key,
|
||||
run_generation=run_generation, _interrupt_depth=_interrupt_depth + 1,
|
||||
event_message_id=next_message_id, inbound_message_id=next_inbound_id,
|
||||
channel_prompt=next_channel_prompt, message_type=next_message_type,
|
||||
)
|
||||
followup_result = await self._run_agent(
|
||||
message=next_message, context_prompt=turn_ctx.context_prompt, history=updated_history,
|
||||
source=next_source, session_id=session_id, session_key=next_session_key,
|
||||
run_generation=run_generation, _interrupt_depth=_interrupt_depth + 1,
|
||||
event_message_id=next_message_id, inbound_message_id=next_inbound_id,
|
||||
channel_prompt=next_channel_prompt, message_type=next_message_type,
|
||||
)
|
||||
except asyncio.CancelledError:
|
||||
await _run_followup_processing_hook(
|
||||
_hook_adapter, pending_event, "on_processing_complete", _followup_cancel_outcome(_hook_adapter))
|
||||
raise
|
||||
except BaseException:
|
||||
await _run_followup_processing_hook(
|
||||
_hook_adapter, pending_event, "on_processing_complete", ProcessingOutcome.FAILURE)
|
||||
raise
|
||||
await _run_followup_processing_hook(
|
||||
_hook_adapter, pending_event, "on_processing_complete", ProcessingOutcome.SUCCESS)
|
||||
merged = _preserve_queued_followup_history_offset(result, followup_result)
|
||||
# The TERMINAL turn of the chain owns the ledger identity for the outer final send, which
|
||||
# the adapter brackets against the event that OPENED the chain. Without this the terminal
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Processing-lifecycle hooks for runner-drained queued follow-ups.
|
||||
|
||||
A message that arrives mid-turn is parked in the adapter's pending slot and drained in-band by
|
||||
``TurnRunner._run_agent_queued_followup``, never by ``BasePlatformAdapter._process_message_background``
|
||||
— the only other call site for ``on_processing_start`` / ``on_processing_complete``. Without firing
|
||||
them here every adapter that renders a read-receipt reaction from the hooks silently skips queued,
|
||||
interrupting and steer-demoted messages (#72502, salvage #72503).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
from gateway.platforms.base import BasePlatformAdapter, MessageEvent, ProcessingOutcome
|
||||
|
||||
|
||||
def _followup_processing_hooks_apply(adapter, event: MessageEvent | None) -> bool:
|
||||
"""Both conditions are necessary: a real inbound platform message to acknowledge (adapters key their
|
||||
marker off ``message_id`` — Slack/Telegram/Feishu/Matrix/Photon — or off ``raw_message`` — Signal,
|
||||
Discord; synthetic drains such as ``/goal`` continuations, wake-ups and startup auto-resume carry
|
||||
neither and must stay silent), and an adapter that overrides ``on_processing_start``. We bracket, so
|
||||
both halves must belong to us: a complete-only adapter (Google Chat reaps its typing card there,
|
||||
webhook ends its per-delivery session) would otherwise be handed a completion for a turn whose reply
|
||||
is delivered only after the drain chain unwinds back into ``_process_message_background``."""
|
||||
if adapter is None or event is None:
|
||||
return False
|
||||
if not (getattr(event, "message_id", None) or getattr(event, "raw_message", None)):
|
||||
return False
|
||||
start_hook = getattr(type(adapter), "on_processing_start", None)
|
||||
return start_hook is not None and start_hook is not BasePlatformAdapter.on_processing_start
|
||||
|
||||
|
||||
def _followup_cancel_outcome(adapter) -> ProcessingOutcome:
|
||||
"""Classify a cancelled follow-up exactly as ``_process_message_background`` does: only cancels the
|
||||
adapter itself routed (``/stop``, ``/new``, ``/reset``, cleanup) are CANCELLED, anything else is a
|
||||
failure. Signal and Matrix leave the in-progress marker in place on CANCELLED, so reporting an
|
||||
unexpected cancellation as CANCELLED would strand it."""
|
||||
expected = getattr(adapter, "_expected_cancelled_tasks", None)
|
||||
if expected is None:
|
||||
return ProcessingOutcome.FAILURE
|
||||
try:
|
||||
current = asyncio.current_task()
|
||||
except RuntimeError:
|
||||
current = None
|
||||
if current is None:
|
||||
return ProcessingOutcome.FAILURE
|
||||
try:
|
||||
return ProcessingOutcome.CANCELLED if current in expected else ProcessingOutcome.FAILURE
|
||||
except TypeError:
|
||||
return ProcessingOutcome.FAILURE
|
||||
|
||||
|
||||
async def _run_followup_processing_hook(adapter, event: MessageEvent | None, hook_name: str, *args) -> None:
|
||||
"""Fire one lifecycle hook for a runner-drained follow-up; no-op per ``_followup_processing_hooks_apply``."""
|
||||
if not _followup_processing_hooks_apply(adapter, event):
|
||||
return
|
||||
run_hook = getattr(adapter, "_run_processing_hook", None)
|
||||
if not callable(run_hook):
|
||||
return
|
||||
await run_hook(hook_name, event, *args)
|
||||
@@ -0,0 +1,313 @@
|
||||
"""Processing-hook parity for queued follow-up turns.
|
||||
|
||||
A message that arrives while a turn is already running is parked in the
|
||||
adapter's ``_pending_messages`` slot and drained *in-band* by
|
||||
``GatewayRunner._run_agent`` rather than by
|
||||
``BasePlatformAdapter._process_message_background``. The runner-side drain
|
||||
must still fire the ``on_processing_start`` / ``on_processing_complete``
|
||||
lifecycle hooks, otherwise every platform that renders a read-receipt
|
||||
reaction from those hooks (Slack 👀, Discord, Telegram, Feishu, Matrix,
|
||||
Signal, ...) silently skips the acknowledgement for mid-turn messages.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import sys
|
||||
import types
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.config import Platform, PlatformConfig
|
||||
from gateway.platforms.base import (
|
||||
BasePlatformAdapter,
|
||||
MessageEvent,
|
||||
MessageType,
|
||||
ProcessingOutcome,
|
||||
SendResult,
|
||||
)
|
||||
from gateway.session import SessionSource
|
||||
|
||||
|
||||
class HookRecordingAdapter(BasePlatformAdapter):
|
||||
"""Adapter that records the processing-hook lifecycle it is driven through."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(PlatformConfig(enabled=True, token="***"), Platform.TELEGRAM)
|
||||
self.started: list = []
|
||||
self.completed: list = []
|
||||
|
||||
async def connect(self) -> bool:
|
||||
return True
|
||||
|
||||
async def disconnect(self) -> None:
|
||||
return None
|
||||
|
||||
async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
|
||||
return SendResult(success=True, message_id="sent-1")
|
||||
|
||||
async def send_typing(self, chat_id, metadata=None) -> None:
|
||||
return None
|
||||
|
||||
async def stop_typing(self, chat_id) -> None:
|
||||
return None
|
||||
|
||||
async def get_chat_info(self, chat_id: str):
|
||||
return {"id": chat_id}
|
||||
|
||||
async def on_processing_start(self, event: MessageEvent) -> None:
|
||||
self.started.append(getattr(event, "message_id", None))
|
||||
|
||||
async def on_processing_complete(self, event, outcome) -> None:
|
||||
self.completed.append((getattr(event, "message_id", None), outcome))
|
||||
|
||||
|
||||
class _TwoTurnAgent:
|
||||
calls: list = []
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self.tools = []
|
||||
|
||||
def run_conversation(self, message, conversation_history=None, task_id=None):
|
||||
type(self).calls.append(message)
|
||||
return {
|
||||
"final_response": f"done-{len(type(self).calls)}",
|
||||
"messages": [],
|
||||
"api_calls": 1,
|
||||
}
|
||||
|
||||
|
||||
class _RaisingSecondTurnAgent:
|
||||
calls: list = []
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self.tools = []
|
||||
|
||||
def run_conversation(self, message, conversation_history=None, task_id=None):
|
||||
type(self).calls.append(message)
|
||||
if len(type(self).calls) >= 2:
|
||||
raise RuntimeError("boom in the queued follow-up turn")
|
||||
return {
|
||||
"final_response": "done-1",
|
||||
"messages": [],
|
||||
"api_calls": 1,
|
||||
}
|
||||
|
||||
|
||||
def _make_runner(adapter):
|
||||
gateway_run = importlib.import_module("gateway.run")
|
||||
runner = object.__new__(gateway_run.GatewayRunner)
|
||||
runner.adapters = {adapter.platform: adapter}
|
||||
runner._voice_mode = {}
|
||||
runner._prefill_messages = []
|
||||
runner._ephemeral_system_prompt = ""
|
||||
runner._reasoning_config = None
|
||||
runner._provider_routing = {}
|
||||
runner._fallback_model = None
|
||||
runner._session_db = None
|
||||
runner._running_agents = {}
|
||||
runner._session_run_generation = {}
|
||||
runner.hooks = SimpleNamespace(loaded_hooks=False)
|
||||
runner.config = SimpleNamespace(
|
||||
thread_sessions_per_user=False,
|
||||
group_sessions_per_user=False,
|
||||
stt_enabled=False,
|
||||
)
|
||||
runner._model = "openai/gpt-4.1-mini"
|
||||
runner._base_url = None
|
||||
return runner
|
||||
|
||||
|
||||
def _install_fake_agent(monkeypatch, tmp_path, agent_cls):
|
||||
fake_dotenv = types.ModuleType("dotenv")
|
||||
fake_dotenv.load_dotenv = lambda *args, **kwargs: None
|
||||
monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
|
||||
|
||||
fake_run_agent = types.ModuleType("run_agent")
|
||||
fake_run_agent.AIAgent = agent_cls
|
||||
monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
|
||||
|
||||
gateway_run = importlib.import_module("gateway.run")
|
||||
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
|
||||
monkeypatch.setattr(
|
||||
gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
|
||||
)
|
||||
|
||||
|
||||
SESSION_KEY = "agent:main:telegram:dm:4242"
|
||||
|
||||
|
||||
def _source():
|
||||
return SessionSource(platform=Platform.TELEGRAM, chat_id="4242", chat_type="dm")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_queued_followup_fires_processing_hooks(monkeypatch, tmp_path):
|
||||
"""The runner-drained follow-up gets the same start/complete hooks as a
|
||||
message that arrives while the session is idle."""
|
||||
_TwoTurnAgent.calls = []
|
||||
_install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent)
|
||||
|
||||
adapter = HookRecordingAdapter()
|
||||
runner = _make_runner(adapter)
|
||||
|
||||
adapter._pending_messages[SESSION_KEY] = MessageEvent(
|
||||
text="the follow-up",
|
||||
message_type=MessageType.TEXT,
|
||||
source=_source(),
|
||||
message_id="queued-1",
|
||||
)
|
||||
|
||||
result = await runner._run_agent(
|
||||
message="the first turn",
|
||||
context_prompt="",
|
||||
history=[],
|
||||
source=_source(),
|
||||
session_id="sess-hooks",
|
||||
session_key=SESSION_KEY,
|
||||
)
|
||||
|
||||
# The follow-up really did run in-band.
|
||||
assert result["final_response"] == "done-2"
|
||||
assert _TwoTurnAgent.calls == ["the first turn", "the follow-up"]
|
||||
|
||||
# ...and it was acknowledged through the lifecycle hooks.
|
||||
assert adapter.started == ["queued-1"]
|
||||
assert adapter.completed == [("queued-1", ProcessingOutcome.SUCCESS)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_queued_followup_failure_completes_the_hook(monkeypatch, tmp_path):
|
||||
"""A follow-up turn that blows up still closes its hook, so a platform
|
||||
never strands a 'still working' marker on the user's message."""
|
||||
_RaisingSecondTurnAgent.calls = []
|
||||
_install_fake_agent(monkeypatch, tmp_path, _RaisingSecondTurnAgent)
|
||||
|
||||
adapter = HookRecordingAdapter()
|
||||
runner = _make_runner(adapter)
|
||||
|
||||
adapter._pending_messages[SESSION_KEY] = MessageEvent(
|
||||
text="the doomed follow-up",
|
||||
message_type=MessageType.TEXT,
|
||||
source=_source(),
|
||||
message_id="queued-2",
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError):
|
||||
await runner._run_agent(
|
||||
message="the first turn",
|
||||
context_prompt="",
|
||||
history=[],
|
||||
source=_source(),
|
||||
session_id="sess-hooks-failure",
|
||||
session_key=SESSION_KEY,
|
||||
)
|
||||
|
||||
assert adapter.started == ["queued-2"]
|
||||
assert adapter.completed == [("queued-2", ProcessingOutcome.FAILURE)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthetic_followup_is_not_acknowledged(monkeypatch, tmp_path):
|
||||
"""Drains with no inbound platform message — /goal continuations, wake-ups,
|
||||
CLI hand-offs — carry no message_id and must stay silent: there is nothing
|
||||
on the platform to react to."""
|
||||
_TwoTurnAgent.calls = []
|
||||
_install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent)
|
||||
|
||||
adapter = HookRecordingAdapter()
|
||||
runner = _make_runner(adapter)
|
||||
|
||||
adapter._pending_messages[SESSION_KEY] = MessageEvent(
|
||||
text="synthetic continuation",
|
||||
message_type=MessageType.TEXT,
|
||||
source=_source(),
|
||||
message_id=None,
|
||||
)
|
||||
|
||||
result = await runner._run_agent(
|
||||
message="the first turn",
|
||||
context_prompt="",
|
||||
history=[],
|
||||
source=_source(),
|
||||
session_id="sess-hooks-synthetic",
|
||||
session_key=SESSION_KEY,
|
||||
)
|
||||
|
||||
# It still ran — we only suppressed the acknowledgement, not the turn.
|
||||
assert result["final_response"] == "done-2"
|
||||
assert _TwoTurnAgent.calls == ["the first turn", "synthetic continuation"]
|
||||
|
||||
assert adapter.started == []
|
||||
assert adapter.completed == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_raw_envelope_only_followup_is_acknowledged(monkeypatch, tmp_path):
|
||||
"""Signal never sets message_id — its hook keys off the raw envelope
|
||||
(sender + timestamp_ms) — and Discord's reads raw_message. An event
|
||||
carrying only a raw envelope is still a real inbound message."""
|
||||
_TwoTurnAgent.calls = []
|
||||
_install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent)
|
||||
|
||||
adapter = HookRecordingAdapter()
|
||||
runner = _make_runner(adapter)
|
||||
|
||||
adapter._pending_messages[SESSION_KEY] = MessageEvent(
|
||||
text="signal-shaped follow-up",
|
||||
message_type=MessageType.TEXT,
|
||||
source=_source(),
|
||||
message_id=None,
|
||||
raw_message={"sender": "+15550100", "timestamp_ms": 1700000000000},
|
||||
)
|
||||
|
||||
await runner._run_agent(
|
||||
message="the first turn",
|
||||
context_prompt="",
|
||||
history=[],
|
||||
source=_source(),
|
||||
session_id="sess-hooks-raw",
|
||||
session_key=SESSION_KEY,
|
||||
)
|
||||
|
||||
assert adapter.started == [None]
|
||||
assert adapter.completed == [(None, ProcessingOutcome.SUCCESS)]
|
||||
|
||||
|
||||
class CompleteOnlyAdapter(HookRecordingAdapter):
|
||||
"""Google Chat and webhook implement on_processing_complete WITHOUT
|
||||
on_processing_start; theirs is end-of-cycle teardown (reap the typing
|
||||
card / end the delivery session), not a reaction."""
|
||||
|
||||
on_processing_start = BasePlatformAdapter.on_processing_start
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_complete_only_adapter_is_left_alone(monkeypatch, tmp_path):
|
||||
"""We bracket, so both halves must belong to us. An adapter that only
|
||||
implements the completion half must not be handed a completion here: at
|
||||
this point the follow-up's reply has not been delivered yet, so its
|
||||
teardown would fire against a live turn."""
|
||||
_TwoTurnAgent.calls = []
|
||||
_install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent)
|
||||
|
||||
adapter = CompleteOnlyAdapter()
|
||||
runner = _make_runner(adapter)
|
||||
|
||||
adapter._pending_messages[SESSION_KEY] = MessageEvent(
|
||||
text="the follow-up",
|
||||
message_type=MessageType.TEXT,
|
||||
source=_source(),
|
||||
message_id="queued-3",
|
||||
)
|
||||
|
||||
result = await runner._run_agent(
|
||||
message="the first turn",
|
||||
context_prompt="",
|
||||
history=[],
|
||||
source=_source(),
|
||||
session_id="sess-hooks-complete-only",
|
||||
session_key=SESSION_KEY,
|
||||
)
|
||||
|
||||
assert result["final_response"] == "done-2"
|
||||
assert adapter.completed == []
|
||||
Reference in New Issue
Block a user