From 000c1e253e5bca7c7ac9be8fb50bed8047bca3d8 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:20:24 +0000 Subject: [PATCH 01/10] Channel contract: read a connection's provider and behaviour The gateway can now announce, in the capabilities frame, which provider a connection belongs to and how Nerve answers its events: post the answer or let the agent answer with a tool, and rotate or keep the conversation's session. A hosted channel needs both to serve a provider other than Slack. Absent members keep today's meaning, so an older gateway changes nothing. The checks follow the gateway's contract, and the contract's new sample is copied byte for byte. Co-Authored-By: Claude Opus 5.5 --- nerve/channels/hosted/contract/__init__.py | 2 + nerve/channels/hosted/contract/model.py | 75 +++++++++++++++- .../capabilities_behaviour.json | 23 +++++ tests/test_channel_contract.py | 85 ++++++++++++++++++- 4 files changed, 183 insertions(+), 2 deletions(-) create mode 100644 tests/fixtures/channel_contract/capabilities_behaviour.json diff --git a/nerve/channels/hosted/contract/__init__.py b/nerve/channels/hosted/contract/__init__.py index 8573344af..b5a36e569 100644 --- a/nerve/channels/hosted/contract/__init__.py +++ b/nerve/channels/hosted/contract/__init__.py @@ -14,6 +14,7 @@ SUBSCRIPTIONS_VERSION, AuthorReference, Capabilities, + ConnectionBehaviour, ConnectionStatus, ContentPart, Drain, @@ -54,6 +55,7 @@ "SUBSCRIPTIONS_VERSION", "AuthorReference", "Capabilities", + "ConnectionBehaviour", "ConnectionStatus", "ContentPart", "Drain", diff --git a/nerve/channels/hosted/contract/model.py b/nerve/channels/hosted/contract/model.py index 24dbba41d..15731ddf5 100644 --- a/nerve/channels/hosted/contract/model.py +++ b/nerve/channels/hosted/contract/model.py @@ -78,6 +78,17 @@ # A provider name is a lowercase token of at most this many bytes. MAX_PROVIDER_BYTES = 64 _PROVIDER_CHARACTERS = frozenset("abcdefghijklmnopqrstuvwxyz0123456789_-.") +# How Nerve answers the turns of a connection's events (ConnectionBehaviour). +ANSWER_POST = "post" +ANSWER_TOOL = "tool" +ANSWER_MODES = (ANSWER_POST, ANSWER_TOOL) +# Which session the turns of a conversation go to (ConnectionBehaviour). +SESSION_ROTATE = "rotate" +SESSION_KEEP = "keep" +SESSION_MODES = (SESSION_ROTATE, SESSION_KEEP) +# The name of a reply tool: 1 to this many characters of a-z, 0-9 and _. +MAX_REPLY_TOOL_BYTES = 128 +_REPLY_TOOL = re.compile(rf"[a-z0-9_]{{1,{MAX_REPLY_TOOL_BYTES}}}") def byte_length(value: str) -> int: @@ -512,18 +523,80 @@ class ConnectionLimits: file_bytes: int = 0 +@_record +class ConnectionBehaviour: + """How Nerve answers the turns of a connection's events, and keeps their sessions. + + ``answer`` is ``post``: Nerve posts the turn's answer in the + conversation. Or it is ``tool``: Nerve posts nothing for the turn, the + answer stays in the session, and the agent answers only with a tool. + ``reply_tool`` is the name of that tool, which Nerve shows to the agent. + Only ``tool`` can have it, and it is optional. ``session`` is + ``rotate``: a conversation gets a new session after the sticky period. + Or it is ``keep``: a conversation keeps its last session, however long + it was idle. An empty member, as from an older gateway, has its default: + ``post``, no tool, and ``rotate``. + """ + + answer: str = omit_empty() + reply_tool: str = omit_empty() + session: str = omit_empty() + + @property + def posts_answer(self) -> bool: + """Whether Nerve posts the turn's answer in the conversation.""" + return self.answer != ANSWER_TOOL + + @property + def keeps_session(self) -> bool: + """Whether a conversation keeps its last session past the sticky period.""" + return self.session == SESSION_KEEP + + def validate(self) -> None: + """Check the modes and the reply tool, as the gateway does.""" + if self.answer: + _one_of("answer mode", self.answer, ANSWER_MODES) + if self.session: + _one_of("session mode", self.session, SESSION_MODES) + if self.reply_tool: + if self.answer != ANSWER_TOOL: + malformed("a reply tool is permitted only with the tool answer mode") + if _REPLY_TOOL.fullmatch(self.reply_tool) is None: + malformed(f"reply tool is not 1 to {MAX_REPLY_TOOL_BYTES} characters of a-z, 0-9 and _") + + @_record class Capabilities: - """A connection's operations and limits.""" + """A connection's operations, limits, provider, and behaviour. + + ``provider`` is the provider of the connection, such as ``slack``. An + older gateway leaves it out, and the connection then serves every hosted + chat channel. ``behaviour`` has its defaults when it is absent. A + connection without operations, such as one that only delivers events, + still sends usable limits. + """ operations: tuple[str, ...] = omit_empty(()) limits: ConnectionLimits = struct(ConnectionLimits) + provider: str = omit_empty() + behaviour: ConnectionBehaviour = struct(ConnectionBehaviour) def validate(self) -> None: # Nerve paces, splits, and times out operations by these limits. limits = self.limits if min(limits.in_flight_operations, limits.operation_deadline_millis, limits.text_characters) < 1: reject(RejectionReason.LIMIT_EXCEEDED, "capabilities have no usable operation or text limit") + # Nerve chooses the connections of a channel by this name. + if self.provider and not valid_provider(self.provider): + malformed("capabilities provider is not a lowercase token") + self.behaviour.validate() + + def serves(self, provider: str) -> bool: + """Whether the connection serves the hosted chat channel of *provider*. + + A connection that names no provider serves every hosted chat channel. + """ + return self.provider in ("", provider) @_record diff --git a/tests/fixtures/channel_contract/capabilities_behaviour.json b/tests/fixtures/channel_contract/capabilities_behaviour.json new file mode 100644 index 000000000..fae080e43 --- /dev/null +++ b/tests/fixtures/channel_contract/capabilities_behaviour.json @@ -0,0 +1,23 @@ +{ + "version": "1", + "kind": "capabilities", + "request_id": "g-2", + "connection_id": "8c82bd61-913c-40f2-b7cb-c50ba5c56a07", + "payload": { + "capabilities": { + "limits": { + "in_flight_operations": 16, + "operation_deadline_millis": 30000, + "text_characters": 4000, + "edit_interval_millis": 1200, + "file_bytes": 16777216 + }, + "provider": "email", + "behaviour": { + "answer": "tool", + "reply_tool": "mcp__email__send", + "session": "keep" + } + } + } +} diff --git a/tests/test_channel_contract.py b/tests/test_channel_contract.py index 9bf6c841d..13478734c 100644 --- a/tests/test_channel_contract.py +++ b/tests/test_channel_contract.py @@ -12,12 +12,14 @@ import json import uuid from datetime import datetime, timedelta, timezone +from pathlib import Path from typing import Any import pytest from nerve.channels.hosted.contract import ( AuthorReference, + ConnectionBehaviour, Drain, Envelope, Heartbeat, @@ -401,7 +403,88 @@ def test_capabilities_need_usable_limits(self): assert reason(value) == "limit_exceeded" -TIME = datetime(2026, 9, 16, 10, 0, 0, 250000, tzinfo=timezone.utc) +# A copy of the channel contract's sample of a connection that announces its +# provider and behaviour. It must stay byte for byte the same as the +# contract's ``testdata/frames/gateway/capabilities_behaviour.json``. +BEHAVIOUR_SAMPLE = Path(__file__).parent / "fixtures" / "channel_contract" / "capabilities_behaviour.json" + + +def behaviour_frame(**behaviour: Any) -> dict[str, Any]: + body = capabilities() + body["provider"] = "slack" + body["behaviour"] = behaviour + return frame("capabilities", body, connection_id=CONNECTION_ID) + + +class TestConnectionBehaviour: + def test_the_contract_sample_decodes_into_its_provider_and_behaviour(self): + envelope = decode(BEHAVIOUR_SAMPLE.read_text(encoding="utf-8")) + body = envelope.body + + assert envelope.connection_id == uuid.UUID("8c82bd61-913c-40f2-b7cb-c50ba5c56a07") + assert (body.provider, body.operations, body.limits.text_characters) == ("email", (), 4000) + assert body.behaviour == ConnectionBehaviour(answer="tool", reply_tool="mcp__email__send", session="keep") + assert (body.behaviour.posts_answer, body.behaviour.keeps_session) == (False, True) + assert body.serves("email") and not body.serves("slack") + + def test_without_provider_and_behaviour_the_defaults_apply(self): + body = decode(frame("capabilities", capabilities(), connection_id=CONNECTION_ID)).body + + assert (body.provider, body.behaviour) == ("", ConnectionBehaviour()) + assert (body.behaviour.posts_answer, body.behaviour.keeps_session) == (True, False) + assert body.serves("slack") and body.serves("email") + + @pytest.mark.parametrize("behaviour", [{}, {"answer": "post", "session": "rotate"}, {"answer": "", "session": ""}]) + def test_the_slack_behaviour_is_the_default(self, behaviour): + body = decode(behaviour_frame(**behaviour)).body + + assert (body.provider, body.behaviour.posts_answer, body.behaviour.keeps_session) == ("slack", True, False) + + def test_a_tool_answer_needs_no_reply_tool(self): + body = decode(behaviour_frame(answer="tool")).body + + assert (body.behaviour.posts_answer, body.behaviour.reply_tool) == (False, "") + + def test_unknown_behaviour_members_are_ignored(self): + body = decode(behaviour_frame(answer="tool", presence="none")).body + + assert body.behaviour == ConnectionBehaviour(answer="tool") + + @pytest.mark.parametrize("behaviour", [{"answer": "none"}, {"session": "pinned"}, {"answer": "POST"}]) + def test_an_unknown_mode_is_refused(self, behaviour): + assert reason(behaviour_frame(**behaviour)) == "kind_unsupported" + + @pytest.mark.parametrize("behaviour", [ + {"reply_tool": "send"}, + {"answer": "post", "reply_tool": "send"}, + {"answer": "tool", "reply_tool": "Send"}, + {"answer": "tool", "reply_tool": "send-reply"}, + {"answer": "tool", "reply_tool": "x" * 129}, + {"answer": "tool", "reply_tool": "send\n"}, + ], ids=["no_answer_mode", "post", "upper_case", "hyphen", "too_long", "line_feed"]) + def test_a_reply_tool_outside_its_rules_is_malformed(self, behaviour): + assert reason(behaviour_frame(**behaviour)) == "malformed_frame" + + def test_a_reply_tool_of_128_characters_is_taken(self): + body = decode(behaviour_frame(answer="tool", reply_tool="x" * 128)).body + + assert body.behaviour.reply_tool == "x" * 128 + + @pytest.mark.parametrize("provider", ["Slack", "slack chat", "x" * 65]) + def test_a_provider_that_is_not_a_lowercase_token_is_malformed(self, provider): + value = behaviour_frame() + value["payload"]["capabilities"]["provider"] = provider + + assert reason(value) == "malformed_frame" + + def test_a_behaviour_of_another_type_is_malformed(self): + value = behaviour_frame() + value["payload"]["capabilities"]["behaviour"] = "tool" + + assert reason(value) == "malformed_frame" + + +TIME =datetime(2026, 9, 16, 10, 0, 0, 250000, tzinfo=timezone.utc) MESSAGE = MessageTarget(conversation=ConversationReference(id="C1"), message=MessageReference(id="1.2")) MESSAGE_JSON = {"conversation": {"id": "C1"}, "message": {"id": "1.2"}} From b591e432f1dfe0be5aba2b490f7a6a22bb3d7976 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:26:03 +0000 Subject: [PATCH 02/10] Hosted channels: a table of chat providers, scoped connections The hosted runtime built a channel only for Slack, and the channel read Slack-shaped settings and a Slack-shaped switch from the provider's config section. A provider without such a section had a channel that was never switched on, which stopped the inbox reader for every provider. A table of hosted chat providers now gives each provider its hosted mode, its switch, its channel settings with defaults, and its presentation rules. Slack is the only entry and keeps its behaviour, including the pause of intake while it is switched off. A provider added later rejects its events while it is off, so it never stops the reader. A channel uses only the connections of its own provider when the gateway names the provider, for the choice of connection, limits, availability, notify, and addressed delivery. Co-Authored-By: Claude Opus 5.5 --- nerve/channels/hosted/channel.py | 122 +++++++---- nerve/channels/hosted/manager.py | 12 +- nerve/channels/hosted/operations.py | 10 +- nerve/channels/hosted/providers.py | 159 ++++++++++++++ nerve/channels/hosted/runtime.py | 57 ++--- tests/fake_channel_gateway.py | 25 ++- tests/test_hosted_channel_behaviour.py | 280 +++++++++++++++++++++++++ 7 files changed, 566 insertions(+), 99 deletions(-) create mode 100644 nerve/channels/hosted/providers.py create mode 100644 tests/test_hosted_channel_behaviour.py diff --git a/nerve/channels/hosted/channel.py b/nerve/channels/hosted/channel.py index 80c5f73ab..68ed26599 100644 --- a/nerve/channels/hosted/channel.py +++ b/nerve/channels/hosted/channel.py @@ -28,14 +28,20 @@ linked to the author, for attribution only: the sender rules do not see it, and it adds no actor. +The channel serves the connections of its provider. A connection whose +capabilities name no provider, as from an older gateway, serves every hosted +chat channel. The provider's entry in +:data:`~nerve.channels.hosted.providers.HOSTED_CHAT_PROVIDERS` gives the +channel's switch and settings. + Outbound messages are contract operations on the connection that the -conversation's latest accepted event came on, or on the only connection that -supports the operation. :meth:`send` posts CommonMark text in parts of at -most the connection's advisory ``text_characters``. Streaming uses a -placeholder message, edits paced by ``edit_interval_millis``, and a delete. -The typing indicator is an ``eyes`` reaction on the message that started the -turn. The gateway decides where the agent may post: a ``forbidden`` result of -a send is a refusal, not a transport failure. +conversation's latest accepted event came on, or on the only connection of +the provider that supports the operation. :meth:`send` posts CommonMark text +in parts of at most the connection's advisory ``text_characters``. Streaming +uses a placeholder message, edits paced by ``edit_interval_millis``, and a +delete. The typing indicator is an ``eyes`` reaction on the message that +started the turn. The gateway decides where the agent may post: a +``forbidden`` result of a send is a refusal, not a transport failure. Files go out through ``file_send``. The attachments of a message that starts a turn are read through ``file_read`` before the turn and use the @@ -109,6 +115,11 @@ truncate_text, valid_identifier, ) +from nerve.channels.hosted.providers import ( + STREAM_MODE_PARTIAL, + HostedChannelSettings, + HostedChatProvider, +) from nerve.channels.observation import ObservationPolicy if TYPE_CHECKING: @@ -265,7 +276,7 @@ class HostedChannel(BaseChannel): def __init__( self, - provider: str, + provider: HostedChatProvider, router: ChannelRouter, config: Callable[[], NerveConfig], *, @@ -276,7 +287,8 @@ def __init__( notifications: Any | None = None, button_style: Callable[[str], str] | None = None, ) -> None: - self._provider = provider + self._spec = provider + self._provider = provider.name self.router = router self._config = config self._max_inflight = max_inflight @@ -334,8 +346,7 @@ def capabilities(self) -> ChannelCapability: | ChannelCapability.TYPING_INDICATOR | ChannelCapability.SEND_FILES ) - section = getattr(self._config(), self._provider, None) - if getattr(section, "stream_mode", "partial") == "partial": + if self._settings().stream_mode == STREAM_MODE_PARTIAL: capabilities |= ChannelCapability.STREAMING return capabilities @@ -351,8 +362,10 @@ def constraints(self) -> ChannelConstraints: editable = True if self._operations is not None: streams = self._operations.streams - senders = streams.connections("send") - editable = senders <= streams.connections("edit") & streams.connections("delete") + senders = streams.connections("send", self._provider) + editable = senders <= ( + streams.connections("edit", self._provider) & streams.connections("delete", self._provider) + ) return ChannelConstraints( max_message_length=min( (item.text_characters for item in limits), default=DEFAULT_TEXT_CHARACTERS, @@ -365,19 +378,20 @@ def constraints(self) -> ChannelConstraints: @property def is_available(self) -> bool: - """Whether the provider is switched on and a stream advertises ``send`` now.""" + """Whether the provider is switched on and a stream advertises ``send`` for it now.""" return ( self._switched_on() and self._operations is not None - and bool(self._operations.streams.connections("send")) + and bool(self._operations.streams.connections("send", self._provider)) ) def _switched_on(self) -> bool: - """Whether the provider is enabled and in hosted mode, read per call.""" - section = getattr(self._config(), self._provider, None) - return bool( - section is not None and section.enabled and getattr(section, "mode", "") == "hosted", - ) + """Whether the provider is switched on, read per call.""" + return bool(self._spec.switched_on(self._config())) + + def _settings(self) -> HostedChannelSettings: + """The provider's channel settings, read per call.""" + return self._spec.settings(self._config()) async def start(self) -> None: self._running = True @@ -431,9 +445,9 @@ async def _connection(self, conversation_id: str, operation: str) -> uuid.UUID: """The connection for outbound work in *conversation_id*. That is the connection of the conversation's latest accepted event, - else the only connection that advertises *operation*, which may take - a short wait for a stream. Raises :class:`OperationFailed` when - neither exists. + else the only connection of the provider that advertises + *operation*, which may take a short wait for a stream. Raises + :class:`OperationFailed` when neither exists. """ if not conversation_id: raise OperationFailed(operation, "unavailable", "invalid_target", local=True, @@ -441,7 +455,9 @@ async def _connection(self, conversation_id: str, operation: str) -> uuid.UUID: connection = self._connections.get(conversation_id) if connection is not None: return connection - candidates = await self._operations.connections(operation) if self._operations else set() + candidates = ( + await self._operations.connections(operation, self._provider) if self._operations else set() + ) if len(candidates) == 1: return next(iter(candidates)) raise OperationFailed( @@ -465,7 +481,10 @@ def _send_limits(self) -> list[ConnectionLimits]: if self._operations is None: return [] streams = self._operations.streams - found = (streams.capabilities_for(connection) for connection in streams.connections("send")) + found = ( + streams.capabilities_for(connection) + for connection in streams.connections("send", self._provider) + ) return [capabilities.limits for capabilities in found if capabilities is not None] async def _perform(self, connection_id: uuid.UUID, kind: str, payload: Any, **options: Any): @@ -694,10 +713,10 @@ async def post_notification( The card goes with ``notify``, and the gateway posts it in the agent's notification channel, which an admin sets in the control plane. The config does not choose the conversation. Nerve sends the - card only when exactly one connection advertises ``notify``. Returns - ``(target, message_id)``, where the target is the card's - conversation, or ``None`` when there is no conversation or the post - fails. + card only when exactly one connection of the provider advertises + ``notify``. Returns ``(target, message_id)``, where the target is the + card's conversation, or ``None`` when there is no conversation or the + post fails. """ if not self.is_available: return None @@ -729,7 +748,7 @@ async def post_notification( return conversation, message_id def _notify_connection(self) -> uuid.UUID | None: - """The only connection that advertises ``notify`` now, else ``None``. + """The only connection of the provider that advertises ``notify`` now, else ``None``. It does not wait for a stream. The caller first makes sure that a stream can send, and a gateway without ``notify`` never advertises @@ -737,7 +756,9 @@ def _notify_connection(self) -> uuid.UUID | None: advertise it, because Nerve cannot tell which connection the card is for. """ - candidates = self._operations.streams.connections("notify") if self._operations else set() + candidates = ( + self._operations.streams.connections("notify", self._provider) if self._operations else set() + ) if len(candidates) > 1: raise OperationFailed( "notify", "unavailable", "invalid_target", local=True, @@ -772,8 +793,8 @@ async def authorize_outbound(self, target: str) -> Decision: """ if not self._switched_on(): return Decision(False, f"{self._provider} is switched off ({self._provider}.enabled is false)") - section = getattr(self._config(), self._provider, None) - if not getattr(section, "allow_outbound", False): + settings = self._settings() + if not settings.allow_outbound: return Decision( False, f"{self._provider}.allow_outbound is not enabled, so the agent may not post " @@ -784,8 +805,8 @@ async def authorize_outbound(self, target: str) -> Decision: return Decision(False, f"no {self._provider} conversation ID in the target") if self._is_id is not None and not self._is_id(conversation): return Decision(False, f"target must be a {self._provider} conversation ID, not a name") - allow = list(getattr(section, "allow_channels", None) or []) - deny = list(getattr(section, "deny_channels", None) or []) + allow = list(settings.allow_channels) + deny = list(settings.deny_channels) if allow or deny: gate = PatternGate("conversation", allow=allow, deny=deny) verdict = gate.check(self._identity(conversation, "", gate)) @@ -796,7 +817,7 @@ async def authorize_outbound(self, target: str) -> Decision: False, f"the destination is not approved by the {self._provider} channel policy", ) if self._connections.get(conversation) is None and self._operations is not None: - candidates = self._operations.streams.connections("send") + candidates = self._operations.streams.connections("send", self._provider) if len(candidates) > 1: return Decision( False, @@ -829,22 +850,32 @@ def remember_message(self, conversation_id: str, message_id: str, target: str, t # ------------------------------------------------------------------ # def can_accept(self) -> bool: - """Whether the channel runs, is still switched on, and has room for a turn. - - The provider section is read per call, so turning the provider off - by a reload stops intake until it is turned on again. Intake also - stops while the engine drains: the gateway keeps the events, and the - next start reads them. + """Whether the channel runs, holds no events, and has room for a turn. + + The provider's switch is read per call. Turning a provider that + holds its events while off (Slack) off by a reload stops intake + until it is turned on again. A provider that does not hold them + never stops intake by its switch: :meth:`deliver` rejects its + events. Intake also stops while the engine drains: the gateway keeps + the events, and the next start reads them. """ return ( self._running - and self._switched_on() + and (self._switched_on() or not self._spec.hold_while_off) and len(self._inflight) < self._max_inflight and self.router.accepting_turns ) async def deliver(self, event: Event) -> Disposition: - """Dispatch one admitted event and say what happened to it.""" + """Dispatch one admitted event and say what happened to it. + + While the provider is switched off, and it does not hold its events, + each event is rejected. + """ + if not self._spec.hold_while_off and not self._switched_on(): + return Disposition.rejected( + "configuration_invalid", f"{self._provider} is switched off ({self._provider}.enabled is false)", + ) purpose = event.admission.purpose if purpose == "invoke" and event.kind == "message": disposition = await self._invoke_message(event) @@ -1109,8 +1140,7 @@ def _note_handled(self, event: Event) -> None: # ------------------------------------------------------------------ # def _source(self) -> ChannelSourceConfig | None: - section = getattr(self._config(), self._provider, None) - return getattr(section, "source", None) + return self._settings().source def _identity(self, id_: str, display_name: str, gate: PatternGate) -> Identity: """An identity whose display name can deny but never grant. diff --git a/nerve/channels/hosted/manager.py b/nerve/channels/hosted/manager.py index 9f59d5812..9588f9978 100644 --- a/nerve/channels/hosted/manager.py +++ b/nerve/channels/hosted/manager.py @@ -223,13 +223,19 @@ def operation_stream( return stream return None - def connections(self, operation: str) -> set[uuid.UUID]: - """The connections that some ready stream advertised *operation* for.""" + def connections(self, operation: str, provider: str | None = None) -> set[uuid.UUID]: + """The connections that some ready stream advertised *operation* for. + + With *provider*, only the connections that serve that provider's + hosted chat channel: those whose capabilities name it, and those + whose capabilities name no provider. + """ return { connection_id for stream in self.streams if stream.ready - for connection_id in stream.capabilities + for connection_id, capabilities in stream.capabilities.items() if stream.supports(connection_id, operation) + and (provider is None or capabilities.serves(provider)) } async def wait_for_change(self, timeout: float) -> None: diff --git a/nerve/channels/hosted/operations.py b/nerve/channels/hosted/operations.py index 665e076ba..09cebae0f 100644 --- a/nerve/channels/hosted/operations.py +++ b/nerve/channels/hosted/operations.py @@ -246,12 +246,16 @@ async def _run( continue raise OperationFailed.from_result(result) - async def connections(self, kind: str) -> set[uuid.UUID]: - """The connections that advertise *kind*, after up to ``stream_wait`` for one.""" + async def connections(self, kind: str, provider: str | None = None) -> set[uuid.UUID]: + """The connections that advertise *kind*, after up to ``stream_wait`` for one. + + With *provider*, only the connections that serve that provider (see + :meth:`StreamManager.connections`). + """ loop = asyncio.get_running_loop() deadline = loop.time() + self._timing.stream_wait while True: - found = self._streams.connections(kind) + found = self._streams.connections(kind, provider) remaining = deadline - loop.time() if found or remaining <= 0: return found diff --git a/nerve/channels/hosted/providers.py b/nerve/channels/hosted/providers.py new file mode 100644 index 000000000..137e9ba02 --- /dev/null +++ b/nerve/channels/hosted/providers.py @@ -0,0 +1,159 @@ +"""The hosted chat providers: the providers that a hosted channel can serve. + +Each :class:`HostedChatProvider` in :data:`HOSTED_CHAT_PROVIDERS` names a +provider, says when the config puts it in hosted mode, reads its switch, and +gives the local settings of its channel. The runtime builds one +:class:`~nerve.channels.hosted.channel.HostedChannel` for each provider that +the config puts in hosted mode, in table order. A provider that the config +does not put in hosted mode has no channel, and the inbox reader rejects its +events. + +A provider needs no Slack-shaped config section. A setting that its config +does not have gets the default of :class:`HostedChannelSettings`. + +A provider that is switched off does one of two things with its events. The +gateway keeps the events of every provider in one ordered inbox, so the two +choices are to hold all of them or to refuse those of the provider: + +- With ``hold_while_off``, its channel takes no events, and the inbox reader + stops reading until a reload switches the provider on. The events wait in + the gateway inbox, also those of the other providers. Slack does this. +- Without it, which is the default, its channel rejects each of its events, + and the reader continues with the events of the other providers. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Callable + +if TYPE_CHECKING: + from nerve.config import ChannelSourceConfig, NerveConfig + + +# Answers stream into a placeholder message (``partial``) or are posted when +# they are complete (``full``). +STREAM_MODE_PARTIAL = "partial" + + +@dataclass(frozen=True) +class HostedChannelSettings: + """The local settings of one hosted chat channel. + + ``stream_mode`` ``partial`` streams an answer into a placeholder message, + and ``full`` posts it when it is complete. ``allow_outbound`` lets the + agent post to a conversation that it names, which must also pass + ``allow_channels`` and ``deny_channels`` when they are set. ``source`` is + the grant of observed messages; without it, observations are rejected. + """ + + stream_mode: str = STREAM_MODE_PARTIAL + allow_outbound: bool = False + allow_channels: tuple[str, ...] = () + deny_channels: tuple[str, ...] = () + source: ChannelSourceConfig | None = None + + +def _no_settings(config: NerveConfig) -> HostedChannelSettings: + return HostedChannelSettings() + + +def _no_options() -> dict[str, Any]: + return {} + + +@dataclass(frozen=True) +class HostedChatProvider: + """One chat provider that a hosted channel can serve. + + ``name`` is the provider token of its events and connections, and the + name of its channel. ``hosted`` says whether the config puts the + provider in hosted mode; the runtime reads it once, at startup. + ``switched_on`` and ``settings`` are read for each use, so they follow a + reload. ``hold_while_off`` says what the channel does with events while + the provider is switched off (see the module docstring). + ``channel_options`` gives the provider's presentation rules, as keyword + arguments of :class:`~nerve.channels.hosted.channel.HostedChannel`: + ``is_id``, ``reaction_names`` and ``button_style``. A provider without + them gets the channel's defaults. + """ + + name: str + hosted: Callable[[NerveConfig], bool] + switched_on: Callable[[NerveConfig], bool] + settings: Callable[[NerveConfig], HostedChannelSettings] = _no_settings + hold_while_off: bool = False + channel_options: Callable[[], dict[str, Any]] = _no_options + + +# ---------------------------------------------------------------------- # +# Slack # +# ---------------------------------------------------------------------- # + + +def _slack_hosted(config: NerveConfig) -> bool: + return config.slack.mode == "hosted" + + +def _slack_switched_on(config: NerveConfig) -> bool: + return bool(config.slack.enabled and config.slack.mode == "hosted") + + +def _slack_settings(config: NerveConfig) -> HostedChannelSettings: + slack = config.slack + return HostedChannelSettings( + stream_mode=slack.stream_mode, + allow_outbound=slack.allow_outbound, + allow_channels=tuple(slack.allow_channels or ()), + deny_channels=tuple(slack.deny_channels or ()), + source=slack.source, + ) + + +def _slack_channel_options() -> dict[str, Any]: + """Slack IDs, short reaction names, and the button styles of notification cards. + + The config does not choose the conversation of a card: the gateway posts + it in the agent's notification channel, which an admin sets in the + control plane. + """ + from nerve.channels.slack import is_slack_id + from nerve.channels.slack_presentation import approval_style, slack_emoji_by_name + + return { + "is_id": is_slack_id, + "reaction_names": slack_emoji_by_name(), + "button_style": approval_style, + } + + +SLACK = HostedChatProvider( + name="slack", + hosted=_slack_hosted, + switched_on=_slack_switched_on, + settings=_slack_settings, + hold_while_off=True, + channel_options=_slack_channel_options, +) + +# Every hosted chat provider, in the order of their channels. +HOSTED_CHAT_PROVIDERS: tuple[HostedChatProvider, ...] = (SLACK,) + + +def hosted_chat_providers(config: NerveConfig) -> list[HostedChatProvider]: + """The providers that *config* puts in hosted mode, in table order. + + A provider that is switched off is included: its channel reads the + switch for each event. + """ + return [provider for provider in HOSTED_CHAT_PROVIDERS if provider.hosted(config)] + + +__all__ = [ + "HOSTED_CHAT_PROVIDERS", + "SLACK", + "STREAM_MODE_PARTIAL", + "HostedChannelSettings", + "HostedChatProvider", + "hosted_chat_providers", +] diff --git a/nerve/channels/hosted/runtime.py b/nerve/channels/hosted/runtime.py index 70c238635..4871c3706 100644 --- a/nerve/channels/hosted/runtime.py +++ b/nerve/channels/hosted/runtime.py @@ -2,9 +2,9 @@ The runtime owns the token verifier, the stream manager, the operation runner, the inbox reader, one :class:`HostedChannel` for each chat provider in -hosted mode, and one :class:`HostedSourceConsumer` for each hosted source. The -hosted settings and the provider modes are read once at startup; a change -needs a restart. A reload reads the gateway key file again and publishes the +hosted mode (see :mod:`nerve.channels.hosted.providers`), and one +:class:`HostedSourceConsumer` for each hosted source. The hosted settings and +the provider modes are read once at startup; a change needs a restart. A reload reads the gateway key file again and publishes the source subscriptions again. The source grant of each chat provider is read per event, so it follows a reload. """ @@ -26,6 +26,7 @@ from nerve.channels.hosted.intake import EventConsumer, InboxReader, ReaderSettings from nerve.channels.hosted.manager import StreamManager from nerve.channels.hosted.operations import OperationRunner, OperationTiming +from nerve.channels.hosted.providers import hosted_chat_providers from nerve.channels.hosted.sources import ( HostedSourceConsumer, hosted_sources, @@ -56,11 +57,13 @@ def hosted_channel_providers(config: NerveConfig) -> list[str]: """The chat providers whose traffic comes from the gateway. - A provider that is switched off is included. Its channel reads - ``enabled`` for each event, so intake stays paused until a reload - switches the provider on. + A provider that is switched off is included. Its channel reads the + provider's switch for each event: a provider that holds its events + while off (Slack) pauses intake until a reload switches it on, and any + other provider rejects its events (see + :mod:`nerve.channels.hosted.providers`). """ - return ["slack"] if config.slack.mode == "hosted" else [] + return [provider.name for provider in hosted_chat_providers(config)] def hosted_providers(config: NerveConfig) -> list[str]: @@ -96,7 +99,7 @@ def __init__( problems = settings.problems() if problems: raise ValueError("; ".join(problems)) - providers = hosted_channel_providers(config) + providers = hosted_chat_providers(config) source_providers = hosted_sources(config) if not providers and not source_providers: raise ValueError("no provider is in hosted mode") @@ -126,13 +129,11 @@ def __init__( ) self.operations = OperationRunner(self.streams, operation_timing) self.channels: dict[str, HostedChannel] = { - provider: HostedChannel( + provider.name: HostedChannel( provider, router, config_getter, - is_id=_identifier_test(provider), operations=self.operations, - reaction_names=_reaction_names(provider), notifications=notification_service, - **_notification_rules(provider), + **provider.channel_options(), ) for provider in providers } @@ -219,38 +220,6 @@ def _stream_nudged(self, stream: ChannelStream, purpose: str) -> None: self.reader.nudge(purpose) -def _identifier_test(provider: str) -> Callable[[str], bool] | None: - """How a provider tells a literal ID from a name in a source pattern.""" - if provider == "slack": - from nerve.channels.slack import is_slack_id - - return is_slack_id - return None - - -def _notification_rules(provider: str) -> dict[str, Any]: - """The button styles of a provider's notification cards. - - The config does not choose the conversation of a card: the gateway posts - it in the agent's notification channel, which an admin sets in the - control plane. - """ - if provider == "slack": - from nerve.channels.slack_presentation import approval_style - - return {"button_style": approval_style} - return {} - - -def _reaction_names(provider: str) -> dict[str, str]: - """A provider's short reaction names and their emoji.""" - if provider == "slack": - from nerve.channels.slack_presentation import slack_emoji_by_name - - return slack_emoji_by_name() - return {} - - __all__ = [ "RECEIVE_LIMITS", "HostedChannelRuntime", diff --git a/tests/fake_channel_gateway.py b/tests/fake_channel_gateway.py index 7eb49e0ca..71b91ea0e 100644 --- a/tests/fake_channel_gateway.py +++ b/tests/fake_channel_gateway.py @@ -116,12 +116,26 @@ def gateway_negotiation(**limits: int) -> dict[str, Any]: } -def capabilities(*, operations: list[str] | None = None, **limits: int) -> dict[str, Any]: - """The sample capabilities, with other operations or limits if given.""" +def capabilities( + *, + operations: list[str] | None = None, + provider: str | None = None, + behaviour: dict[str, str] | None = None, + **limits: int, +) -> dict[str, Any]: + """The sample capabilities, with other operations, limits, provider, or behaviour if given. + + Without a provider and a behaviour, the capabilities are those of a + gateway that announces neither. + """ body = copy.deepcopy(CAPABILITIES) if operations is not None: body["operations"] = operations body["limits"].update(limits) + if provider is not None: + body["provider"] = provider + if behaviour is not None: + body["behaviour"] = behaviour return body @@ -138,6 +152,7 @@ def message_event( event_id: str | None = None, connection_id: str = CONNECTION_ID, kind: str = "message", + provider: str = "slack", ) -> dict[str, Any]: """A message event in the shape of the contract samples, without a delivery. @@ -145,6 +160,7 @@ def message_event( as the gateway sets it. """ event = copy.deepcopy(MESSAGE_EVENT) + event["provider"] = provider event["kind"] = kind event["event_id"] = event_id or f"ev-{purpose}-{conversation}-{message_id}-{kind}" event["message_revision_id"] = f"revision:{message_id}:0" @@ -700,10 +716,13 @@ async def advertise( *, connection_id: str = CONNECTION_ID, operations: list[str] | None = None, + provider: str | None = None, + behaviour: dict[str, str] | None = None, **limits: int, ) -> None: await self.send( - "capabilities", capabilities(operations=operations, **limits), + "capabilities", + capabilities(operations=operations, provider=provider, behaviour=behaviour, **limits), connection_id=connection_id, ) diff --git a/tests/test_hosted_channel_behaviour.py b/tests/test_hosted_channel_behaviour.py new file mode 100644 index 000000000..f8ce4ce89 --- /dev/null +++ b/tests/test_hosted_channel_behaviour.py @@ -0,0 +1,280 @@ +"""Hosted chat providers other than Slack, and the behaviour of a connection. + +A second provider, ``fixture_chat``, joins Slack in the table of hosted chat +providers. It has no config section and no settings of its own. The harness +is the one of ``test_hosted_channel``: the fake gateway against the real +stream endpoint, with a real router, session manager, and database, and a +stand-in for the agent turn. + +Covered here: the table and its defaults, a switched-off provider that +refuses its events while Slack goes on, and channels that use only the +connections of their own provider. +""" + +from __future__ import annotations + +import asyncio +import uuid +from dataclasses import dataclass +from typing import Any + +import pytest +import pytest_asyncio + +from nerve.channels.base import ChannelCapability, OutboundMessage +from nerve.channels.hosted import providers as providers_module +from nerve.channels.hosted.providers import ( + HOSTED_CHAT_PROVIDERS, + SLACK, + HostedChannelSettings, + HostedChatProvider, +) +from nerve.channels.hosted.runtime import hosted_channel_providers +from nerve.config import NerveConfig, SlackConfig + +from tests.fake_channel_gateway import CONNECTION_ID, message_event +from tests.test_hosted_channel import Hosted, start_hosted, stop_hosted + +CHAT = "fixture_chat" +CHAT_CONNECTION = "0d7c4b1e-6a3f-4e92-8b5d-2f1a9c7e3b60" +SLACK_BEHAVIOUR = {"answer": "post", "session": "rotate"} +OPERATIONS = ["send", "edit", "delete", "reaction", "notify", "file_send"] + + +@dataclass +class Switch: + """The switch of the test provider, which a test turns as a reload would.""" + + on: bool = True + hosted: bool = True + + +def chat_provider(switch: Switch, **fields: Any) -> HostedChatProvider: + """A provider without a config section: only a name, a mode, and a switch.""" + return HostedChatProvider( + name=CHAT, + hosted=lambda config: switch.hosted, + switched_on=lambda config: switch.on, + **fields, + ) + + +def chat_event(**changes: Any) -> dict[str, Any]: + """A direct message of the test provider on its own connection.""" + values: dict[str, Any] = { + "provider": CHAT, "conversation": "room-1", "conversation_kind": "direct", + "message_id": "m-1", "connection_id": CHAT_CONNECTION, "author": "member-1", + } + values.update(changes) + return message_event(**values) + + +def slack_event(message_id: str = "1700000100.000100", **changes: Any) -> dict[str, Any]: + return message_event(conversation="D0DIRECT", conversation_kind="direct", message_id=message_id, **changes) + + +async def start_two(tmp_path, db, monkeypatch, switch: Switch, **fields: Any) -> Hosted: + """The hosted harness with Slack and the test provider in the table.""" + monkeypatch.setattr(providers_module, "HOSTED_CHAT_PROVIDERS", (SLACK, chat_provider(switch, **fields))) + return await start_hosted(tmp_path, db, monkeypatch) + + +@pytest.fixture +def switch() -> Switch: + return Switch() + + +@pytest_asyncio.fixture +async def two(tmp_path, db, monkeypatch, switch): + harness = await start_two(tmp_path, db, monkeypatch, switch) + try: + yield harness + finally: + await stop_hosted(harness) + + +async def announce(stream, provider: str | None, connection_id: str, **behaviour: str) -> None: + await stream.advertise( + connection_id=connection_id, provider=provider, behaviour=behaviour or None, operations=OPERATIONS, + ) + + +# ---------------------------------------------------------------------- # +# The table of hosted chat providers # +# ---------------------------------------------------------------------- # + + +class TestProviderTable: + def test_slack_is_the_only_provider_and_holds_its_events_while_off(self): + assert HOSTED_CHAT_PROVIDERS == (SLACK,) + assert SLACK.hold_while_off + + def test_a_provider_is_hosted_as_its_config_says(self): + config = NerveConfig() + assert hosted_channel_providers(config) == [] + config.slack = SlackConfig(mode="hosted", enabled=False) + assert hosted_channel_providers(config) == ["slack"] + + def test_slack_settings_come_from_its_section(self): + config = NerveConfig() + config.slack = SlackConfig( + mode="hosted", enabled=True, stream_mode="full", allow_outbound=True, + allow_channels=["C0123ABCD"], deny_channels=["C0DENIED"], + ) + + settings = SLACK.settings(config) + + assert (settings.stream_mode, settings.allow_outbound) == ("full", True) + assert (settings.allow_channels, settings.deny_channels) == (("C0123ABCD",), ("C0DENIED",)) + assert settings.source is config.slack.source + assert SLACK.switched_on(config) + config.slack.enabled = False + assert not SLACK.switched_on(config) + + def test_a_new_provider_refuses_its_events_while_off_and_has_default_settings(self, switch): + provider = chat_provider(switch) + + assert not provider.hold_while_off + assert provider.settings(NerveConfig()) == HostedChannelSettings() + assert provider.channel_options() == {} + + +@pytest.mark.asyncio +class TestProviderWithoutSettings: + async def test_its_channel_takes_a_turn_with_the_default_settings(self, two): + stream = await two.stream() + inbox_id = two.gateway.store(chat_event(text="hello from the test provider")) + + await stream.nudge("invoke") + + assert (await two.acknowledged(inbox_id))[inbox_id] == "accepted" + await two.turns.wait_for(1) + mapping = await two.db.get_channel_session(f"{CHAT}:room-1") + run = two.turns.runs[0] + assert (run["session_id"], run["channel"], run["user_message"]) == ( + mapping["session_id"], CHAT, "hello from the test provider", + ) + channel = two.runtime.channels[CHAT] + assert two.router.get_channel(CHAT) is channel + assert ChannelCapability.STREAMING in channel.capabilities + + async def test_it_refuses_addressed_delivery_and_observations(self, two): + stream = await two.stream() + channel = two.runtime.channels[CHAT] + inbox_id = two.gateway.store(chat_event(purpose="observe", conversation_kind="channel")) + + await stream.nudge("observe") + + assert (await two.acknowledged(inbox_id))[inbox_id] == "rejected" + assert two.gateway.acks[-1][0]["reason_code"] == "policy_denied" + verdict = await channel.authorize_outbound("room-1") + assert not verdict.allowed + assert "allow_outbound" in verdict.reason + + async def test_switched_off_it_refuses_its_events_and_slack_goes_on(self, two, switch): + switch.on = False + stream = await two.stream() + refused = two.gateway.store(chat_event()) + taken = two.gateway.store(slack_event()) + + await stream.nudge("invoke") + + assert await two.acknowledged(refused, taken) == {refused: "rejected", taken: "accepted"} + assert two.gateway.acks[-1][0]["reason_code"] == "configuration_invalid" + await two.turns.wait_for(1) + assert [run["channel"] for run in two.turns.runs] == ["slack"] + assert two.runtime.channels[CHAT].can_accept() + + switch.on = True + again = two.gateway.store(chat_event(message_id="m-2")) + await stream.nudge("invoke") + assert (await two.acknowledged(again))[again] == "accepted" + + async def test_a_switched_off_slack_holds_its_events_and_the_test_providers_too(self, two): + two.config.slack.enabled = False + stream = await two.stream() + held = two.gateway.store(chat_event()) + + await stream.nudge("invoke") + await asyncio.sleep(0.3) + + assert two.gateway.reads == [] + two.config.slack.enabled = True + assert (await two.acknowledged(held))[held] == "accepted" + + async def test_a_provider_that_the_config_does_not_host_has_no_channel(self, tmp_path, db, monkeypatch): + harness = await start_two(tmp_path, db, monkeypatch, Switch(hosted=False)) + try: + stream = await harness.stream() + refused = harness.gateway.store(chat_event()) + taken = harness.gateway.store(slack_event()) + + await stream.nudge("invoke") + + assert await harness.acknowledged(refused, taken) == {refused: "rejected", taken: "accepted"} + assert harness.gateway.acks[-1][0]["reason_code"] == "configuration_invalid" + assert list(harness.runtime.channels) == ["slack"] + finally: + await stop_hosted(harness) + + +# ---------------------------------------------------------------------- # +# Connections of a provider # +# ---------------------------------------------------------------------- # + + +@pytest.mark.asyncio +class TestProviderScoping: + async def test_each_channel_uses_only_the_connections_of_its_provider(self, two): + two.config.slack.allow_outbound = True + stream = await two.stream(advertise=False) + await announce(stream, "slack", CONNECTION_ID, **SLACK_BEHAVIOUR) + await stream.advertise( + connection_id=CHAT_CONNECTION, provider=CHAT, operations=["send"], text_characters=500, + ) + streams = two.runtime.streams + await two.gateway.wait_for(lambda: len(streams.connections("send")) == 2) + slack, chat = two.runtime.channels["slack"], two.runtime.channels[CHAT] + + assert streams.connections("send", "slack") == {uuid.UUID(CONNECTION_ID)} + assert streams.connections("send", CHAT) == {uuid.UUID(CHAT_CONNECTION)} + assert (slack.constraints.max_message_length, chat.constraints.max_message_length) == (4000, 500) + assert slack.constraints.supports_message_edit and not chat.constraints.supports_message_edit + assert slack.is_available and chat.is_available + assert (await slack.authorize_outbound("C0123ABCD")).allowed + + await slack.send(OutboundMessage(target="C0123ABCD", text="to slack")) + await chat.send(OutboundMessage(target="room-9", text="to the test provider")) + assert [(record["connection_id"], record["kind"]) for record in two.gateway.operations] == [ + (CONNECTION_ID, "send"), (CHAT_CONNECTION, "send"), + ] + + async def test_notify_goes_on_the_only_connection_of_the_provider(self, two): + stream = await two.stream(advertise=False) + await announce(stream, "slack", CONNECTION_ID, **SLACK_BEHAVIOUR) + await announce(stream, CHAT, CHAT_CONNECTION) + await two.gateway.wait_for(lambda: len(two.runtime.streams.connections("notify")) == 2) + + posted = await two.runtime.channels["slack"].post_notification("n-1", "Deploy now?", [("Yes", "yes")]) + + assert posted is not None + [record] = [record for record in two.gateway.operations if record["kind"] == "notify"] + assert record["connection_id"] == CONNECTION_ID + + async def test_a_connection_that_names_no_provider_serves_every_channel(self, two): + stream = await two.stream(advertise=False) + await announce(stream, None, CONNECTION_ID) + await two.gateway.wait_for(lambda: two.runtime.streams.connections("send")) + + assert two.runtime.streams.connections("send", "slack") == {uuid.UUID(CONNECTION_ID)} + assert two.runtime.streams.connections("send", CHAT) == {uuid.UUID(CONNECTION_ID)} + assert two.runtime.channels[CHAT].is_available + + async def test_a_connection_of_another_provider_does_not_make_a_channel_available(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION) + await two.gateway.wait_for(lambda: two.runtime.streams.connections("send")) + + assert two.runtime.channels[CHAT].is_available + assert not two.runtime.channels["slack"].is_available + assert (await two.runtime.channels["slack"].post_notification("n-2", "Deploy?", [("Yes", "yes")])) is None From 7d12db04a16d03bf6e772b891c357465a2b55ee5 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:30:09 +0000 Subject: [PATCH 03/10] Hosted channels: the tool answer mode posts nothing A provider such as email has no conversation that Nerve can post a turn's answer to: the agent answers only with a tool. The router always showed typing, streamed a placeholder, and posted the answer or an "Error:" message, so such a provider needed a way to stay silent. An inbound message now says whether its turn posts its answer. When it does not, the router sends nothing to the channel for the turn, and the answer stays in the session. A hosted channel takes this from the behaviour that the event's connection announced, and puts one generic line before the turn's text: where the message came from, that the answer goes to no one, and which tool can answer. A connection without an announced behaviour posts, so Slack does not change. Co-Authored-By: Claude Opus 5.5 --- nerve/channels/base.py | 6 + nerve/channels/hosted/channel.py | 51 +++++- nerve/channels/router.py | 60 ++++--- tests/test_hosted_channel_behaviour.py | 229 ++++++++++++++++++++++++- 4 files changed, 310 insertions(+), 36 deletions(-) diff --git a/nerve/channels/base.py b/nerve/channels/base.py index a1af4ef6f..85463bb69 100644 --- a/nerve/channels/base.py +++ b/nerve/channels/base.py @@ -54,6 +54,11 @@ class InboundMessage: turn as that person. A self-hosted channel has no mapping from a provider user to a person, so it leaves ``actor`` empty. A hosted channel sets the person that the channel gateway resolved. + + ``post_answer`` tells the router to show the turn in the channel. When it + is ``False``, the router sends nothing to the channel for the turn: no + typing indicator, no streamed or final answer, and no error message. The + answer stays in the session, and the agent can answer with a tool. """ channel_name: str # "telegram", "web", "discord" @@ -64,6 +69,7 @@ class InboundMessage: metadata: dict[str, Any] = field(default_factory=dict) actor: Actor | None = None conversation_title: str = "" # "general"; "" when unknown or direct + post_answer: bool = True @dataclass diff --git a/nerve/channels/hosted/channel.py b/nerve/channels/hosted/channel.py index 68ed26599..b782ada65 100644 --- a/nerve/channels/hosted/channel.py +++ b/nerve/channels/hosted/channel.py @@ -13,6 +13,14 @@ takes part in, a reaction, an interaction, or a command. A mention of the agent carries the ``self`` flag, so intake needs no capabilities. +The capabilities of the event's connection can give a behaviour (see +:class:`~nerve.channels.hosted.contract.model.ConnectionBehaviour`). With the +``tool`` answer mode, the router sends nothing to the conversation for the +turn, and the answer stays in the session. The turn's text then starts with +a line that says so and names the reply tool, when the connection gives one +(:func:`answer_note`). A connection without a behaviour, or whose +capabilities Nerve does not have, has the ``post`` answer mode. + An invoke event names the person that the gateway resolved from its author. In external mode, a message or reaction turn runs as that person, and an interaction answers its notification as that person. In local mode, the @@ -78,7 +86,7 @@ OutboundRefused, ) from nerve.channels.hosted.attachments import extract_attachments -from nerve.channels.hosted.contract import ContentPart, Event +from nerve.channels.hosted.contract import ConnectionBehaviour, ContentPart, Event from nerve.channels.hosted.contract.model import ( MAX_TRANSFER_BYTES, AttachmentReference, @@ -195,6 +203,16 @@ def describe_attachments(event: Event) -> str: ) +def answer_note(provider: str, reply_tool: str = "") -> str: + """The line before the text of a turn whose answer Nerve does not post. + + It names the provider, says that the answer in the session goes to no + one, and names the tool that can answer, when the connection gives one. + """ + how = f"To answer, use the {reply_tool} tool." if reply_tool else "This channel does not send answers." + return f"This message came through {provider}. Your answer in this session is not sent to anyone. {how}" + + def _why(error: Exception) -> str: """The log text of an error, with a failed operation's reason and detail.""" return getattr(error, "detailed", None) or str(error) @@ -1092,17 +1110,34 @@ async def _invoke_reaction(self, event: Event) -> Disposition: def _dispatch(self, message: InboundMessage, event: Event, *, attachments: bool = False) -> Disposition: """Start the turn as a task. Acceptance does not wait for the turn. - The task first resolves the event's person into the message's actor. + The behaviour of the event's connection goes with the message. The + task first resolves the event's person into the message's actor. With *attachments*, it also reads the event's attachments. """ if not self.can_accept(): return Disposition.deferred(f"{self._provider} has {len(self._inflight)} turns starting") - task = asyncio.create_task(self._run_turn(message, event, attachments)) + behaviour = self._behaviour(event) + message.post_answer = behaviour.posts_answer + task = asyncio.create_task(self._run_turn(message, event, attachments, behaviour)) self._inflight.add(task) task.add_done_callback(self._inflight.discard) return Disposition.accepted() - async def _run_turn(self, message: InboundMessage, event: Event, attachments: bool) -> None: + def _behaviour(self, event: Event) -> ConnectionBehaviour: + """The behaviour that the event's connection announced, else the defaults.""" + capabilities = ( + self._operations.streams.capabilities_for(event.connection_id) if self._operations else None + ) + return capabilities.behaviour if capabilities is not None else ConnectionBehaviour() + + async def _run_turn( + self, message: InboundMessage, event: Event, attachments: bool, behaviour: ConnectionBehaviour, + ) -> None: + """Run the turn of one message. + + When the connection does not post answers, :func:`answer_note` comes + first, before the attachments that were read and the text. + """ try: message.actor = await self.router.actor_for_principal( event.admission.invoke.resolved_principal_id, @@ -1110,11 +1145,15 @@ async def _run_turn(self, message: InboundMessage, event: Event, attachments: bo ) if attachments: await self._add_attachments(message, event) + if not behaviour.posts_answer: + note = answer_note(self._provider, behaviour.reply_tool) + message.text = f"{note}\n\n{message.text}" if message.text else note await self.router.handle_message(message) except asyncio.CancelledError: raise except NotAcceptingTurnsError: - # The stream adapter already sent the engine's error event. + # When the turn posts its answer, the stream adapter already sent + # the engine's error event. logger.info("Draining: refused a hosted %s turn for %s", self._provider, message.sender_id) except Exception as error: # noqa: BLE001 - one turn must not stop intake logger.error( @@ -1230,4 +1269,4 @@ async def _observe_message(self, event: Event) -> Disposition: return Disposition.accepted() -__all__ = ["HostedChannel", "describe_attachments", "render_content"] +__all__ = ["HostedChannel", "answer_note", "describe_attachments", "render_content"] diff --git a/nerve/channels/router.py b/nerve/channels/router.py index 639613cc2..68c76817d 100644 --- a/nerve/channels/router.py +++ b/nerve/channels/router.py @@ -199,6 +199,10 @@ async def handle_message(self, msg: InboundMessage) -> str: one actor, each consecutive run of messages from one actor is a separate turn, in arrival order. + A turn whose messages do not ask to post their answer + (``post_answer``) sends nothing to the channel: no typing + indicator, no streamed or final answer, and no error message. + Called by channel implementations when they receive a user message. """ channel = self._channels.get(msg.channel_name) @@ -278,17 +282,9 @@ async def _run_single( """Run the engine for a single message with streaming.""" self._note_message_context(session_id, msg) - if ChannelCapability.TYPING_INDICATOR in channel.capabilities: - try: - await channel.send_typing(msg.sender_id) - except Exception as e: - logger.debug( - "Typing indicator failed for %s: %s", msg.channel_name, e, - ) - - adapter = await self._setup_streaming( - channel, msg.sender_id, session_id, - ) + if msg.post_answer: + await self._show_typing(channel, msg.sender_id, msg.channel_name) + await self._setup_streaming(channel, msg.sender_id, session_id) images = msg.metadata.get("images") if msg.metadata else None task = asyncio.create_task( @@ -313,9 +309,10 @@ async def _run_single( return task.result() return "" finally: - await self._teardown_streaming( - channel.name, msg.sender_id, session_id, - ) + if msg.post_answer: + await self._teardown_streaming( + channel.name, msg.sender_id, session_id, + ) async def _run_batch( self, @@ -325,10 +322,12 @@ async def _run_batch( ) -> str: """Combine pending messages into one turn and run. - Every message in *pending* has the same actor. + Every message in *pending* has the same actor. The turn posts its + answer only when every message asks for it. """ last_msg = pending[-1][0] sender_id = last_msg.sender_id + post_answer = all(m.post_answer for m, _ in pending) # Combine texts (each message on its own line) combined_text = "\n\n".join(m.text for m, _ in pending) @@ -343,16 +342,9 @@ async def _run_batch( # Update reaction context to the last message in the batch self._note_message_context(session_id, last_msg) - if ChannelCapability.TYPING_INDICATOR in channel.capabilities: - try: - await channel.send_typing(sender_id) - except Exception as e: - logger.debug( - "Typing indicator failed for %s: %s", - last_msg.channel_name, e, - ) - - adapter = await self._setup_streaming(channel, sender_id, session_id) + if post_answer: + await self._show_typing(channel, sender_id, last_msg.channel_name) + await self._setup_streaming(channel, sender_id, session_id) task = asyncio.create_task( self.engine.run( @@ -373,8 +365,22 @@ async def _run_batch( return task.result() return "" finally: - await self._teardown_streaming( - channel.name, sender_id, session_id, + if post_answer: + await self._teardown_streaming( + channel.name, sender_id, session_id, + ) + + async def _show_typing( + self, channel: BaseChannel, target: str, channel_name: str, + ) -> None: + """Show the typing indicator when the channel has one. A failure is only logged.""" + if ChannelCapability.TYPING_INDICATOR not in channel.capabilities: + return + try: + await channel.send_typing(target) + except Exception as e: + logger.debug( + "Typing indicator failed for %s: %s", channel_name, e, ) def _note_message_context(self, session_id: str, msg: InboundMessage) -> None: diff --git a/tests/test_hosted_channel_behaviour.py b/tests/test_hosted_channel_behaviour.py index f8ce4ce89..8f1cfa1e9 100644 --- a/tests/test_hosted_channel_behaviour.py +++ b/tests/test_hosted_channel_behaviour.py @@ -7,8 +7,11 @@ stand-in for the agent turn. Covered here: the table and its defaults, a switched-off provider that -refuses its events while Slack goes on, and channels that use only the -connections of their own provider. +refuses its events while Slack goes on, channels that use only the +connections of their own provider, and the ``tool`` answer mode, in which +the router sends nothing to the conversation and the turn's text starts with +a note for the agent. The router's part is also tested on its own, with a +channel that records every call. """ from __future__ import annotations @@ -18,10 +21,21 @@ from dataclasses import dataclass from typing import Any +from unittest.mock import MagicMock + import pytest import pytest_asyncio -from nerve.channels.base import ChannelCapability, OutboundMessage +from nerve.agent.sessions import SessionManager +from nerve.agent.streaming import broadcaster +from nerve.channels.base import ( + BaseChannel, + ChannelCapability, + ChannelConstraints, + InboundMessage, + OutboundMessage, +) +from nerve.channels.hosted.channel import answer_note from nerve.channels.hosted import providers as providers_module from nerve.channels.hosted.providers import ( HOSTED_CHAT_PROVIDERS, @@ -30,6 +44,7 @@ HostedChatProvider, ) from nerve.channels.hosted.runtime import hosted_channel_providers +from nerve.channels.router import ChannelRouter from nerve.config import NerveConfig, SlackConfig from tests.fake_channel_gateway import CONNECTION_ID, message_event @@ -278,3 +293,211 @@ async def test_a_connection_of_another_provider_does_not_make_a_channel_availabl assert two.runtime.channels[CHAT].is_available assert not two.runtime.channels["slack"].is_available assert (await two.runtime.channels["slack"].post_notification("n-2", "Deploy?", [("Yes", "yes")])) is None + + +# ---------------------------------------------------------------------- # +# The answer mode # +# ---------------------------------------------------------------------- # + + +class StreamingTurns: + """An agent turn that streams an answer, or an error, into its session.""" + + def __init__(self, *, error: str | None = None) -> None: + self.runs: list[dict[str, Any]] = [] + self.error = error + + async def run(self, **kwargs: Any) -> str: + self.runs.append(kwargs) + session_id = kwargs["session_id"] + if self.error is not None: + await broadcaster.broadcast_error(session_id, self.error) + return "" + for word in ("The ", "report ", "is ", "fine."): + await broadcaster.broadcast_token(session_id, word) + await broadcaster.broadcast_done(session_id) + return "The report is fine." + + async def wait_for(self, count: int) -> None: + for _ in range(500): + if len(self.runs) >= count: + return + await asyncio.sleep(0.01) + raise AssertionError(f"expected {count} turns, saw {len(self.runs)}") + + +async def turn_of(hosted: Hosted, stream, event: dict[str, Any], turns: StreamingTurns) -> dict[str, Any]: + """Deliver *event*, wait for its turn, and give the turn's settled output time to arrive.""" + hosted.router.engine.run = turns.run + inbox_id = hosted.gateway.store(event) + await stream.nudge("invoke") + assert (await hosted.acknowledged(inbox_id))[inbox_id] == "accepted" + await turns.wait_for(len(turns.runs) + 1) + await asyncio.sleep(0.3) + return turns.runs[-1] + + +def kinds(hosted: Hosted) -> list[str]: + return [record["kind"] for record in hosted.gateway.operations] + + +@pytest.mark.asyncio +class TestToolAnswer: + async def test_the_turn_posts_nothing_and_its_text_starts_with_the_note(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", reply_tool="send_reply", session="rotate") + turns = StreamingTurns() + + run = await turn_of(two, stream, chat_event(text="Please check the report."), turns) + + assert kinds(two) == [] + assert run["user_message"] == ( + "This message came through fixture_chat. Your answer in this session is not sent to anyone. " + "To answer, use the send_reply tool.\n\nPlease check the report." + ) + mapping = await two.db.get_channel_session(f"{CHAT}:room-1") + assert run["session_id"] == mapping["session_id"] + + async def test_without_a_reply_tool_the_note_says_that_nothing_answers(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool") + + run = await turn_of(two, stream, chat_event(text="FYI"), StreamingTurns()) + + assert kinds(two) == [] + assert run["user_message"] == f"{answer_note(CHAT)}\n\nFYI" + assert answer_note(CHAT).endswith("This channel does not send answers.") + + async def test_an_error_of_the_turn_is_not_posted(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool") + + await turn_of(two, stream, chat_event(), StreamingTurns(error="the model is unavailable")) + + assert kinds(two) == [] + + async def test_the_same_provider_with_the_post_mode_posts_the_answer(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="post", session="rotate") + + run = await turn_of(two, stream, chat_event(text="Please check the report."), StreamingTurns()) + + assert run["user_message"] == "Please check the report." + assert "send" in kinds(two) and "reaction" in kinds(two) + final = two.gateway.sent_operations("send")[-1] + assert final["content"][0]["text"]["body"] == "The report is fine." + + @pytest.mark.parametrize("behaviour", [None, SLACK_BEHAVIOUR], ids=["absent", "explicit"]) + async def test_slack_posts_its_answer_as_before(self, two, behaviour): + stream = await two.stream(advertise=False) + await announce(stream, "slack", CONNECTION_ID, **(behaviour or {})) + + run = await turn_of(two, stream, slack_event(text="how are the deploys?"), StreamingTurns()) + + assert run["user_message"] == "how are the deploys?" + assert {"reaction", "send", "delete"} <= set(kinds(two)) + assert two.gateway.sent_operations("send")[-1]["content"][0]["text"]["body"] == "The report is fine." + + async def test_a_connection_without_capabilities_posts_its_answer(self, two): + stream = await two.stream(advertise=False) + + run = await turn_of(two, stream, chat_event(text="hello"), StreamingTurns()) + + assert run["user_message"] == "hello" + + +class RecordingChannel(BaseChannel): + """A channel that streams and shows typing, and records every call.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, ...]] = [] + + @property + def name(self) -> str: + return "recording" + + @property + def capabilities(self) -> ChannelCapability: + return ChannelCapability.SEND_TEXT | ChannelCapability.STREAMING | ChannelCapability.TYPING_INDICATOR + + @property + def constraints(self) -> ChannelConstraints: + return ChannelConstraints(max_message_length=4000, supports_message_edit=True) + + async def start(self) -> None: + pass + + async def stop(self) -> None: + pass + + async def send(self, message: OutboundMessage) -> None: + self.calls.append(("send", message.text)) + + async def send_placeholder(self, target: str, session_id: str) -> str | None: + self.calls.append(("placeholder", target)) + return "placeholder-1" + + async def edit_message(self, target: str, message_id: str, text: str, *, throttle: bool = False) -> None: + self.calls.append(("edit", text)) + + async def delete_message(self, target: str, message_id: str) -> None: + self.calls.append(("delete", message_id)) + + async def send_typing(self, target: str) -> None: + self.calls.append(("typing", target)) + + +def recording_router(db, turns: StreamingTurns) -> tuple[ChannelRouter, RecordingChannel]: + engine = MagicMock() + engine.db = db + engine.sessions = SessionManager(db) + engine.run = turns.run + router = ChannelRouter(engine) + router.BATCH_DEBOUNCE = 0.0 + channel = RecordingChannel() + router.register(channel) + return router, channel + + +def inbound(text: str = "hello", *, post_answer: bool = True) -> InboundMessage: + return InboundMessage( + channel_name="recording", channel_key="recording:room-1", sender_id="room-1", text=text, + post_answer=post_answer, + ) + + +@pytest.mark.asyncio +class TestRouterAnswerMode: + async def test_a_turn_that_posts_shows_typing_streams_and_sends(self, db): + router, channel = recording_router(db, StreamingTurns()) + + assert await router.handle_message(inbound()) == "The report is fine." + + assert {"typing", "placeholder", "send", "delete"} <= {call[0] for call in channel.calls} + + @pytest.mark.parametrize("error", [None, "the model is unavailable"], ids=["answer", "error"]) + async def test_a_turn_that_does_not_post_sends_nothing_to_the_channel(self, db, error): + turns = StreamingTurns(error=error) + router, channel = recording_router(db, turns) + + await router.handle_message(inbound(post_answer=False)) + + assert channel.calls == [] + [run] = turns.runs + assert run["user_message"] == "hello" + assert await db.get_channel_session("recording:room-1") is not None + + async def test_a_batch_posts_only_when_every_message_asks_for_it(self, db): + turns = StreamingTurns() + router, channel = recording_router(db, turns) + router.BATCH_DEBOUNCE = 0.1 + await router.create_session("recording:room-1", source="recording") + + await asyncio.gather( + router.handle_message(inbound("first", post_answer=False)), + router.handle_message(inbound("second")), + ) + + assert channel.calls == [] + [run] = turns.runs + assert run["user_message"] == "first\n\nsecond" From abed6e69b6f99885dcb54d5896a2a0e63fee5c56 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:31:22 +0000 Subject: [PATCH 04/10] Hosted channels: the keep session mode pins the last session A conversation of a provider such as email is one thread that can go quiet for days, and Nerve's sticky period moved its next message into a new session after two idle hours. The agent then lost the thread's context. A connection that announces the keep session mode now sends each turn of a conversation to that conversation's last session, however long it was idle, while the session is not archived or deleted. Rotate, and a connection that announces nothing, keep the sticky period. Co-Authored-By: Claude Opus 5.5 --- nerve/channels/hosted/channel.py | 13 ++++- tests/test_hosted_channel_behaviour.py | 80 +++++++++++++++++++++++++- 2 files changed, 87 insertions(+), 6 deletions(-) diff --git a/nerve/channels/hosted/channel.py b/nerve/channels/hosted/channel.py index b782ada65..025c62ac7 100644 --- a/nerve/channels/hosted/channel.py +++ b/nerve/channels/hosted/channel.py @@ -18,8 +18,11 @@ ``tool`` answer mode, the router sends nothing to the conversation for the turn, and the answer stays in the session. The turn's text then starts with a line that says so and names the reply tool, when the connection gives one -(:func:`answer_note`). A connection without a behaviour, or whose -capabilities Nerve does not have, has the ``post`` answer mode. +(:func:`answer_note`). With the ``keep`` session mode, a conversation +continues its last session however long it was idle, while that session is +not archived or deleted. A connection without a behaviour, or whose +capabilities Nerve does not have, has the ``post`` answer mode and the +``rotate`` session mode: the session changes after the sticky period. An invoke event names the person that the gateway resolved from its author. In external mode, a message or reaction turn runs as that person, and an @@ -1136,7 +1139,9 @@ async def _run_turn( """Run the turn of one message. When the connection does not post answers, :func:`answer_note` comes - first, before the attachments that were read and the text. + first, before the attachments that were read and the text. When the + connection keeps sessions, the turn goes to the conversation's last + session, if it still has one. """ try: message.actor = await self.router.actor_for_principal( @@ -1148,6 +1153,8 @@ async def _run_turn( if not behaviour.posts_answer: note = answer_note(self._provider, behaviour.reply_tool) message.text = f"{note}\n\n{message.text}" if message.text else note + if behaviour.keeps_session: + message.session_id = await self.router.get_last_session(message.channel_key) await self.router.handle_message(message) except asyncio.CancelledError: raise diff --git a/tests/test_hosted_channel_behaviour.py b/tests/test_hosted_channel_behaviour.py index 8f1cfa1e9..e99c715fc 100644 --- a/tests/test_hosted_channel_behaviour.py +++ b/tests/test_hosted_channel_behaviour.py @@ -8,9 +8,10 @@ Covered here: the table and its defaults, a switched-off provider that refuses its events while Slack goes on, channels that use only the -connections of their own provider, and the ``tool`` answer mode, in which -the router sends nothing to the conversation and the turn's text starts with -a note for the agent. The router's part is also tested on its own, with a +connections of their own provider, the ``tool`` answer mode, in which the +router sends nothing to the conversation and the turn's text starts with a +note for the agent, and the ``keep`` session mode, in which a conversation +keeps its last session past the sticky period. The router's part is also tested on its own, with a channel that records every call. """ @@ -501,3 +502,76 @@ async def test_a_batch_posts_only_when_every_message_asks_for_it(self, db): assert channel.calls == [] [run] = turns.runs assert run["user_message"] == "first\n\nsecond" + + +# ---------------------------------------------------------------------- # +# The session mode # +# ---------------------------------------------------------------------- # + + +# Long before any sticky period. +LONG_AGO = "2020-01-01T00:00:00+00:00" + + +async def idle_session(hosted: Hosted, channel_key: str, *, status: str = "idle") -> str: + """The conversation's session, idle since long ago.""" + session_id = await hosted.router.create_session(channel_key, source=CHAT) + await hosted.db.update_session_fields(session_id, {"status": status, "last_activity_at": LONG_AGO}) + return session_id + + +@pytest.mark.asyncio +class TestKeepSession: + async def test_keep_continues_the_last_session_however_long_it_was_idle(self, two): + old = await idle_session(two, f"{CHAT}:room-1") + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", session="keep") + + run = await turn_of(two, stream, chat_event(), StreamingTurns()) + + assert run["session_id"] == old + assert (await two.db.get_channel_session(f"{CHAT}:room-1"))["session_id"] == old + + @pytest.mark.parametrize("behaviour", [{"session": "rotate"}, {}], ids=["rotate", "absent"]) + async def test_rotate_starts_a_new_session_after_the_sticky_period(self, two, behaviour): + old = await idle_session(two, f"{CHAT}:room-1") + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", **behaviour) + + run = await turn_of(two, stream, chat_event(), StreamingTurns()) + + assert run["session_id"] != old + assert (await two.db.get_channel_session(f"{CHAT}:room-1"))["session_id"] == run["session_id"] + + async def test_keep_starts_a_new_session_when_the_last_one_is_archived(self, two): + old = await idle_session(two, f"{CHAT}:room-1", status="archived") + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", session="keep") + + run = await turn_of(two, stream, chat_event(), StreamingTurns()) + + assert run["session_id"] != old + + async def test_keep_starts_the_first_session_and_keeps_it_for_the_next_message(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", session="keep") + turns = StreamingTurns() + + first = await turn_of(two, stream, chat_event(message_id="m-1"), turns) + await two.db.update_session_fields(first["session_id"], {"status": "idle", "last_activity_at": LONG_AGO}) + second = await turn_of(two, stream, chat_event(message_id="m-2"), turns) + + assert second["session_id"] == first["session_id"] + assert [run["channel"] for run in turns.runs] == [CHAT, CHAT] + + async def test_keep_holds_each_conversation_to_its_own_session(self, two): + first = await idle_session(two, f"{CHAT}:room-1") + second = await idle_session(two, f"{CHAT}:room-2") + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", session="keep") + turns = StreamingTurns() + + one = await turn_of(two, stream, chat_event(conversation="room-2", message_id="m-3"), turns) + other = await turn_of(two, stream, chat_event(conversation="room-1", message_id="m-4"), turns) + + assert (one["session_id"], other["session_id"]) == (second, first) From d62c9fdd5440eedfa90f97de0af3c056926abd27 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:32:17 +0000 Subject: [PATCH 05/10] Document hosted chat providers and connection behaviour The hosted mode docs now say that a channel uses only the connections of its own provider, what the answer and session modes of a connection do, the line that Nerve puts before a turn in tool mode, and what a switched-off provider does with its events. Hosted Slack still holds the events of every provider while it is off. Co-Authored-By: Claude Opus 5.5 --- docs/architecture.md | 2 +- docs/config.md | 40 +++++++++++++++++++++++++++++++++++++--- 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index fbaf1f64b..fc94dde27 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -110,7 +110,7 @@ Abstract communication layer with three components: Implementations: - **Telegram** — python-telegram-bot v21+ with partial message streaming (edit-in-place, 1.5s rate limit), inline keyboard buttons for notification questions, `/reply` command for free-text answers; a reply to a question card answers it too. With `telegram.reply_routes_to_origin_session`, a reply routes to the session that produced the replied-to message; the message→session map is the `channel_reply_routes` table (recorded whether or not the flag is on, pruned after 90 days by the maintenance tick), so it survives restarts - **Slack** — slack_sdk Socket Mode (outbound WebSocket, no public URL) with partial streaming via `chat.update`, Block Kit buttons, `/nerve` slash command, and per-thread sessions. `SlackAccessPolicy` (`slack_access.py`) composes user, channel, and DM guardrails; in channels the bot answers only on mention or in a thread it already owns. -- **Hosted** (`hosted/`): provider traffic that a shared channel gateway holds (`slack.mode: hosted`). Nerve has no provider tokens. Each gateway replica opens a WebSocket to `/_internal/channel/v1/stream`, authenticated with a token that the gateway signs (ES256, issuer `nerve-gateway`, audience `nerve-channel`, subject naming this exact tenant and agent, at most 300 seconds of lifetime). One inbox reader pulls admitted events over the preferred stream: read, process, acknowledge, and read again until a page is empty. `HostedChannel` registers under the provider's name, so the router and tools see `slack` as with Socket Mode, and uses the same session keys. The gateway decides which events invoke the agent: a direct message, a mention of the agent, a thread reply, a reaction, an interaction, or a command. A mention of the agent carries the `self` flag, so intake needs no capabilities. Each invoke event names the person that the gateway resolved from the provider user; in external mode, the router resolves it to a human actor (`ChannelRouter.actor_for_principal`), and the turn runs as that person. An observe event can name the person that the gateway linked to its author; the inbox record shows that person for attribution only, and no actor comes from it. Outbound messages are contract operations: `OperationRunner` (`hosted/operations.py`) sends each on a stream that advertised it for the connection, within the stream and connection in-flight limits, and never sends a side-effecting operation again after its result is lost. File uploads and attachment reads use transfers on the stream of their operation; a read is sent only while its length fits Nerve's stream memory limit. Notification cards carry buttons, and a press arrives as an `interaction` event that answers the notification. A card goes with `notify`, and the gateway posts it in the notification channel that the control plane set for the agent; the config does not choose this conversation. `hosted/contract/` holds the version 1 channel contract frames that Nerve reads and sends. The gateway checks the contract; Nerve checks only the members and values that its code uses, and its size limits. +- **Hosted** (`hosted/`): provider traffic that a shared channel gateway holds (`slack.mode: hosted`). Nerve has no provider tokens. Each gateway replica opens a WebSocket to `/_internal/channel/v1/stream`, authenticated with a token that the gateway signs (ES256, issuer `nerve-gateway`, audience `nerve-channel`, subject naming this exact tenant and agent, at most 300 seconds of lifetime). One inbox reader pulls admitted events over the preferred stream: read, process, acknowledge, and read again until a page is empty. The runtime builds one `HostedChannel` for each provider in the table of hosted chat providers (`hosted/providers.py`), which gives each provider its hosted mode, its switch, and its channel settings, with defaults for a provider whose config has none; Slack is the only entry. `HostedChannel` registers under the provider's name, so the router and tools see `slack` as with Socket Mode, and uses the same session keys. When the gateway names the provider in a connection's capabilities, a channel uses only the connections of its own provider. The connection's behaviour can tell the router to post nothing for a turn (`InboundMessage.post_answer`), so that the agent answers only with a tool, and can keep a conversation in its last session past the sticky period. The gateway decides which events invoke the agent: a direct message, a mention of the agent, a thread reply, a reaction, an interaction, or a command. A mention of the agent carries the `self` flag, so intake needs no capabilities. Each invoke event names the person that the gateway resolved from the provider user; in external mode, the router resolves it to a human actor (`ChannelRouter.actor_for_principal`), and the turn runs as that person. An observe event can name the person that the gateway linked to its author; the inbox record shows that person for attribution only, and no actor comes from it. Outbound messages are contract operations: `OperationRunner` (`hosted/operations.py`) sends each on a stream that advertised it for the connection, within the stream and connection in-flight limits, and never sends a side-effecting operation again after its result is lost. File uploads and attachment reads use transfers on the stream of their operation; a read is sent only while its length fits Nerve's stream memory limit. Notification cards carry buttons, and a press arrives as an `interaction` event that answers the notification. A card goes with `notify`, and the gateway posts it in the notification channel that the control plane set for the agent; the config does not choose this conversation. `hosted/contract/` holds the version 1 channel contract frames that Nerve reads and sends. The gateway checks the contract; Nerve checks only the members and values that its code uses, and its size limits. - **Web** — Passive channel using gateway WebSocket Adding a new channel (Discord, WhatsApp, etc.) requires implementing ~5 methods and zero session/routing logic. diff --git a/docs/config.md b/docs/config.md index 4b94b41d8..6670ef2ef 100644 --- a/docs/config.md +++ b/docs/config.md @@ -256,7 +256,7 @@ A reload is always explicit. Two things cause one: | `external_agents.targets` (including each target's `enabled`), `.sync_interval_minutes`, `.conflict_policy` | ✅ from the next sweep, provided at least one target existed at startup (see the restart table) | | `sessions.sticky_period_minutes` | ✅ | | `telegram.dm_policy`, `.stream_mode`, `.reply_routes_to_origin_session` | ✅ read per update. Tightening `open` to `pairing` takes effect on the next message; toggling `reply_routes_to_origin_session` takes effect on the next message, and applies to replies to earlier messages too (routes are recorded while it is off); `allowed_users` does not follow it (see the restart table) | -| `slack.*` | ✅ `enabled` starts or stops the channel; same-workspace token changes reconnect and roll back on failure; other changes apply to the next event. In hosted mode `enabled: false` pauses intake, the records of a hosted GitHub source included, and notifications until it is `true` again, `source.*` applies to the next event, and `mode` needs a restart (see the restart table) | +| `slack.*` | ✅ `enabled` starts or stops the channel; same-workspace token changes reconnect and roll back on failure; other changes apply to the next event. In hosted mode `enabled: false` pauses intake, the records of a hosted GitHub source and the events of the other hosted providers included, and notifications until it is `true` again, `source.*` applies to the next event, and `mode` needs a restart (see the restart table) | | `workflows.*` and `workflows.review_loop.*` — budget caps, concurrency, the warning fraction, iteration and criteria caps, leg engines/models, the verifier sandbox | ✅ read per use, by loops and runs already in flight as well as new ones. The two `enabled` flags and the two loop cadences are the exceptions; see the restart table | | `provider.*` and the API keys it selects (`aws_region`, `aws_profile`, `aws_access_key_id`, and the effective Anthropic key) | ✅ for sessions started **after** the reload. Each client's environment is built from the live reference when the session is created, by the same seam as `agent.*` below | | **`agent.*` and `codex.*`**: backend choice and models (`agent.backend`, `agent.cron_model`, `agent.model`, `codex.model`, `codex.cron_model`, `codex.models`), `max_turns`, `agent.effort`/`cron_effort` and `codex.effort_map`, `agent.thinking`, `agent.context_1m*`, `agent.background_agent_permissions`, `agent.agent_teams`, `agent.subagents.*`, idle timeouts, cache TTL, `codex.sandbox`, `.approval_policy`, `.web_search`, `.extra_config`, `.tool_timeout_sec`, `.bin_path`, `.auth`/`.api_key`/`.api_key_env`, `.base_url`, `.pricing`, `.min_version`/`.max_version`, `.ultracode.*` | ✅ for sessions and turns **started after** the reload. The engine and both backends resolve these through one live reference, so a key cannot be hot in one and frozen in the other | @@ -1372,7 +1372,10 @@ hosted mode the switch is ignored, with a warning. Nerve accepts events from every Slack connection that the gateway serves for this agent. For each conversation, Nerve records the connection of the -latest accepted event. +latest accepted event. When the gateway names the provider of each +connection, the Slack channel uses only the Slack connections for its +replies, notifications, and limits. A connection whose provider the gateway +does not name serves every hosted chat channel. Each event that starts a turn or answers a notification names the person that the gateway resolved from the Slack user. In external mode, the turn runs as @@ -1394,7 +1397,8 @@ gateway's inbox until a reload sets `enabled` to `true`. While it is paused, Nerve sends no notifications and refuses `send_channel_message`. A turn that is already running still sends its reply. Nerve reads nothing from the gateway's inbox while Slack is paused, so the records of a hosted GitHub -source wait there too (see below). +source and the events of the other hosted providers wait there too (see +below and [Connection behaviour](#connection-behaviour)). Replies, reactions, and `send_channel_message` go to Slack as operations on a gateway stream that advertises them for the connection. A reply goes on the @@ -1463,6 +1467,36 @@ Slack switched on while GitHub is hosted; `nerve doctor` warns when it is off. This is a known limitation: a read for each provider on its own, which would keep GitHub records arriving while hosted Slack is off, is a follow-up. +#### Connection behaviour + +Nerve builds one hosted channel for each chat provider in hosted mode; Slack +is the only hosted chat provider of this version. The capabilities that the +gateway sends for a connection can name the connection's provider and a +behaviour, which tells Nerve how to answer the turns of its events and which +session they go to: + +| Member | Values | Default | Effect | +|--------|--------|---------|--------| +| `answer` | `post`, `tool` | `post` | `post`: Nerve posts the turn's answer in the conversation, with the typing indicator, the streamed placeholder, and an `Error:` message when the turn fails. `tool`: Nerve posts nothing in the conversation for the turn. The answer stays in the session, and the agent answers only with a tool | +| `reply_tool` | tool name | none | Only with `tool`: the tool that the agent can answer with | +| `session` | `rotate`, `keep` | `rotate` | `rotate`: a conversation gets a new session after `sessions.sticky_period_minutes` of idle time. `keep`: a conversation continues its last session, however long it was idle, until that session is archived or deleted | + +Slack connections post and rotate, so a gateway that sends no behaviour +changes nothing. A connection whose capabilities Nerve does not have posts +and rotates too. In `tool` mode, Nerve puts one line before the text of the +turn: "This message came through *provider*. Your answer in this session is +not sent to anyone. To answer, use the *reply tool* tool." Without a reply +tool, the line ends with "This channel does not send answers." + +Each hosted chat provider has its own switch. Because the gateway keeps the +events of every provider in one ordered inbox, a provider that is switched +off either holds the events of all providers or refuses its own events. +Hosted Slack holds them: while `slack.enabled` is `false`, Nerve reads +nothing, and the events of the other hosted providers wait in the gateway +inbox too. Any other hosted chat provider refuses: while it is switched off, +Nerve rejects its events and continues to read the events of the other +providers. + ### Setting up the Slack app Socket Mode means the bot dials out to Slack, so Nerve needs no public URL. From 9e45777ae0459a0ee0cede47abf8b98bd7672343 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:32:51 +0000 Subject: [PATCH 06/10] Hosted channels: name no config key in a switched-off rejection A provider's switch need not be a key named after the provider, so the log detail of a rejected event says only that the provider is off. The runtime docstring also says that a provider's switch and settings follow a reload. Co-Authored-By: Claude Opus 5.5 --- nerve/channels/hosted/channel.py | 4 +--- nerve/channels/hosted/runtime.py | 7 ++++--- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/nerve/channels/hosted/channel.py b/nerve/channels/hosted/channel.py index 025c62ac7..f4438c8fe 100644 --- a/nerve/channels/hosted/channel.py +++ b/nerve/channels/hosted/channel.py @@ -894,9 +894,7 @@ async def deliver(self, event: Event) -> Disposition: each event is rejected. """ if not self._spec.hold_while_off and not self._switched_on(): - return Disposition.rejected( - "configuration_invalid", f"{self._provider} is switched off ({self._provider}.enabled is false)", - ) + return Disposition.rejected("configuration_invalid", f"{self._provider} is switched off") purpose = event.admission.purpose if purpose == "invoke" and event.kind == "message": disposition = await self._invoke_message(event) diff --git a/nerve/channels/hosted/runtime.py b/nerve/channels/hosted/runtime.py index 4871c3706..dd819c744 100644 --- a/nerve/channels/hosted/runtime.py +++ b/nerve/channels/hosted/runtime.py @@ -4,9 +4,10 @@ runner, the inbox reader, one :class:`HostedChannel` for each chat provider in hosted mode (see :mod:`nerve.channels.hosted.providers`), and one :class:`HostedSourceConsumer` for each hosted source. The hosted settings and -the provider modes are read once at startup; a change needs a restart. A reload reads the gateway key file again and publishes the -source subscriptions again. The source grant of each chat provider is read -per event, so it follows a reload. +the provider modes are read once at startup; a change needs a restart. A +reload reads the gateway key file again and publishes the source +subscriptions again. The switch, the settings, and the source grant of each +chat provider are read per event, so they follow a reload. """ from __future__ import annotations From d416a4ac5699004a13ed4fd87a0334cafac6cdc7 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:54:49 +0000 Subject: [PATCH 07/10] Router: resolve one conversation's sessions one at a time Two messages of a new conversation that arrived together could each find no session and create one, so one conversation was split across two sessions. With the keep session mode this breaks the rule of one session per conversation; in rotate mode it was a race as well. The router now resolves the session and queues the message under a lock per channel key. The second message then finds the session that the first one created, and the queue keeps the arrival order. Co-Authored-By: Claude Opus 5.5 --- nerve/channels/router.py | 44 +++++++++++++++----------- tests/test_hosted_channel_behaviour.py | 44 ++++++++++++++++++++++++-- 2 files changed, 67 insertions(+), 21 deletions(-) diff --git a/nerve/channels/router.py b/nerve/channels/router.py index 68c76817d..54c3c50d4 100644 --- a/nerve/channels/router.py +++ b/nerve/channels/router.py @@ -104,6 +104,9 @@ def __init__(self, engine: AgentEngine): self._pending_batches: dict[ str, list[tuple[InboundMessage, asyncio.Future[str]]] ] = {} + # Per-channel-key locks: the messages of one conversation resolve + # their session and join its queue one at a time. + self._resolution_locks: dict[str, asyncio.Lock] = {} # ------------------------------------------------------------------ # # Channel registry # @@ -203,32 +206,37 @@ async def handle_message(self, msg: InboundMessage) -> str: (``post_answer``) sends nothing to the channel: no typing indicator, no streamed or final answer, and no error message. + The messages of one channel key resolve their session one at a + time, in arrival order. Simultaneous first messages of a + conversation thus share the session that the first one creates. + Called by channel implementations when they receive a user message. """ channel = self._channels.get(msg.channel_name) if not channel: raise ValueError(f"Unknown channel: {msg.channel_name}") - # Resolve session - if msg.session_id: - session_id = msg.session_id - await self.engine.sessions.set_active_session( - msg.channel_key, session_id, - ) - else: - session_id = await self.engine.sessions.get_active_session( - msg.channel_key, source=msg.channel_name, actor=msg.actor, - conversation_title=msg.conversation_title, - ) + future: asyncio.Future[str] = asyncio.get_running_loop().create_future() + async with self._resolution_locks.setdefault(msg.channel_key, asyncio.Lock()): + # Resolve session + if msg.session_id: + session_id = msg.session_id + await self.engine.sessions.set_active_session( + msg.channel_key, session_id, + ) + else: + session_id = await self.engine.sessions.get_active_session( + msg.channel_key, source=msg.channel_name, actor=msg.actor, + conversation_title=msg.conversation_title, + ) - # Store message context for reaction support - self._note_message_context(session_id, msg) + # Store message context for reaction support + self._note_message_context(session_id, msg) - # Queue the message; a Future carries the result back to the caller. - future: asyncio.Future[str] = asyncio.get_running_loop().create_future() - self._pending_batches.setdefault(session_id, []).append( - (msg, future), - ) + # Queue the message; a Future carries the result back to the caller. + self._pending_batches.setdefault(session_id, []).append( + (msg, future), + ) # If the session is already busy, the message is queued — # the driver coroutine will include it in the next batch. diff --git a/tests/test_hosted_channel_behaviour.py b/tests/test_hosted_channel_behaviour.py index e99c715fc..b2904c80f 100644 --- a/tests/test_hosted_channel_behaviour.py +++ b/tests/test_hosted_channel_behaviour.py @@ -10,9 +10,10 @@ refuses its events while Slack goes on, channels that use only the connections of their own provider, the ``tool`` answer mode, in which the router sends nothing to the conversation and the turn's text starts with a -note for the agent, and the ``keep`` session mode, in which a conversation -keeps its last session past the sticky period. The router's part is also tested on its own, with a -channel that records every call. +note for the agent, the ``keep`` session mode, in which a conversation +keeps its last session past the sticky period, and simultaneous first +messages of one conversation. The router's part is also tested on its own, +with a channel that records every call. """ from __future__ import annotations @@ -575,3 +576,40 @@ async def test_keep_holds_each_conversation_to_its_own_session(self, two): other = await turn_of(two, stream, chat_event(conversation="room-1", message_id="m-4"), turns) assert (one["session_id"], other["session_id"]) == (second, first) + + +# ---------------------------------------------------------------------- # +# Simultaneous first messages # +# ---------------------------------------------------------------------- # + + +@pytest.mark.asyncio +class TestSimultaneousFirstMessages: + async def test_two_first_messages_of_one_conversation_share_one_session(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", session="keep") + await two.gateway.wait_for(lambda: two.runtime.streams.connections("send", CHAT)) + turns = StreamingTurns() + two.router.engine.run = turns.run + first = two.gateway.store(chat_event(message_id="m-1", text="first")) + second = two.gateway.store(chat_event(message_id="m-2", text="second")) + + await stream.nudge("invoke") + + assert await two.acknowledged(first, second) == {first: "accepted", second: "accepted"} + await turns.wait_for(1) + await asyncio.sleep(0.3) + mapping = await two.db.get_channel_session(f"{CHAT}:room-1") + assert {run["session_id"] for run in turns.runs} == {mapping["session_id"]} + said = "\n\n".join(run["user_message"] for run in turns.runs) + assert said.index("first") < said.index("second") + + async def test_the_router_resolves_one_session_for_simultaneous_messages(self, db): + turns = StreamingTurns() + router, _ = recording_router(db, turns) + + await asyncio.gather(*(router.handle_message(inbound(f"message {n}")) for n in range(3))) + + mapping = await db.get_channel_session("recording:room-1") + assert {run["session_id"] for run in turns.runs} == {mapping["session_id"]} + From 69d1d8eb6710107fecb964f8c85f25ffc7e4f062 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:54:57 +0000 Subject: [PATCH 08/10] Hosted channels: refuse connections that serve another provider A channel remembered the connection of every accepted event and used it for replies, and it read the behaviour of that connection, without a check of the connection's provider. An event of one provider on a connection of another, from a gateway fault or a changed announcement, could thus take the other provider's behaviour and send through its connection. A channel now rejects an event on a connection whose capabilities name another provider, and it sends no operation through such a connection: a remembered connection that now serves another provider is skipped, and the channel chooses among its own connections. A connection that names no provider still serves every channel. Co-Authored-By: Claude Opus 5.5 --- docs/config.md | 4 +++ nerve/channels/hosted/channel.py | 32 ++++++++++++++++++---- tests/test_hosted_channel_behaviour.py | 37 ++++++++++++++++++++++++++ 3 files changed, 68 insertions(+), 5 deletions(-) diff --git a/docs/config.md b/docs/config.md index 6670ef2ef..458f499e3 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1497,6 +1497,10 @@ inbox too. Any other hosted chat provider refuses: while it is switched off, Nerve rejects its events and continues to read the events of the other providers. +An event on a connection whose capabilities name another provider is +rejected, and Nerve sends no operation of a channel through such a +connection. + ### Setting up the Slack app Socket Mode means the bot dials out to Slack, so Nerve needs no public URL. diff --git a/nerve/channels/hosted/channel.py b/nerve/channels/hosted/channel.py index f4438c8fe..e47815f40 100644 --- a/nerve/channels/hosted/channel.py +++ b/nerve/channels/hosted/channel.py @@ -41,7 +41,8 @@ The channel serves the connections of its provider. A connection whose capabilities name no provider, as from an older gateway, serves every hosted -chat channel. The provider's entry in +chat channel. An event on a connection whose capabilities name another +provider is rejected, and no operation goes to such a connection. The provider's entry in :data:`~nerve.channels.hosted.providers.HOSTED_CHAT_PROVIDERS` gives the channel's switch and settings. @@ -474,7 +475,7 @@ async def _connection(self, conversation_id: str, operation: str) -> uuid.UUID: raise OperationFailed(operation, "unavailable", "invalid_target", local=True, detail="the target names no conversation") connection = self._connections.get(conversation_id) - if connection is not None: + if connection is not None and not self._foreign(connection): return connection candidates = ( await self._operations.connections(operation, self._provider) if self._operations else set() @@ -487,6 +488,17 @@ async def _connection(self, conversation_id: str, operation: str) -> uuid.UUID: detail=f"no single {self._provider} connection serves conversation {conversation_id}", ) + def _foreign(self, connection_id: uuid.UUID) -> bool: + """Whether the capabilities of *connection_id* name another provider. + + A connection without capabilities now, or whose capabilities name no + provider, is not foreign. + """ + capabilities = ( + self._operations.streams.capabilities_for(connection_id) if self._operations else None + ) + return capabilities is not None and not capabilities.serves(self._provider) + def _limits(self, connection_id: uuid.UUID) -> ConnectionLimits: capabilities = ( self._operations.streams.capabilities_for(connection_id) if self._operations else None @@ -512,6 +524,9 @@ async def _perform(self, connection_id: uuid.UUID, kind: str, payload: Any, **op if self._operations is None: raise OperationFailed(kind, "unavailable", "provider_unavailable", local=True, detail="no gateway streams") + if self._foreign(connection_id): + raise OperationFailed(kind, "unavailable", "invalid_target", local=True, + detail=f"connection {connection_id} serves another provider") return await self._operations.perform(connection_id, kind, payload, **options) async def _fitting( @@ -837,7 +852,8 @@ async def authorize_outbound(self, target: str) -> Decision: return Decision( False, f"the destination is not approved by the {self._provider} channel policy", ) - if self._connections.get(conversation) is None and self._operations is not None: + known = self._connections.get(conversation) + if (known is None or self._foreign(known)) and self._operations is not None: candidates = self._operations.streams.connections("send", self._provider) if len(candidates) > 1: return Decision( @@ -891,10 +907,16 @@ async def deliver(self, event: Event) -> Disposition: """Dispatch one admitted event and say what happened to it. While the provider is switched off, and it does not hold its events, - each event is rejected. + each event is rejected. So is an event on a connection whose + capabilities name another provider: it is not remembered, and its + connection's behaviour does not apply. """ if not self._spec.hold_while_off and not self._switched_on(): return Disposition.rejected("configuration_invalid", f"{self._provider} is switched off") + if self._foreign(event.connection_id): + return Disposition.rejected( + "configuration_invalid", f"connection {event.connection_id} serves another provider", + ) purpose = event.admission.purpose if purpose == "invoke" and event.kind == "message": disposition = await self._invoke_message(event) @@ -981,7 +1003,7 @@ async def _read_attachment( self, event: Event, attachment: AttachmentReference, length: int, ) -> bytes | None: """Read up to *length* bytes of an attachment, or ``None`` on failure.""" - if self._operations is None: + if self._operations is None or self._foreign(event.connection_id): return None try: return await self._operations.read(event.connection_id, "file_read", FileReadOperation( diff --git a/tests/test_hosted_channel_behaviour.py b/tests/test_hosted_channel_behaviour.py index b2904c80f..a2981f2d1 100644 --- a/tests/test_hosted_channel_behaviour.py +++ b/tests/test_hosted_channel_behaviour.py @@ -38,6 +38,7 @@ OutboundMessage, ) from nerve.channels.hosted.channel import answer_note +from nerve.channels.hosted.operations import OperationFailed from nerve.channels.hosted import providers as providers_module from nerve.channels.hosted.providers import ( HOSTED_CHAT_PROVIDERS, @@ -296,6 +297,42 @@ async def test_a_connection_of_another_provider_does_not_make_a_channel_availabl assert not two.runtime.channels["slack"].is_available assert (await two.runtime.channels["slack"].post_notification("n-2", "Deploy?", [("Yes", "yes")])) is None + async def test_an_event_on_a_connection_of_another_provider_is_rejected(self, two): + stream = await two.stream(advertise=False) + await announce(stream, "slack", CONNECTION_ID, **SLACK_BEHAVIOUR) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", session="keep") + await two.gateway.wait_for(lambda: len(two.runtime.streams.connections("send")) == 2) + on_slack = two.gateway.store(chat_event(connection_id=CONNECTION_ID)) + on_chat = two.gateway.store(slack_event(connection_id=CHAT_CONNECTION)) + + await stream.nudge("invoke") + + assert await two.acknowledged(on_slack, on_chat) == {on_slack: "rejected", on_chat: "rejected"} + assert [item["reason_code"] for item in two.gateway.acks[-1]] == ["configuration_invalid"] * 2 + await asyncio.sleep(0.2) + assert two.turns.runs == [] + assert two.runtime.channels[CHAT].connection_for("room-1") is None + assert two.runtime.channels["slack"].connection_for("D0DIRECT") is None + + async def test_a_known_connection_that_now_serves_another_provider_gets_no_operation(self, two): + stream = await two.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool") + inbox_id = two.gateway.store(chat_event()) + await stream.nudge("invoke") + assert (await two.acknowledged(inbox_id))[inbox_id] == "accepted" + await two.turns.wait_for(1) + chat = two.runtime.channels[CHAT] + assert str(chat.connection_for("room-1")) == CHAT_CONNECTION + + await announce(stream, "slack", CHAT_CONNECTION) + await two.gateway.wait_for(lambda: not two.runtime.streams.connections("send", CHAT)) + + with pytest.raises(OperationFailed): + await chat.send(OutboundMessage(target="room-1", text="through the wrong connection")) + await chat.set_reaction("room-1", "m-1", "\U0001f44d") + await chat.edit_message("room-1", "m-1", "edited") + assert two.gateway.operations == [] + # ---------------------------------------------------------------------- # # The answer mode # From 123f4ae02d23af3a6b3700883089df53598f7ea7 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:55:03 +0000 Subject: [PATCH 09/10] Document that a full hosted channel pauses every provider The inbox of the gateway is one ordered queue, so Nerve reads no event of any provider while one hosted channel has its maximum of turns in progress. The hosted mode docs now say so where they describe the switched-off hold. Co-Authored-By: Claude Opus 5.5 --- docs/config.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/config.md b/docs/config.md index 458f499e3..48ba29800 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1495,7 +1495,8 @@ Hosted Slack holds them: while `slack.enabled` is `false`, Nerve reads nothing, and the events of the other hosted providers wait in the gateway inbox too. Any other hosted chat provider refuses: while it is switched off, Nerve rejects its events and continues to read the events of the other -providers. +providers. Nerve also stops reading, for every provider, while one hosted +channel has 100 turns in progress, and continues when one of them ends. An event on a connection whose capabilities name another provider is rejected, and Nerve sends no operation of a channel through such a From 22fb6721e689480e55bf1bf42a332a1d9b371f5a Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 3 Nov 2026 07:55:03 +0000 Subject: [PATCH 10/10] Tests: pin the contract sample's bytes, store a tool turn's answer The copy of the contract's capabilities sample must stay byte for byte the same as the gateway's, so a test checks its SHA-256. A test with a real engine, whose model process is a stand-in, shows that the answer of a tool turn is stored in the session while nothing is posted. Co-Authored-By: Claude Opus 5.5 --- tests/test_channel_contract.py | 6 ++ tests/test_hosted_channel_behaviour.py | 113 +++++++++++++++++++++++-- 2 files changed, 114 insertions(+), 5 deletions(-) diff --git a/tests/test_channel_contract.py b/tests/test_channel_contract.py index 13478734c..ec6e27f25 100644 --- a/tests/test_channel_contract.py +++ b/tests/test_channel_contract.py @@ -9,6 +9,7 @@ import copy import dataclasses +import hashlib import json import uuid from datetime import datetime, timedelta, timezone @@ -407,6 +408,8 @@ def test_capabilities_need_usable_limits(self): # provider and behaviour. It must stay byte for byte the same as the # contract's ``testdata/frames/gateway/capabilities_behaviour.json``. BEHAVIOUR_SAMPLE = Path(__file__).parent / "fixtures" / "channel_contract" / "capabilities_behaviour.json" +# The SHA-256 of the contract's sample, which the copy must match. +BEHAVIOUR_SAMPLE_SHA256 = "62b4e7b2549cbe19e41bf44699f621db0db8685b35c612f6387b0b28a61f83ed" def behaviour_frame(**behaviour: Any) -> dict[str, Any]: @@ -417,6 +420,9 @@ def behaviour_frame(**behaviour: Any) -> dict[str, Any]: class TestConnectionBehaviour: + def test_the_copy_of_the_contract_sample_has_its_exact_bytes(self): + assert hashlib.sha256(BEHAVIOUR_SAMPLE.read_bytes()).hexdigest() == BEHAVIOUR_SAMPLE_SHA256 + def test_the_contract_sample_decodes_into_its_provider_and_behaviour(self): envelope = decode(BEHAVIOUR_SAMPLE.read_text(encoding="utf-8")) body = envelope.body diff --git a/tests/test_hosted_channel_behaviour.py b/tests/test_hosted_channel_behaviour.py index a2981f2d1..11fff8635 100644 --- a/tests/test_hosted_channel_behaviour.py +++ b/tests/test_hosted_channel_behaviour.py @@ -13,7 +13,9 @@ note for the agent, the ``keep`` session mode, in which a conversation keeps its last session past the sticky period, and simultaneous first messages of one conversation. The router's part is also tested on its own, -with a channel that records every call. +with a channel that records every call. One test runs a real engine whose +model process is a stand-in, to show that the answer of a ``tool`` turn is +stored in the session. """ from __future__ import annotations @@ -28,6 +30,9 @@ import pytest import pytest_asyncio +import nerve.config as config_module +from nerve.agent.backends import claude as claude_backend +from nerve.agent.engine import AgentEngine from nerve.agent.sessions import SessionManager from nerve.agent.streaming import broadcaster from nerve.channels.base import ( @@ -46,12 +51,29 @@ HostedChannelSettings, HostedChatProvider, ) -from nerve.channels.hosted.runtime import hosted_channel_providers +from nerve.channels.hosted.runtime import HostedChannelRuntime, hosted_channel_providers from nerve.channels.router import ChannelRouter from nerve.config import NerveConfig, SlackConfig - -from tests.fake_channel_gateway import CONNECTION_ID, message_event -from tests.test_hosted_channel import Hosted, start_hosted, stop_hosted +from nerve.gateway import server as gateway_server +from nerve.inference_principal import CUSTOM_HEADERS_ENV + +from tests.fake_channel_gateway import ( + CONNECTION_ID, + EVENT_TIMEOUT, + FakeChannelGateway, + NerveServer, + message_event, +) +from tests.test_hosted_channel import ( + FAST_OPERATIONS, + FAST_READER, + FAST_STREAM, + Hosted, + hosted_config, + start_hosted, + stop_hosted, +) +from tests.test_inference_principal import _FakeCli CHAT = "fixture_chat" CHAT_CONNECTION = "0d7c4b1e-6a3f-4e92-8b5d-2f1a9c7e3b60" @@ -650,3 +672,84 @@ async def test_the_router_resolves_one_session_for_simultaneous_messages(self, d mapping = await db.get_channel_session("recording:room-1") assert {run["session_id"] for run in turns.runs} == {mapping["session_id"]} + +# ---------------------------------------------------------------------- # +# A real engine keeps the answer of a tool turn # +# ---------------------------------------------------------------------- # + + +@pytest.fixture +def cli(monkeypatch) -> list[_FakeCli]: + """The model processes that the engine starts: stand-ins that answer "ok".""" + started: list[_FakeCli] = [] + monkeypatch.setattr(_FakeCli, "started", started) + monkeypatch.setattr(claude_backend, "ClaudeClient", _FakeCli) + monkeypatch.setattr( + claude_backend.ClaudeBackend, "validate_resume_target", + lambda self, native_id, cwd: native_id.startswith("native-"), + ) + return started + + +@pytest_asyncio.fixture +async def engine_two(tmp_path, db, monkeypatch, cli): + """Slack and the test provider in front of a real engine.""" + monkeypatch.setattr(providers_module, "HOSTED_CHAT_PROVIDERS", (SLACK, chat_provider(Switch()))) + monkeypatch.delenv(CUSTOM_HEADERS_ENV, raising=False) + config = hosted_config(tmp_path) + config.agent.backend = "claude" + config.codex.home_dir = str(tmp_path / "codex-home") + monkeypatch.setattr(config_module, "_config", config) + engine = AgentEngine(config, db) + engine.router.BATCH_DEBOUNCE = 0.0 + gateway = FakeChannelGateway(jwks_file=config.channels.hosted.gateway_jwks_file) + runtime = HostedChannelRuntime( + config, engine.router, lambda: config, stream_timing=FAST_STREAM, + reader_settings=FAST_READER, operation_timing=FAST_OPERATIONS, + ) + await runtime.start() + monkeypatch.setattr(gateway_server, "_hosted_channels", runtime) + server = NerveServer(gateway_server.create_app()) + await server.__aenter__() + try: + yield Hosted(config, db, engine.router, None, gateway, runtime, server) + finally: + await runtime.stop() + await gateway.close_all() + await server.__aexit__(None, None, None) + for session_id in list(engine._idle_watchers): + engine._stop_idle_watcher(session_id) + + +async def transcript(db, channel_key: str, count: int) -> list[tuple[str, str]]: + """The session's messages, once it has *count* of them.""" + deadline = asyncio.get_running_loop().time() + EVENT_TIMEOUT + while True: + mapping = await db.get_channel_session(channel_key) + if mapping is not None: + rows = await db.get_messages(mapping["session_id"]) + if len(rows) >= count: + return [(row["role"], row["content"]) for row in rows] + if asyncio.get_running_loop().time() > deadline: + raise AssertionError(f"{channel_key} did not get {count} messages") + await asyncio.sleep(0.02) + + +@pytest.mark.asyncio +class TestToolAnswerInTheSession: + async def test_the_answer_is_stored_in_the_session_and_not_posted(self, engine_two, cli): + hosted = engine_two + stream = await hosted.stream(advertise=False) + await announce(stream, CHAT, CHAT_CONNECTION, answer="tool", reply_tool="send_reply", session="keep") + inbox_id = hosted.gateway.store(chat_event(text="Please check the report.")) + + await stream.nudge("invoke") + + assert (await hosted.acknowledged(inbox_id))[inbox_id] == "accepted" + said = await transcript(hosted.db, f"{CHAT}:room-1", 2) + assert said == [ + ("user", f"{answer_note(CHAT, 'send_reply')}\n\nPlease check the report."), + ("assistant", "ok"), + ] + await asyncio.sleep(0.3) + assert hosted.gateway.operations == []