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..48ba29800 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,41 @@ 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. 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 +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/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 80c5f73ab..e47815f40 100644 --- a/nerve/channels/hosted/channel.py +++ b/nerve/channels/hosted/channel.py @@ -13,6 +13,17 @@ 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`). 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 interaction answers its notification as that person. In local mode, the @@ -28,14 +39,21 @@ 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. 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. + 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 @@ -72,7 +90,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, @@ -109,6 +127,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: @@ -184,6 +207,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) @@ -265,7 +298,7 @@ class HostedChannel(BaseChannel): def __init__( self, - provider: str, + provider: HostedChatProvider, router: ChannelRouter, config: Callable[[], NerveConfig], *, @@ -276,7 +309,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 +368,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 +384,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 +400,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,17 +467,19 @@ 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, 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) 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( @@ -450,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 @@ -465,13 +514,19 @@ 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): 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( @@ -694,10 +749,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 +784,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 +792,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 +829,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 +841,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)) @@ -795,8 +852,9 @@ 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: - candidates = self._operations.streams.connections("send") + 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( False, @@ -829,22 +887,36 @@ 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. 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) @@ -931,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( @@ -1061,17 +1133,36 @@ 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. 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( event.admission.invoke.resolved_principal_id, @@ -1079,11 +1170,17 @@ 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 + 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 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( @@ -1109,8 +1206,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. @@ -1200,4 +1296,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/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/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..dd819c744 100644 --- a/nerve/channels/hosted/runtime.py +++ b/nerve/channels/hosted/runtime.py @@ -2,11 +2,12 @@ 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 -source subscriptions again. The source grant of each chat provider is read -per event, so it follows a reload. +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 switch, the settings, and the source grant of each +chat provider are read per event, so they follow a reload. """ from __future__ import annotations @@ -26,6 +27,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 +58,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 +100,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 +130,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 +221,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/nerve/channels/router.py b/nerve/channels/router.py index 639613cc2..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 # @@ -199,32 +202,41 @@ 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. + + 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. @@ -278,17 +290,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 +317,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 +330,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 +350,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 +373,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/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/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..ec6e27f25 100644 --- a/tests/test_channel_contract.py +++ b/tests/test_channel_contract.py @@ -9,15 +9,18 @@ import copy import dataclasses +import hashlib 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 +404,93 @@ 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" +# 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]: + body = capabilities() + body["provider"] = "slack" + body["behaviour"] = behaviour + return frame("capabilities", body, connection_id=CONNECTION_ID) + + +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 + + 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"}} diff --git a/tests/test_hosted_channel_behaviour.py b/tests/test_hosted_channel_behaviour.py new file mode 100644 index 000000000..11fff8635 --- /dev/null +++ b/tests/test_hosted_channel_behaviour.py @@ -0,0 +1,755 @@ +"""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, 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, 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. 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 + +import asyncio +import uuid +from dataclasses import dataclass +from typing import Any + +from unittest.mock import MagicMock + +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 ( + BaseChannel, + ChannelCapability, + ChannelConstraints, + InboundMessage, + 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, + SLACK, + HostedChannelSettings, + HostedChatProvider, +) +from nerve.channels.hosted.runtime import HostedChannelRuntime, hosted_channel_providers +from nerve.channels.router import ChannelRouter +from nerve.config import NerveConfig, SlackConfig +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" +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 + + 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 # +# ---------------------------------------------------------------------- # + + +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" + + +# ---------------------------------------------------------------------- # +# 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) + + +# ---------------------------------------------------------------------- # +# 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"]} + + +# ---------------------------------------------------------------------- # +# 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 == []