diff --git a/config.example.yaml b/config.example.yaml index c10f1797..b47075d7 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -100,15 +100,30 @@ gateway: # Channels telegram: enabled: true - # DM authorization: + # Live-message authorization in private and group chats: # pairing — only users in allowed_users may talk to the bot. New users # authorize with a one-time code: run `nerve pair` on the - # server, then send the bot `/pair `. Paired IDs are - # persisted to config.local.yaml automatically. + # server, then send the bot `/pair ` in a private chat. + # Paired IDs are persisted to config.local.yaml automatically. # open — anyone can talk to the bot (dangerous: full agent access). dm_policy: pairing - # allowed_users: [123456789] # numeric Telegram user IDs (or pair instead) + # allowed_users: [123456789] # users allowed to use the bot (or pair instead) stream_mode: partial # "partial" (edit messages) or "full" (wait for complete) + # + # Feed the source inbox from watched group chats, reaching it as the source + # "telegram:observed". This grant is independent of allowed_users: a sender + # who cannot use the bot may still reach the source. A live-routed message is + # not collected unless include_handled_messages is true. Only numeric IDs may + # grant; titles and @usernames can only deny. Private chats, other bots, and + # /pair commands are never collected. With privacy mode enabled, a non-admin + # bot does not receive ordinary group messages. Make it an admin or disable + # privacy mode with BotFather. + # See docs/sources.md before enabling. + # source: + # enabled: true + # allow_chats: ["-1001234567890"] + # deny_senders: ["12345"] + # include_handled_messages: false # true also collects live-routed messages slack: # Omitting this key leaves Slack off until both tokens below are set, so an @@ -152,6 +167,34 @@ slack: # [all] — everything, including doctor and restart # doctor/restart affect the host; sessions lists other channels. Opt in. # commands: [sessions, new, stop, reply] + # + # Feed the source inbox from watched channels, reaching it as the source + # "slack:observed", where poll_source and the `messages` cron gate read it. + # Nothing here starts an agent turn. + # + # A SEPARATE grant from the access rules above, decided independently of + # them: those say who may drive the agent, this says whose traffic may reach + # its inbox. So an empty allow_channels collects NOTHING rather than + # everything. A mention from someone access refuses is still collected; a + # live-routed message is not collected unless include_handled_messages is true. + # DMs (group DMs included), other apps, and the bot's own posts are never + # collected. + # + # allow_channels takes a channel ID or a channel name, with globs and a + # leading "#" tolerated; deny_channels the same. allow_senders takes a + # member ID, handle, or email — a display name can only ever deny, since + # its owner picks it. Prefer uppercase IDs: everything else uses cached Slack + # API lookups. + # + # Everything collected is untrusted input: the grant decides whose messages + # are kept, not whether their contents can be believed. See docs/sources.md. + # source: + # enabled: true + # allow_channels: ["C0456DEF", "eng-*"] + # deny_channels: ["*-social"] + # deny_senders: ["*-bot"] + # include_handled_messages: false # true also collects live-routed messages + # schedule: "*/5 * * * *" # how often the buffer drains into the inbox # Where notify, ask_user, and propose_action deliver. The list replaces the # default rather than adding to it, so name every transport you want. A diff --git a/docs/config.md b/docs/config.md index b8899d2f..28929655 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1052,7 +1052,7 @@ carry text. A `.png` or `.ico` has to be committed by a human. | `telegram.enabled` | bool | `true` | Enable Telegram bot | | `telegram.bot_token` | string | - | Bot token from @BotFather | | `telegram.dm_policy` | string | `pairing` | `pairing` (allowlist + one-time pairing codes) or `open` (anyone — dangerous) | -| `telegram.allowed_users` | list[int] | `[]` | Telegram user IDs allowed to DM the bot | +| `telegram.allowed_users` | list[int] | `[]` | User IDs allowed to use the bot in private and group chats | | `telegram.stream_mode` | string | `partial` | `partial` (edit msgs) or `full` | ### Pairing @@ -1064,13 +1064,72 @@ editing config files: 1. Run `nerve pair` on the server — it prints a one-time 6-digit code (valid 1 hour). On a fresh install with no `allowed_users`, a code is also generated automatically at startup and printed to the log. -2. Send the bot `/pair ` from the Telegram account to authorize. +2. In a private chat, send the bot `/pair ` from the Telegram account to + authorize. 3. The user ID is appended to `telegram.allowed_users` in `config.local.yaml` and takes effect immediately. -An unauthorized `/start` gets a reply with the sender's numeric ID and -pairing instructions (rate-limited); all other messages from unauthorized -users are ignored. +An unauthorized `/start` gets a reply with the sender's numeric ID and pairing +instructions (rate-limited). Other unauthorized messages cannot start a live +channel turn. Matching group messages may still go to the independent channel +source. + +### Channel source + +Group traffic can feed the source inbox, where `poll_source`, `read_source`, +and the `messages` cron gate read it. Nothing here starts an agent turn. It +is a grant of its own, independent of `allowed_users`: see +[sources.md](sources.md#chat-channels-slack-telegram) for the routing rules +and the threat model. + +```yaml +telegram: + source: + enabled: true + allow_chats: ["-1001234567890"] + deny_senders: ["12345"] + schedule: "*/5 * * * *" +``` + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `telegram.source.enabled` | bool | `false` | Feed the inbox from watched chats | +| `telegram.source.allow_chats` | list[str] | `[]` | Chats to collect. **Empty means none** | +| `telegram.source.deny_chats` | list[str] | `[]` | Never collect these | +| `telegram.source.allow_senders` | list[str] | `[]` | Restrict to these senders | +| `telegram.source.deny_senders` | list[str] | `[]` | Skip these senders | +| `telegram.source.include_handled_messages` | bool | `false` | Also collect messages accepted for live routing | +| `telegram.source.schedule` | cron/interval | `*/5 * * * *` | Drain cadence | +| `telegram.source.batch_size` | int | `50` | Records per drain | +| `telegram.source.condense` | bool | `false` | LLM-condense long messages | +| `telegram.source.max_stored_messages` | int | `10000` | Buffer trim target for all Telegram chats together; checked every 100 writes | + +Reaches the inbox as the source `telegram:observed` — distinct from the +`telegram` sync source, which is the Telethon pull from your *user* account. +The keys say *chats* rather than *channels* because on Telegram a channel is +a specific entity type distinct from a group. + +**Setup: the bot must receive group messages.** With privacy mode enabled, a +non-admin bot does not receive ordinary group messages. Make it an admin or +disable privacy mode with BotFather (`/setprivacy` → Disable). See Telegram's +[privacy-mode documentation](https://core.telegram.org/bots/faq#what-messages-will-my-bot-get). + +**What the lists match.** Only numeric IDs may grant. + +| List | Matches | Example | +|---|---|---| +| `allow_chats` | numeric chat ID **only** | `-1001234567890` | +| `deny_chats` | chat ID, `@username`, or title | `-1001234567890`, `ops-room` | +| `allow_senders` | numeric user ID **only** | `42` | +| `deny_senders` | user ID, `@username`, or profile name | `42`, `mallory` | + +A group's title is set by whoever runs it and a `@username` is claimable and +movable, so treating either as grantable would let anyone create a group +named `ops-room` and walk into a grant meant for someone else's. Both stay +deny-eligible, where a spoofable name can only subtract access. + +**Never collected:** private chats, the bot's own messages, other bots, and +`/pair` commands. ## Slack @@ -1194,6 +1253,65 @@ slack: Deny rules alone never enable access. - If a required name lookup fails or omits data, Nerve refuses the message. +### Channel source + +Channel traffic can feed the source inbox, where `poll_source`, `read_source`, +and the `messages` cron gate read it. Nothing here starts an agent turn. It is +a grant of its own, independent of the access rules above: see +[sources.md](sources.md#chat-channels-slack-telegram) for the routing rules +and the threat model. + +```yaml +slack: + source: + enabled: true + allow_channels: ["C0123ABCD", "eng-*"] + deny_channels: ["*-social"] + deny_senders: ["*-bot"] + schedule: "*/5 * * * *" # how often the buffer drains into the inbox + batch_size: 50 + max_stored_messages: 10000 # all Slack channels together +``` + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `slack.source.enabled` | bool | `false` | Feed the inbox from watched channels | +| `slack.source.allow_channels` | list[str] | `[]` | Channels to collect. **Empty means none** | +| `slack.source.deny_channels` | list[str] | `[]` | Never collect these | +| `slack.source.allow_senders` | list[str] | `[]` | Restrict to these senders | +| `slack.source.deny_senders` | list[str] | `[]` | Skip these senders | +| `slack.source.include_handled_messages` | bool | `false` | Also collect messages accepted for live routing | +| `slack.source.schedule` | cron/interval | `*/5 * * * *` | Drain cadence | +| `slack.source.batch_size` | int | `50` | Records per drain | +| `slack.source.condense` | bool | `false` | LLM-condense long messages | +| `slack.source.max_stored_messages` | int | `10000` | Buffer trim target for all Slack channels together; checked every 100 writes | + +Reaches the inbox as the source `slack:observed`. + +**What the lists match.** Case-insensitive, `*` / `?` globs, and a leading `#` +or `@` is stripped so a name pasted from Slack works as-is. Deny always wins; +a non-empty allow list must match. + +| List | Matches | Example | +|---|---|---| +| `allow_channels` / `deny_channels` | channel ID, or channel name | `C0123ABCD`, `eng-backend`, `eng-*` | +| `allow_senders` | member ID, handle, or email | `U0456DEFG`, `alice`, `*@example.com` | +| `deny_senders` | same, **plus** display and real names | `*-bot`, `Alice Smith` | + +A display name is edited by its owner, so it can only ever deny — the handle +and email are workspace-assigned, so they may grant. + +**Never collected:** DMs, group DMs, the bot's own posts, other apps, and +join/leave noise. A group DM arrives as `channel_type="mpim"` on a `G` id, and +a `G` whose type cannot be established is skipped rather than guessed at. + +**Prefer IDs for a busy channel.** Names and globs use cached Slack API lookups +(`conversations.info` or `users.info`). An ID skips the lookup, but only in +uppercase: `beckyjones` is a legal handle and `c0456def` a legal channel name, +so a lowercase pattern is resolved rather than assumed. It still matches. It +just costs the lookup. A channel the grant does not name is refused before any +sender is resolved. + ### Message behavior - In an allowed DM, the bot answers every message. @@ -1289,9 +1407,16 @@ Three differences from the inbound policy: ## Sources (sync) -Sources pull data from external services on a schedule. See [sources.md](sources.md) for full details. +Most sources pull data from external services on a schedule. See +[sources.md](sources.md) for full details. + +Chat channels also feed the inbox, but are configured on the channel — see +`slack.source.*` and `telegram.source.*` above. For each message delivered by +the channel transport, matching messages are written to a local buffer. +`ChannelSource` drains that buffer; it does not poll the chat service. Its +`schedule` is a drain cadence, not a poll interval. -**Common fields** (available on all sources): +**Common pull-source fields** (under `sync.`; availability varies): | Key | Type | Default | Description | |-----|------|---------|-------------| diff --git a/docs/sources.md b/docs/sources.md index 30aa4790..4e117a2d 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -2,9 +2,12 @@ ## Overview -Sources are cursor-based data streams that pull records from external services and persist them to a local inbox (`source_messages` table). The architecture follows a **producer/consumer** pattern inspired by Kafka: +Sources are cursor-based data streams that persist records to a local inbox +(`source_messages` table). Most fetch an external service. Channel sources +drain records that Slack or Telegram already delivered to a local buffer. The +architecture follows a **producer/consumer** pattern inspired by Kafka: -- **Producers** (source runners) — Fetch records from external APIs, preprocess and condense them, persist to the inbox, and advance the source cursor. No agent processing happens here. +- **Producers** (source runners) — Fetch or drain records, preprocess and condense them, persist to the inbox, and advance the source cursor. No agent processing happens here. - **Consumers** (agent tools) — Read from the inbox using independent persistent cursors. Multiple consumers can read the same messages without interfering with each other. ``` @@ -31,7 +34,7 @@ CONSUMERS (agent tools): ### Ingestion (Producer Side) -1. **Fetch** — The source adapter calls an external API (gh CLI, gog CLI, Telethon) and returns normalized `SourceRecord` objects with an opaque cursor +1. **Fetch** — The source adapter reads an external service (gh CLI, gog CLI, Telethon) or a local channel buffer and returns normalized `SourceRecord` objects with an opaque cursor 2. **Preprocess** — Two-stage content cleanup: - **Source-specific** (`source.preprocess()`) — Each source can override this for programmatic cleanup. Gmail strips boilerplate paragraphs (legal disclaimers, unsubscribe blocks, tracking URLs). Default: no-op - **LLM condensation** (`condense: true`) — Records still over 800 chars are sent to a fast model (Haiku) that extracts only essential information. Configurable per source, runs concurrently with a 30s timeout per record, falls back to original content on failure @@ -159,8 +162,10 @@ visible rather than failing the fetch. - **Note:** GitHub's Events API returns truncated payloads (e.g., PR titles and URLs may be missing). The source constructs URLs from repo name + number and handles missing fields gracefully - **Default schedule:** `*/15 * * * *` (every 15 min) -### Telegram -- **Adapter:** `nerve/sources/telegram.py` — uses Telethon (user account API) +### Telegram account sync +- **Adapter:** `nerve/sources/telegram.py` — uses Telethon (user account API). + This is the pull sync for your own Telegram account, and is separate from the + `telegram:observed` bot-channel source below - **Mechanism:** Telegram's native `updates.getDifference` API — asks "what's new since this state?" using PTS/QTS/date - **Cursor:** JSON-encoded Telegram state `{pts, qts, date, seq}` — the server's own update tracking - **First run:** Calls `updates.getState()` to snapshot current position, returns 0 records (prevents flooding the agent with history) @@ -168,9 +173,124 @@ visible rather than failing the fetch. - **Default schedule:** `*/5 * * * *` (every 5 min) - **Requires setup:** Run `nerve sync telegram` interactively once to authenticate with Telethon (phone number + code). The session is stored at `~/.nerve/telegram_sync.session` +### Chat channels (Slack, Telegram) + +A chat channel can feed the inbox with the group traffic it already receives. + +- **Adapter:** `nerve/sources/channel.py` — one `ChannelSource` per enabled + channel source +- **Mechanism:** for each message delivered by the channel transport, matching + messages are written to the `channel_observations` buffer. `ChannelSource` + drains that buffer into the inbox; it does not poll the chat service. +- **Source name:** `:observed` — `slack:observed`, + `telegram:observed` — so a cron gate reads `sources: [slack:observed]`. + Compound like `gmail:`, and deliberately not the bare channel name: + `telegram` is already the Telethon pull source for your *user* account, and + sharing a name would mean sharing a cron job id and a cursor key. +- **Config:** `slack.source.*` / `telegram.source.*`, not `sync.*` — see + [config.md](config.md). The runner carries its own `schedule` because there + is no `config.sync.` section to look one up in. +- **Default schedule:** `*/5 * * * *` +- **Cursor:** the buffer's autoincrement row id. `AUTOINCREMENT` is + load-bearing: the buffer is pruned, and a plain SQLite rowid is reused after + the highest row is deleted, which would reissue ids the cursor has already + passed and skip everything after them. +- **Idempotent:** the record id is `:`, so the same + message collected twice collapses on the inbox's `(source, id)` key. +- **Buffer target:** `max_stored_messages` (default 10 000) is one budget per + *transport*, covering all of its watched conversations together, not one per + conversation. Every 100 writes, Nerve drops the oldest rows back to that + target and logs a warning. The buffer can therefore exceed it by up to 99 + rows. Values below 100 are refused and 100 used instead: `0` reads like "no + limit" but tells the trim to keep nothing. Daily cleanup removes rows after + `sync.message_ttl_days`. +- **Backlog:** one run keeps fetching while the buffer reports more, up to 20 + batches, so a chat busier than one `batch_size` per tick still drains. Each + batch is persisted before the next starts, so a run that stops at the bound + resumes there and logs a warning. Raise `batch_size` if that repeats. +- **Requires setup (Telegram only):** with privacy mode enabled, a non-admin bot + does not receive ordinary group messages. Make it an admin or disable privacy + mode with BotFather. See Telegram's + [privacy-mode documentation](https://core.telegram.org/bots/faq#what-messages-will-my-bot-get). + +#### Routing: which messages get collected + +For eligible shared-channel and group-chat messages, the live route and source +route use separate rules. Access rules (`slack.allow_users`, +`telegram.allowed_users`, …) decide who may drive the agent. +`.source.*` decides whose traffic may reach the inbox. A message can +take either route, both routes, or neither. + +| Live route | Source grant | `include_handled_messages` | Result | +|---|---|---|---| +| accepts | no match | — | channel only | +| accepts | matches | `false` (default) | channel only | +| accepts | matches | `true` | channel **and** source | +| refuses | matches | — | source only | +| refuses | no match | — | dropped | + +By default, a message accepted by the live route is not also copied to the +source. Set `include_handled_messages: true` only when source consumers should +receive the same message later. "Handled" means accepted for live routing, not +that the agent produced a reply. + +Whether the live route accepts differs by transport, and only affects which row +of that table you land in: + +- **Slack** accepts a shared-channel message for live routing when it mentions + the bot or continues a thread with an active session, and the sender passes + the access policy. A matching source can still collect a message that fails + the live access policy. +- **Telegram** has no "addressed to me" test: authorization alone decides. + +**Never collected, whatever the config says:** direct and group DMs, the +agent's own messages, other bots and apps, join/leave and similar service +events, malformed events, and Telegram `/pair` commands. Duplicate buffered +records collapse to one inbox record on the `(source, id)` key. + +**The source is off by default and fail-closed.** An empty allow list collects +nothing rather than everything — the inverse of the access rules, because this +is a standing grant to record other people's messages. `enabled: true` with +nothing listed logs a warning and registers no drain. + +#### Threat model + +**Everything collected is untrusted input.** Most of it comes from people not +authorized to instruct the agent — that is the usual reason to watch a room — +so treat a buffered message as attacker-controlled text an agent will later +read. What actually protects you: + +- `.source.*` decides **whose messages are collected** before anything + is written. +- At drain time, Nerve rechecks conversation and sender deny rules against raw + IDs. Name-based denies apply only before buffering. +- Inbox filters inspect metadata, not content. They cannot distinguish a report + from an instruction. Polling advances a cursor; it does not delete inbox + records. Daily cleanup removes them after their TTL expires. Agents see them + only through explicit source reads or cron gates, never in a live turn. + +So scope the grant to conversations whose participants you would already trust +to file a ticket, and treat any workflow that acts on collected content without +review as accepting prompt injection. + +**Slack lookup cost.** Names and globs use cached `conversations.info` or +`users.info` lookups. An ID skips the lookup, but only in uppercase. A +lowercase pattern could equally be a name, so it is resolved rather than +assumed. A channel the grant does not name is refused before any sender is +resolved. + +**Reply context is not expanded.** Slack stores `thread_ts`; Telegram stores +`reply_to_message_id`. The parent message is not fetched. + +**Attachments are named, not fetched.** An upload contributes `[File: ]`, +`[Photo]`, or `[Sticker: ]` ahead of any caption. A message with neither +text nor an attachment is not collected. + ## Configuration -Sources are configured under the `sync:` key in `config.yaml` / `config.local.yaml`: +Pull sources are configured under the `sync:` key in `config.yaml` / +`config.local.yaml`. Channel sources use `slack.source.*` and +`telegram.source.*` instead. ```yaml sync: @@ -480,6 +600,7 @@ The Sources page (`/sources`) has three tabs: - `consumer_cursors` — Per (consumer, source) read position with TTL and session linking - `source_messages` — Inbox messages with `raw_content` (original HTML), `processed_content` (LLM-condensed), TTL-based expiry - `source_run_log` — Per-run diagnostics (records ingested, errors, timestamps) +- `channel_observations` — Push buffer for chat messages a channel collected for the inbox, drained by `ChannelSource`. Trimmed toward a per-transport row target and TTL-swept by daily cleanup - `cron_logs` — Job execution history (source jobs use `source:` as job ID) ### API Endpoints diff --git a/nerve/bootstrap.py b/nerve/bootstrap.py index b0770fd0..0470c9e7 100644 --- a/nerve/bootstrap.py +++ b/nerve/bootstrap.py @@ -201,7 +201,7 @@ class SetupChoices: user_name: str = "" telegram_bot_token: str = "" # Telegram user IDs authorized to DM the bot. Empty = pair after setup - # via `nerve pair` + /pair . + # via `nerve pair` + /pair in a private chat. telegram_allowed_users: list[int] = field(default_factory=list) password: str = "" # plaintext during wizard, hashed at write time enabled_crons: list[str] = field(default_factory=list) @@ -1218,7 +1218,8 @@ def _step_channels(self) -> None: " numeric Telegram user ID to authorize yourself now\n" " (message @userinfobot on Telegram to get it).\n\n" " Or press Enter to skip — after setup, run 'nerve pair'\n" - " and send the bot /pair to authorize.", + " and send the bot /pair in a private chat to " + "authorize.", dim=True, ) click.echo() @@ -2504,7 +2505,7 @@ def _done(self) -> None: if self.choices.telegram_bot_token and not self.choices.telegram_allowed_users: click.secho(" Telegram pairing:", bold=True) click.echo(" 1. Start Nerve, then run: nerve pair") - click.echo(" 2. Send the bot: /pair ") + click.echo(" 2. In a private chat, send the bot: /pair ") click.echo(" (until paired, the bot ignores all DMs)") click.echo() click.secho( diff --git a/nerve/channels/base.py b/nerve/channels/base.py index 44b03a4f..da4b1a9b 100644 --- a/nerve/channels/base.py +++ b/nerve/channels/base.py @@ -67,6 +67,32 @@ class OutboundMessage: metadata: dict[str, Any] = field(default_factory=dict) +@dataclass +class ObservedMessage: + """A message a channel collected for the source inbox. + + Not an :class:`InboundMessage`: nothing here starts an agent turn. It is + a record headed for the source inbox, where the existing consumer tools + and cron gates can act on it — so the fields are the ones a reader needs + to make sense of a line of chat, not the ones the router needs to route. + + Names are left empty when unresolved. Collection runs on the dispatch + path and a display name costs an API call, so the channel buffers raw IDs + and a reader resolves them later — or does not, if nothing asked. + """ + + channel_name: str # "slack", "telegram" + channel_key: str # "slack:C0123ABCD:1700000000.000100" + conversation_id: str # "C0123ABCD" + sender_id: str # "U0456DEFG" + text: str + message_id: str # transport-native id (Slack ts) + timestamp: str # ISO 8601 + conversation_title: str = "" # "" until resolved + sender_name: str = "" # "" until resolved + metadata: dict[str, Any] = field(default_factory=dict) + + class BaseChannel(abc.ABC): """Abstract base for all communication channels. diff --git a/nerve/channels/observation.py b/nerve/channels/observation.py new file mode 100644 index 00000000..02901dcf --- /dev/null +++ b/nerve/channels/observation.py @@ -0,0 +1,71 @@ +"""Who the agent may watch — a grant distinct from who may command it. + +An access policy answers "may this person drive the agent?". This one +answers "may this conversation feed the agent's inbox?". They are not the +same question, and conflating them fails in both directions: reusing the +access policy either blocks watching a channel the agent takes no orders +from, or silently widens command access to everything worth watching. + +So the source gets its own gate, composed from the same +:mod:`nerve.channels.access` primitives, and the two are evaluated +independently: a message may be live-routed, collected, both, or neither. This +gate is off unless configured, and a conversation must be named explicitly — +there is no "watch everything the bot can see" by omission, because that is +what a misconfiguration looks like. + +Most of what it approves comes from people who are *not* authorized to +instruct the agent — that is the usual reason to watch a room. Everything +buffered is untrusted. This gate limits which messages are stored; neither it +nor the inbox filter validates message content. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from nerve.channels.access import Decision, Identity, PatternGate + + +@dataclass +class ObservationPolicy: + """Whether a conversation and sender may be buffered to the inbox. + + ``conversations`` is fail-closed by design: an empty allow list collects + nothing at all, rather than everything. That inverts + :class:`~nerve.channels.access.PatternGate`'s default, which is right for + an access check composed after a user gate and wrong for a standing grant + to record other people's messages. + + ``senders`` is the opposite — usually empty, meaning "anyone talking in an + approved conversation". Narrowing it to specific people is possible but + unusual; the interesting unit here is the room, not the speaker. A deny + list is the common use: skip a noisy bot. + """ + + enabled: bool = False + conversations: PatternGate = field( + default_factory=lambda: PatternGate("conversation"), + ) + senders: PatternGate = field(default_factory=lambda: PatternGate("sender")) + + @property + def active(self) -> bool: + """Whether this policy can ever approve anything.""" + return self.enabled and bool(self.conversations.allow) + + def check(self, conversation: Identity, sender: Identity) -> Decision: + """Decide whether one message may be buffered.""" + if not self.enabled: + return Decision(False, "the channel source is not enabled") + if not self.conversations.allow: + return Decision( + False, + "no conversations are approved for the channel source", + ) + verdict = self.conversations.check(conversation) + if not verdict.allowed: + return verdict + return self.senders.check(sender) + + +__all__ = ["ObservationPolicy"] diff --git a/nerve/channels/router.py b/nerve/channels/router.py index 9587e4cf..a5b69cd8 100644 --- a/nerve/channels/router.py +++ b/nerve/channels/router.py @@ -13,6 +13,7 @@ import asyncio import logging import uuid +from dataclasses import asdict from typing import Any, TYPE_CHECKING from nerve.agent.interactive import get_handler @@ -22,9 +23,11 @@ BaseChannel, ChannelCapability, InboundMessage, + ObservedMessage, OutboundMessage, ) from nerve.channels.stream_adapter import StreamAdapter +from nerve.db.observations import DEFAULT_MAX_ROWS if TYPE_CHECKING: from nerve.agent.engine import AgentEngine @@ -366,6 +369,44 @@ async def send_file( target = ctx["target"] if ctx and ctx.get("channel_name") == channel else "" return await chan_obj.send_file(target, file_path) + # ------------------------------------------------------------------ # + # Observation buffer # + # ------------------------------------------------------------------ # + + async def observe( + self, + msg: ObservedMessage, + ttl_days: int = 7, + max_stored_messages: int = DEFAULT_MAX_ROWS, + ) -> bool: + """Buffer a message a channel collected for the source inbox. + + Channels reach the database through the router, never through the + engine directly, so this is the seam. Returns True if the record was + buffered. + + A failure is swallowed and logged. This sits on a channel's dispatch + path, ahead of any live handling, so a database hiccup must not take + down message handling for the sake of a buffered copy. + """ + db = getattr(self.engine, "db", None) + if db is None: + return False + try: + await db.insert_channel_observation( + channel=msg.channel_name, + channel_key=msg.channel_key, + payload=asdict(msg), + ttl_days=ttl_days, + max_rows=max_stored_messages, + ) + return True + except Exception as e: + logger.warning( + "Failed to buffer an observation from %s: %s", msg.channel_name, e, + ) + return False + # ------------------------------------------------------------------ # # Interactive tool response routing # # ------------------------------------------------------------------ # diff --git a/nerve/channels/slack.py b/nerve/channels/slack.py index d0dd7ccb..173bd577 100644 --- a/nerve/channels/slack.py +++ b/nerve/channels/slack.py @@ -13,10 +13,16 @@ import logging import re import time +from datetime import datetime, timezone from pathlib import Path from typing import Any, TYPE_CHECKING -from nerve.channels.access import Decision, Identity, needs_name_resolution +from nerve.channels.access import ( + Decision, + Identity, + PatternGate, + needs_name_resolution, +) from nerve.channels.archives import ( IMAGE_EXT_TO_MIME, MAX_TEXT_SIZE, @@ -28,8 +34,10 @@ ChannelCapability, ChannelConstraints, InboundMessage, + ObservedMessage, OutboundMessage, ) +from nerve.channels.observation import ObservationPolicy from nerve.channels.slack_access import SlackAccessPolicy from nerve.channels.slack_presentation import ( MAX_MSG_LEN, @@ -110,14 +118,17 @@ # Slack object ids: a type letter then uppercase alphanumerics. U/W users, -# B bots, C/G/D/T conversations and teams. Used to decide whether an -# allow/deny pattern can be matched against the id alone, so the shape has to -# be exact — anything looser skips a name lookup a deny list depends on. +# B bots, C/G/D/T conversations and teams. Decides whether a pattern can be +# matched against the id alone, so anything looser skips a name lookup a deny +# list depends on. Case-sensitive: `beckyjones` is a legal handle and +# `c0456def` a legal channel name, so only a lookup tells a lowercase id from +# a lowercase name. Matching is case-insensitive either way, so a lowercase id +# still matches. It just costs the lookup. _SLACK_ID_RE = re.compile(r"^[UWBCDGT][A-Z0-9]{7,}$") def is_slack_id(pattern: str) -> bool: - """Whether *pattern* is a literal Slack object id rather than a name.""" + """Whether *pattern* can only be a literal Slack object id, never a name.""" return bool(_SLACK_ID_RE.match(pattern)) @@ -136,6 +147,30 @@ def parse_target(target: str) -> tuple[str, str | None]: return channel_id, (thread_ts if sep and thread_ts else None) +def _describe_slack_files(files: list[dict[str, Any]]) -> str: + """Name a message's attachments without downloading them. + + For an upload posted with no comment the file name is the whole message. + ``_extract_files`` says more but fetches the bytes to do it. + """ + return " ".join( + f"[File: {f.get('name') or 'unnamed'}]" for f in files + ) + + +def slack_ts_to_iso(ts: str) -> str: + """Turn a Slack ``.`` stamp into ISO 8601 UTC. + + Falls back to now rather than raising. An unreadable stamp should cost an + observation its exact time, not drop the message. + """ + try: + seconds = float(ts) + except (AttributeError, TypeError, ValueError): + return datetime.now(timezone.utc).isoformat() + return datetime.fromtimestamp(seconds, tz=timezone.utc).isoformat() + + class SlackUnavailable(RuntimeError): """The channel cannot carry traffic for the generation now running.""" @@ -931,6 +966,136 @@ async def _should_answer( return bool(await self.router.get_last_session(channel_key)) return False + @property + def observation(self) -> ObservationPolicy: + """The source grant, rebuilt per read so reloads apply at once.""" + source = self.config.slack.source + return ObservationPolicy( + enabled=source.enabled, + conversations=PatternGate( + "conversation", + allow=list(source.allow_conversations), + deny=list(source.deny_conversations), + ), + senders=PatternGate( + "sender", + allow=list(source.allow_senders), + deny=list(source.deny_senders), + ), + ) + + async def _observe( + self, + event: dict[str, Any], + channel_id: str, + user_id: str, + ts: str, + channel_key: str, + handled: bool = False, + ) -> None: + """Buffer a message for the source inbox, if the source grant allows. + + Asked of every shared-channel message, whatever the live route + decided: the two grants are independent, so being answered neither + earns nor forfeits a place in the inbox. ``handled`` says the live + route accepted this message, and by default that is where it stops — + one message should not arrive twice. ``include_handled_messages`` + turns the copy back on for a source meant as a record. + + Private conversations are never a source. A DM the agent declined to + answer is a refusal, and quietly filing it away is not what "we do not + talk to you" led the sender to expect. That covers multi-person DMs, + which arrive as ``channel_type="mpim"`` on a ``G`` id — checking only + for ``D`` would file a group DM away as if it were a channel. + + Raw IDs are buffered and names are resolved only when a pattern needs + one, so watching a busy channel costs no Slack API call per message. + """ + # Cheapest gates first: this runs on every shared-channel message + # now, not only the ones the live route declined. + source = self.config.slack.source + if not source.enabled: + return + if handled and not source.include_handled_messages: + return + policy = self.observation + if not policy.active: + return + # The *raw* type, not the caller's derived one: that default turns an + # absent type into "channel", which is exactly the ambiguity this has + # to refuse rather than resolve. + declared = event.get("channel_type") or "" + if declared in ("im", "mpim") or channel_id.startswith("D"): + return + # A `G` is either a legacy private channel or a multi-person DM, and + # only the declared type distinguishes them cheaply. Without one, + # decline: the source is opt-in, so not recording something is + # always the safe outcome. + if channel_id.startswith("G") and declared not in ("channel", "group"): + logger.debug( + "Slack did not observe a message in %s: conversation type %r " + "does not distinguish a private channel from a group DM", + channel_id, declared or "unset", + ) + return + + # Conversation first: it settles most refusals, and an id-only channel + # grant needs no lookup. Resolving the sender first would cost one per + # speaker in every channel the bot sits in, watched or not. + conversation = await self._identify_conversation( + channel_id, + "channel", + needs_name_resolution(policy.conversations, is_id=is_slack_id), + ) + verdict = policy.conversations.check(conversation) + if not verdict.allowed: + logger.debug("Slack did not observe a message: %s", verdict.reason) + return + + sender = await self._identify_user( + user_id, + needs_name_resolution(policy.senders, is_id=is_slack_id), + need_email=policy.senders.any_deny_pattern(lambda p: "@" in p), + ) + verdict = policy.senders.check(sender) + if not verdict.allowed: + logger.debug("Slack did not observe a message: %s", verdict.reason) + return + + text = slack_to_plain(event.get("text") or "", self._bot_user_id) + attachments = _describe_slack_files(event.get("files") or []) + if attachments: + text = f"{attachments}\n\n{text}" if text else attachments + # Neither text nor a file name is nothing worth a row. + if not text: + return + + observed = ObservedMessage( + channel_name="slack", + channel_key=channel_key, + conversation_id=channel_id, + sender_id=user_id, + text=text, + message_id=ts, + timestamp=slack_ts_to_iso(ts), + conversation_title=next(iter(conversation.names), ""), + sender_name=next( + iter((*sender.names, *sender.self_set_names)), "", + ), + # thread_ts lets a reader pull the parent later. A reply without + # its thread is often meaningless, and expanding it here would + # cost an API call per observation. + metadata={ + "thread_ts": event.get("thread_ts") or "", + "subtype": event.get("subtype") or "", + }, + ) + await self.router.observe( + observed, + ttl_days=self.config.sync.message_ttl_days, + max_stored_messages=source.max_stored_messages, + ) + async def _handle_message_event(self, event: dict[str, Any]) -> None: """Turn a Slack message into an InboundMessage and hand it to the router.""" if self._is_own_message(event): @@ -961,9 +1126,25 @@ async def _handle_message_event(self, event: dict[str, Any]) -> None: target = format_target(channel_id, thread_ts) channel_key = f"slack:{target}" - if not await self._should_answer(event, channel_type, channel_key): - return - if not await self._authorize(user_id, channel_id, channel_type): + # Two independent routes. Live handling needs the message to be + # addressed to the agent *and* its sender authorized; the source + # needs its own grant and neither of those. Both are asked, then + # `handled` reconciles them, so an addressed message from someone + # refused still reaches a source that asked for that channel. + # + # The authorize call stays behind the addressed check: it is the + # expensive one, and running it for every remark in a busy channel + # would cost a lookup and a refusal log line per message. + handled = ( + await self._should_answer(event, channel_type, channel_key) + and await self._authorize(user_id, channel_id, channel_type) + ) + # Sits below the early returns above on purpose, so our own posts, + # join/leave noise, and other apps never reach the inbox. + await self._observe( + event, channel_id, user_id, ts, channel_key, handled=handled, + ) + if not handled: return text = slack_to_plain(event.get("text") or "", self._bot_user_id) diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index e53fa952..4ce5585c 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -19,7 +19,7 @@ import time from datetime import datetime, timezone from pathlib import Path -from typing import Any, Callable, TYPE_CHECKING +from typing import Any, Awaitable, Callable, TYPE_CHECKING from zoneinfo import ZoneInfo from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update @@ -32,13 +32,16 @@ TEXT_EXTENSIONS, extract_zip, ) +from nerve.channels.access import Identity, PatternGate from nerve.channels.base import ( BaseChannel, ChannelCapability, ChannelConstraints, InboundMessage, + ObservedMessage, OutboundMessage, ) +from nerve.channels.observation import ObservationPolicy from nerve.config import NerveConfig if TYPE_CHECKING: @@ -226,6 +229,23 @@ def _format_reply_context(message: Any) -> str: return "\n".join(parts) +def _describe_telegram_attachment(message: Any) -> str: + """Name a message's attachment without downloading it. + + An uncaptioned photo or sticker still says something. ``_extract_sticker`` + and ``_extract_document`` say more but fetch the file to do it. + """ + sticker = getattr(message, "sticker", None) + if sticker: + return f"[Sticker: {sticker.emoji}]" if sticker.emoji else "[Sticker]" + document = getattr(message, "document", None) + if document: + return f"[File: {document.file_name or 'unnamed'}]" + if getattr(message, "photo", None): + return "[Photo]" + return "" + + # Inline-keyboard /sessions rendering ------------------------------------- # _SESSIONS_PAGE_SIZE = 10 # sessions shown per /sessions page; ⬅️/➡️ page the rest _SESSION_LABEL_MAX = 40 # Telegram wraps long button labels poorly @@ -558,18 +578,42 @@ def _build_application(self) -> Application: ) app = builder.build() - # Register handlers - app.add_handler(CommandHandler("start", self._handle_start)) - app.add_handler(CommandHandler("pair", self._handle_pair)) - app.add_handler(CommandHandler("session", self._handle_session)) - app.add_handler(CommandHandler("sessions", self._handle_sessions)) - app.add_handler(CommandHandler("star", self._handle_star)) - app.add_handler(CommandHandler("unstar", self._handle_unstar)) - app.add_handler(CommandHandler("new", self._handle_new_session)) - app.add_handler(CommandHandler("stop", self._handle_stop)) - app.add_handler(CommandHandler("restart", self._handle_restart)) - app.add_handler(CommandHandler("doctor", self._handle_doctor)) - app.add_handler(CommandHandler("reply", self._handle_reply)) + # A CommandHandler wins over the generic MessageHandler in this group, + # so commands need the source hook on their own callback. + app.add_handler(CommandHandler( + "start", self._source_routed_command(self._handle_start), + )) + # Never persist a one-time pairing code in the source buffer. + app.add_handler(CommandHandler( + "pair", self._source_routed_command(self._handle_pair, collect=False), + )) + app.add_handler(CommandHandler( + "session", self._source_routed_command(self._handle_session), + )) + app.add_handler(CommandHandler( + "sessions", self._source_routed_command(self._handle_sessions), + )) + app.add_handler(CommandHandler( + "star", self._source_routed_command(self._handle_star), + )) + app.add_handler(CommandHandler( + "unstar", self._source_routed_command(self._handle_unstar), + )) + app.add_handler(CommandHandler( + "new", self._source_routed_command(self._handle_new_session), + )) + app.add_handler(CommandHandler( + "stop", self._source_routed_command(self._handle_stop), + )) + app.add_handler(CommandHandler( + "restart", self._source_routed_command(self._handle_restart), + )) + app.add_handler(CommandHandler( + "doctor", self._source_routed_command(self._handle_doctor), + )) + app.add_handler(CommandHandler( + "reply", self._source_routed_command(self._handle_reply), + )) app.add_handler(CallbackQueryHandler(self._handle_callback_query)) app.add_handler(MessageHandler( filters.TEXT | filters.PHOTO | filters.COMMAND | filters.Sticker.ALL | filters.Document.ALL, @@ -640,7 +684,8 @@ def _announce_auth_state(self) -> None: code = pairing.get_or_create_pairing_code() logger.info( "Telegram: no allowed_users configured — pairing mode active. " - "Send the bot: /pair %s to authorize your account " + "In a private chat, send the bot: /pair %s to authorize " + "your account " "(code valid for 1h; run `nerve pair` to get a fresh one).", code, ) @@ -1019,6 +1064,31 @@ def _touch(self) -> None: """Record that we received an update from Telegram.""" self._last_update_time = time.monotonic() + def _source_routed_command( + self, + callback: Callable[[Update, Any], Awaitable[None]], + *, + collect: bool = True, + ) -> Callable[[Update, Any], Awaitable[None]]: + """Add source routing to a command without changing PTB precedence.""" + + async def wrapped(update: Update, context: Any) -> None: + chat = update.effective_chat + user = update.effective_user + message = update.message + if chat is None or user is None or message is None: + return + + source = self.config.telegram.source + if collect and source.enabled and source.allow_conversations: + await self._observe( + update, + handled=self._is_authorized(user.id), + ) + await callback(update, context) + + return wrapped + async def _handle_start(self, update: Update, context: Any) -> None: """Handle /start command.""" self._touch() @@ -1034,7 +1104,8 @@ async def _handle_start(self, update: Update, context: Any) -> None: await update.message.reply_text( "This Nerve instance isn't paired with your account.\n" f"Your Telegram ID: {user_id}\n\n" - "To pair, run `nerve pair` on the server, then send me:\n" + "To pair, run `nerve pair` on the server, then open a " + "private chat with me and send:\n" "/pair " ) return @@ -1043,9 +1114,21 @@ async def _handle_start(self, update: Update, context: Any) -> None: ) async def _handle_pair(self, update: Update, context: Any) -> None: - """Handle /pair — authorize a user via a one-time pairing code.""" + """Authorize a user who sends /pair in a private chat.""" self._touch() - user_id = update.effective_user.id + chat = update.effective_chat + user = update.effective_user + message = update.message + if chat is None or user is None or message is None: + return + if ( + chat.type != "private" + or user.is_bot + or getattr(message, "sender_chat", None) is not None + ): + logger.warning("Ignoring /pair outside a private human chat") + return + user_id = user.id if self._is_authorized(user_id): await update.message.reply_text("Already paired — you're authorized.") @@ -1488,10 +1571,149 @@ async def _extract_zip( return extract_zip(bytes(data), meta_line) + @property + def observation(self) -> ObservationPolicy: + """The source grant, rebuilt per read so reloads apply at once.""" + source = self.config.telegram.source + return ObservationPolicy( + enabled=source.enabled, + conversations=PatternGate( + "chat", + allow=list(source.allow_conversations), + deny=list(source.deny_conversations), + ), + senders=PatternGate( + "sender", + allow=list(source.allow_senders), + deny=list(source.deny_senders), + ), + ) + + async def _observe(self, update: Update, handled: bool = False) -> None: + """Buffer a group message for the source inbox, if the grant allows. + + Asked of every group message, whatever ``telegram.allowed_users`` + decided: the source is its own grant, so a sender being authorized + neither earns nor forfeits a place in the inbox. ``handled`` says the + live route accepted this message, and by default that is where it + stops — one message should not arrive twice. + ``include_handled_messages`` turns the copy back on for a source + meant as a record. + + Private chats are never a source. A stranger's DM is a refusal, and + filing it away is not what the silence led them to expect; a group an + operator listed is a different matter. + + The bot must be a group administrator, or have privacy mode disabled + via BotFather, for Telegram to deliver ordinary group traffic at all. + Without that it only sees commands and replies to itself, and this + collects almost nothing however the grant is written. + """ + # Cheapest gates first: this runs on every message now, not only the + # ones from a sender the allowlist refused. + source = self.config.telegram.source + if not source.enabled: + return + if handled and not source.include_handled_messages: + return + policy = self.observation + if not policy.active: + return + chat = update.effective_chat + user = update.effective_user + message = update.message + if chat is None or user is None or message is None: + return + if chat.type == "private": + return + # Pairing codes are credentials. A command addressed to another bot + # bypasses our CommandHandler and reaches the generic message handler, + # so enforce this exclusion at the source gate too. + text = message.text or message.caption or "" + words = text.split(maxsplit=1) + command = words[0].partition("@")[0].casefold() if words else "" + if command == "/pair": + return + # Telegram delivers other bots' messages to group handlers under + # bot-to-bot mode. Slack drops them via _is_another_app_talking, and + # two agents feeding each other's inboxes is no better than two + # agents answering each other. + if user.is_bot: + return + + # Only the numeric id is stable. A chat title is set by whoever runs + # the group, so treating it as allow-eligible would let anyone create + # a group called "ops-room" and walk into a grant meant for someone + # else's; a @username is claimable and movable for the same reason. + # Both stay deny-eligible, where a spoofable name can only ever + # subtract access. + conversation = Identity( + id=str(chat.id), + self_set_names=tuple(n for n in (chat.username, chat.title) if n), + ) + sender = Identity( + id=str(user.id), + self_set_names=tuple( + n for n in ( + user.username, user.full_name, user.first_name, user.last_name, + ) if n + ), + ) + + verdict = policy.check(conversation, sender) + if not verdict.allowed: + logger.debug("Telegram did not observe a message: %s", verdict.reason) + return + + text = message.text or message.caption or "" + attachment = _describe_telegram_attachment(message) + if attachment: + text = f"{attachment}\n\n{text}" if text else attachment + # Neither text nor an attachment is nothing worth a row. + if not text: + return + + sent_at = message.date or datetime.now(timezone.utc) + observed = ObservedMessage( + channel_name="telegram", + channel_key=f"telegram:{chat.id}", + conversation_id=str(chat.id), + sender_id=str(user.id), + text=text, + message_id=str(message.message_id), + timestamp=sent_at.isoformat(), + conversation_title=chat.title or chat.username or "", + sender_name=user.username or user.full_name or "", + metadata={ + "chat_type": chat.type or "", + "reply_to_message_id": ( + str(message.reply_to_message.message_id) + if message.reply_to_message + else "" + ), + }, + ) + await self.router.observe( + observed, + ttl_days=self.config.sync.message_ttl_days, + max_stored_messages=source.max_stored_messages, + ) + async def _handle_message(self, update: Update, context: Any) -> None: """Handle incoming text and photo messages — delegate to router.""" self._touch() - if not self._is_authorized(update.effective_user.id): + chat = update.effective_chat + user = update.effective_user + message = update.message + if chat is None or user is None or message is None: + return + # Two independent routes, as on Slack. Authorization decides the live + # one; the source has its own grant and is asked either way, so a + # group can feed the inbox whether or not its members may also drive + # the agent. + handled = self._is_authorized(user.id) + await self._observe(update, handled=handled) + if not handled: return # Media group (album) — collect all parts before processing diff --git a/nerve/cli.py b/nerve/cli.py index 9374aba3..4a6c2dbc 100644 --- a/nerve/cli.py +++ b/nerve/cli.py @@ -1103,7 +1103,7 @@ def doctor_report(config, config_source: str = "", check_api: bool = False) -> s warnings.append( "[WARN] telegram.allowed_users is empty — the bot rejects " "all DMs until you pair (run 'nerve pair', then send the " - "bot /pair )" + "bot /pair in a private chat)" ) else: errors.append("[ERR] Telegram enabled but bot_token not set") @@ -1745,8 +1745,8 @@ def codex_doctor(ctx: click.Context, json_output: bool) -> None: def pair(ctx: click.Context) -> None: """Generate a Telegram pairing code. - Send /pair to your bot within an hour to authorize your - Telegram account. The paired user ID is persisted to + Send /pair to your bot in a private chat within an hour to + authorize your Telegram account. The paired user ID is persisted to config.local.yaml (telegram.allowed_users). """ config = ctx.obj["config"] @@ -1773,7 +1773,7 @@ def pair(ctx: click.Context) -> None: click.echo() click.secho(f" Pairing code: {code}", bold=True) click.echo() - click.echo(f" Send this to your bot: /pair {code}") + click.echo(f" In a private chat, send this to your bot: /pair {code}") click.echo(f" Valid for {CODE_TTL_SECONDS // 60} minutes, single use.") if not running: click.echo() diff --git a/nerve/config.py b/nerve/config.py index 0aea2480..1b85f7cd 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -940,6 +940,203 @@ def context_1m_enabled_for(self, model: str | None) -> bool: ) +_DEFAULT_CHANNEL_SOURCE_SCHEDULE = "*/5 * * * *" + + +def _pattern_list(value: object, label: str) -> list[str]: + """Coerce a ``source`` allow/deny value to a list of patterns. + + Stricter than the generic coercion, because these lists decide whose + messages get recorded. The case that matters is a **mapping**: + ``allow_conversations: {"*": false}`` survives the general coercion + unchanged, and ``list()`` of it yields its *keys* — so a rule that reads + like a disabled wildcard silently becomes one that grants everything. + A mapping is discarded rather than salvaged. + + A bare scalar still wraps to a single pattern. That is how a ``${VAR}`` + env reference arrives, and dropping it would break the documented way to + configure a list field from the environment. + """ + if value is None: + return [] + if isinstance(value, (dict, set)): + logger.warning( + "source.%s must be a list of patterns, got %s — discarding it. " + "A mapping's keys would read as patterns, so a rule that looks " + "switched off would grant them.", + label, type(value).__name__, + ) + return [] + if not isinstance(value, (list, tuple)): + text = _strip_display_sigil(value) + return [text] if text else [] + patterns: list[str] = [] + for entry in value: + if isinstance(entry, (dict, list, tuple, set)): + logger.warning( + "source.%s: ignoring non-scalar entry %r", label, entry, + ) + continue + text = _strip_display_sigil(entry) + if text: + patterns.append(text) + return patterns + + +def _strip_display_sigil(value: object) -> str: + """Drop a leading ``#`` or ``@`` from a pattern. + + Chat UIs render a channel as ``#eng-backend`` and a person as ``@alice``, + so that is what an operator copies — but the API names carry no sigil and + matching is literal. Left alone, ``#eng-backend`` matches nothing, which + on an allow list means an inbox that stays empty with no error to explain + why. + + Unambiguous to strip: neither platform lets a channel, chat, or user name + begin with ``#`` or ``@``. A ``@`` *inside* the value is untouched, so an + email pattern like ``*@example.com`` still works. + """ + text = str(value).strip() + return text[1:].strip() if text[:1] in "#@" else text + + +_MIN_STORED_MESSAGES = 100 + + +def _stored_message_target(value: object, default: int) -> int: + """Coerce a ``source.max_stored_messages``, holding it above a floor. + + Zero is the case to catch: it reads like "no limit" and tells the trim to + keep no rows, emptying the buffer before the drain reads it. A negative + does the same, since SQLite reads a negative OFFSET as zero. + """ + target = _lenient_int(value, default, label="source.max_stored_messages") + if target >= _MIN_STORED_MESSAGES: + return target + logger.warning( + "source.max_stored_messages is %r, which would trim the buffer to " + "nothing before the drain reads it. Using %d instead.", + value, _MIN_STORED_MESSAGES, + ) + return _MIN_STORED_MESSAGES + + +def _channel_source_schedule(value: object) -> str: + """Coerce a ``source.schedule``, falling back to the default. + + A channel source carries its own schedule, so an unusable value has + no ``sync.`` section to fall back to — it would leave the channel + buffering with nothing ever draining it. Fall back loudly instead. + """ + if value is None: + return _DEFAULT_CHANNEL_SOURCE_SCHEDULE + if not isinstance(value, str) or not value.strip(): + logger.warning( + "source.schedule must be a non-empty crontab or interval " + "string, got %r — falling back to %r. Left unset, the buffer " + "would fill with nothing draining it.", + value, _DEFAULT_CHANNEL_SOURCE_SCHEDULE, + ) + return _DEFAULT_CHANNEL_SOURCE_SCHEDULE + return value.strip() + + +@dataclass +class ChannelSourceConfig: + """Which conversations feed the inbox, independent of live routing. + + A separate grant from the access rules, and evaluated independently of + them. "May this person drive the agent?" and "may this room's traffic + reach the agent's inbox?" are different questions. A message can qualify + for the live route, the source route, both, or neither. Deriving one answer + from the other would block watching a channel the agent takes no orders + from, or widen command access to everything worth watching. + + Fail-closed twice over: off unless ``enabled``, and collecting nothing + unless the allow list names something. An empty allow list here means + "nothing", not "everything" — the opposite of the access gates, because + this one is a standing grant to record other people's messages rather + than a check that already ran a sender rule first. + + The fields are transport-neutral, but the YAML keys are not: each channel + passes the ``subject`` its own users would recognise, so Slack reads + ``allow_channels`` and Telegram reads ``allow_chats``. Sharing one noun + was worse than it looks — on Telegram a "channel" is a specific entity + type distinct from a group, so ``allow_channels`` there would name the + wrong thing, and "conversations" named nothing anyone could act on. + + What actually matches is per transport and documented in ``config.md``; + the short version is that a platform id always works, and a name works + only where the platform, not the subject, controls it. + + ``schedule`` is the drain cadence, not a poll: the messages are already + in the buffer by the time it fires. + """ + + enabled: bool = False + allow_conversations: list[str] = field(default_factory=list) + deny_conversations: list[str] = field(default_factory=list) + allow_senders: list[str] = field(default_factory=list) + deny_senders: list[str] = field(default_factory=list) + schedule: str = _DEFAULT_CHANNEL_SOURCE_SCHEDULE + batch_size: int = 50 + # Off by default: most chat messages are shorter than the runner's + # 800-char condense threshold, so this would build an LLM client that + # never gets used. + condense: bool = False + # Trim target per transport, checked every 100 writes. One budget for all + # of a transport's watched conversations, not one each. + max_stored_messages: int = 10_000 + # Whether a message accepted for live routing is also collected. + # Off by default. Turn it on only when source consumers should receive the + # same message later, such as when the source is an archive or digest. + include_handled_messages: bool = False + + @classmethod + @_coerced + def from_dict( + cls, d: dict, *, subject: str = "conversations", + ) -> ChannelSourceConfig: + """Parse a ``.source`` block. + + ``subject`` is the noun this transport calls the thing being watched + — ``"channels"`` for Slack, ``"chats"`` for Telegram — and names both + the YAML keys and every warning, so an operator is told about the key + they actually wrote. + """ + enabled = _as_bool( + d.get("enabled", False), False, label="source.enabled", + ) + allow_key, deny_key = f"allow_{subject}", f"deny_{subject}" + allow = _pattern_list(d.get(allow_key), allow_key) + if enabled and not allow: + logger.warning( + "source.enabled is set but source.%s names nothing usable — " + "nothing will be collected. List the %s to watch; an empty " + "list is not a wildcard here.", + allow_key, subject, + ) + return cls( + enabled=enabled, + allow_conversations=allow, + deny_conversations=_pattern_list(d.get(deny_key), deny_key), + allow_senders=_pattern_list(d.get("allow_senders"), "allow_senders"), + deny_senders=_pattern_list(d.get("deny_senders"), "deny_senders"), + schedule=_channel_source_schedule(d.get("schedule")), + batch_size=d.get("batch_size", 50), + condense=_as_bool( + d.get("condense", False), False, label="source.condense", + ), + max_stored_messages=_stored_message_target( + d.get("max_stored_messages"), 10_000, + ), + include_handled_messages=_as_bool( + d.get("include_handled_messages", False), False, + label="source.include_handled_messages", + ), + ) + + @dataclass class TelegramConfig: enabled: bool = True @@ -953,6 +1150,8 @@ class TelegramConfig: # agent access for any Telegram user. A warning # is logged at startup. dm_policy: str = "pairing" + # Chats whose traffic feeds the inbox — see ChannelSourceConfig. + source: ChannelSourceConfig = field(default_factory=ChannelSourceConfig) @classmethod @_coerced @@ -996,6 +1195,9 @@ def from_dict(cls, d: dict, locked: bool = False) -> TelegramConfig: allowed_users=d.get("allowed_users") or [], stream_mode=d.get("stream_mode", "partial"), dm_policy=dm_policy, + source=ChannelSourceConfig.from_dict( + d.get("source", {}), subject="chats", + ), ) @@ -1077,6 +1279,9 @@ class SlackConfig: # None keeps safe defaults; [] disables commands. Host-wide and # cross-channel commands are opt-in. See SLACK_*_COMMANDS. commands: list[str] | None = None + # Channels whose traffic feeds the inbox. Its own grant, not derived + # from allow_channels — see ChannelSourceConfig. + source: ChannelSourceConfig = field(default_factory=ChannelSourceConfig) @classmethod @_coerced @@ -1114,6 +1319,9 @@ def from_dict(cls, d: dict, locked: bool = False) -> SlackConfig: ), stream_mode=stream_mode, commands=_slack_commands(d.get("commands")), + source=ChannelSourceConfig.from_dict( + d.get("source", {}), subject="channels", + ), ) diff --git a/nerve/cron/service.py b/nerve/cron/service.py index 707e7925..0c116abc 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -547,9 +547,21 @@ async def _reload_from_disk( def _source_schedule(self, runner) -> str | None: """The configured schedule for *runner*, or ``None`` if it has no config. - Source names can be compound (e.g. "gmail:account@email.com"). The - config key is the base type before the colon. + A runner that carries its own schedule wins. Sources configured + outside ``config.sync`` — a channel observation drain, say — have no + section here to look up, and returning ``None`` for them means the + runner is built and then silently never scheduled. + + Otherwise the source name picks the section. Names can be compound + (e.g. "gmail:account@email.com"); the config key is the base type + before the colon. """ + # Must actually be a string. This reads a duck-typed attribute, and a + # test double hands back a truthy stand-in for anything asked of it — + # which would put a MagicMock where a crontab expression belongs. + carried = getattr(runner, "schedule", "") + if isinstance(carried, str) and carried: + return carried config_key = runner.source.source_name.split(":")[0] source_config = getattr(self.config.sync, config_key, None) if source_config is None: @@ -1287,16 +1299,18 @@ async def _run_source_wrapper(self, runner: SourceRunner) -> None: ) async def _cleanup_expired(self) -> None: - """Clean up expired source messages, consumer cursors, and old cron logs.""" + """Clean up expired source messages, cursors, cron logs, and observations.""" try: msg_count = await self.db.cleanup_expired_messages() cursor_count = await self.db.cleanup_expired_consumer_cursors() cron_log_count = await self.db.cleanup_old_cron_logs(days=14) - if msg_count or cursor_count or cron_log_count: + obs_count = await self.db.cleanup_expired_channel_observations() + if msg_count or cursor_count or cron_log_count or obs_count: logger.info( "Cleanup: %d expired messages, %d expired consumer cursors, " - "%d cron logs older than 14 days", - msg_count, cursor_count, cron_log_count, + "%d cron logs older than 14 days, %d expired channel " + "observations", + msg_count, cursor_count, cron_log_count, obs_count, ) except Exception as e: logger.error("Cleanup failed: %s", e, exc_info=True) diff --git a/nerve/db/base.py b/nerve/db/base.py index 0fc4c531..8c9ed88f 100644 --- a/nerve/db/base.py +++ b/nerve/db/base.py @@ -23,6 +23,7 @@ from nerve.db.messages import MessageStore from nerve.db.migrations.runner import discover_migrations, run_migrations from nerve.db.notifications import NotificationStore +from nerve.db.observations import ObservationStore from nerve.db.plans import PlanStore from nerve.db.review_loops import ReviewLoopStore from nerve.db.sessions import SessionStore @@ -95,6 +96,7 @@ class Database( TaskStatusStore, PlanStore, NotificationStore, + ObservationStore, SourceStore, CronStore, SkillStore, diff --git a/nerve/db/migrations/v046_channel_observations.py b/nerve/db/migrations/v046_channel_observations.py new file mode 100644 index 00000000..03461143 --- /dev/null +++ b/nerve/db/migrations/v046_channel_observations.py @@ -0,0 +1,56 @@ +"""V46: durable buffer for messages a channel saw but did not answer. + +Sources are pull, cursor, and cron. Channels are push. This table is the +join between them: the channel appends on the dispatch path, and a +``ChannelSource`` drains it on the source runner's cadence, which buys +filtering, condensing, TTL, health, and cursor handling for free. + +The alternative — writing straight to ``source_messages`` from the socket — +would skip the inbox guardrail, which is the one layer standing between +untrusted chat text and an autonomous agent. Buffering first keeps that +choke point where it already is. + +``AUTOINCREMENT`` is load-bearing rather than decorative. A plain rowid is +reused after the highest row is deleted, and this table is pruned by design, +so a drained-and-pruned buffer would hand out ids the cursor has already +passed and the next observations would be skipped. AUTOINCREMENT gives a +strictly increasing id that survives pruning, which is what makes +``WHERE id > cursor`` correct. + +Payload is JSON rather than columns because the shape belongs to +:class:`~nerve.channels.base.ObservedMessage`, not to the database: only the +drain reads it, and a channel gaining a field should not need a migration. +The columns that exist are the ones the drain and the pruner filter on. +""" + +from __future__ import annotations + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + +SQL = """ +CREATE TABLE IF NOT EXISTS channel_observations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + channel TEXT NOT NULL, + channel_key TEXT NOT NULL DEFAULT '', + payload TEXT NOT NULL, + created_at TEXT NOT NULL, + expires_at TEXT NOT NULL +); + +-- The drain: one channel's backlog past a cursor, in id order. +CREATE INDEX IF NOT EXISTS idx_channel_observations_drain + ON channel_observations(channel, id); + +-- The TTL sweep. +CREATE INDEX IF NOT EXISTS idx_channel_observations_expires + ON channel_observations(expires_at); +""" + + +async def up(db: aiosqlite.Connection) -> None: + await db.executescript(SQL) + logger.info("v046: channel_observations buffer created") diff --git a/nerve/db/observations.py b/nerve/db/observations.py new file mode 100644 index 00000000..455df98f --- /dev/null +++ b/nerve/db/observations.py @@ -0,0 +1,149 @@ +"""Channel observation buffer — the push-to-pull join for the sources layer. + +A channel appends here on its dispatch path; a +:class:`~nerve.sources.channel.ChannelSource` drains it on the source +runner's cadence. See ``v046_channel_observations`` for why the id is +``AUTOINCREMENT`` and the payload is JSON. +""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timedelta, timezone +from typing import Any + +logger = logging.getLogger(__name__) + +# A busy transport must not fill the disk between drains. Past this many rows +# for one transport, the oldest are dropped — losing the stale end of a +# backlog nobody drained beats losing the daemon. Trimming is amortized +# (see _TRIM_EVERY) so the dispatch path stays a single INSERT. +DEFAULT_MAX_ROWS = 10_000 +_TRIM_EVERY = 100 + + +class ObservationStore: + """Mixin for the ``channel_observations`` buffer.""" + + @property + def _observation_writes(self) -> dict[str, int]: + """Transport name -> inserts since that transport's last trim check. + + Built on first use: mixins here have no ``__init__``, and a class + attribute would share one counter across every Database instance. + """ + counts = self.__dict__.get("_observation_write_counts") + if counts is None: + counts = self.__dict__["_observation_write_counts"] = {} + return counts + + # Longest a buffered message may be before it is truncated. Slack caps + # posts near 40k and Telegram near 4k, so this only bites on a pathological + # sender — but the cap is per row and the row count is capped separately, + # so without it the two together still bound nothing in bytes. + MAX_OBSERVED_TEXT = 16_000 + + async def insert_channel_observation( + self, + channel: str, + channel_key: str, + payload: dict[str, Any], + ttl_days: int = 7, + max_rows: int = DEFAULT_MAX_ROWS, + ) -> int: + """Append one observation. Returns its id. + + This runs on the channel's dispatch path, so it is one INSERT and + nothing else. The row cap is enforced once every ``_TRIM_EVERY`` + inserts rather than on each one; overshooting the cap by under a + hundred rows is cheaper than a COUNT per message. + """ + text = payload.get("text") + if isinstance(text, str) and len(text) > self.MAX_OBSERVED_TEXT: + payload = { + **payload, + "text": text[: self.MAX_OBSERVED_TEXT], + "truncated": True, + } + + now = datetime.now(timezone.utc) + result = await self._write( + "INSERT INTO channel_observations " + "(channel, channel_key, payload, created_at, expires_at) " + "VALUES (?, ?, ?, ?, ?)", + ( + channel, + channel_key, + json.dumps(payload), + now.isoformat(), + (now + timedelta(days=ttl_days)).isoformat(), + ), + ) + + # Start at the threshold rather than zero, so the first insert after + # a restart trims. A purely process-local counter never fires on a + # daemon that restarts more often than it writes _TRIM_EVERY rows, + # and the cap silently stops existing. + seen = self._observation_writes.get(channel, _TRIM_EVERY) + 1 + if seen >= _TRIM_EVERY: + self._observation_writes[channel] = 0 + await self._trim_channel_observations(channel, max_rows) + else: + self._observation_writes[channel] = seen + + return result.lastrowid or 0 + + async def _trim_channel_observations(self, channel: str, max_rows: int) -> None: + """Drop the oldest rows for one transport past ``max_rows``.""" + result = await self._write( + "DELETE FROM channel_observations WHERE channel = ? AND id <= (" + " SELECT id FROM channel_observations WHERE channel = ?" + " ORDER BY id DESC LIMIT 1 OFFSET ?" + ")", + (channel, channel, max_rows), + ) + if result.rowcount: + logger.warning( + "Transport %s observation buffer exceeded its %d-row target — " + "dropped %d of the oldest rows. The drain is behind or not " + "scheduled.", + channel, max_rows, result.rowcount, + ) + + async def read_channel_observations( + self, channel: str, after_id: int = 0, limit: int = 50, + ) -> list[tuple[int, dict[str, Any] | None]]: + """Buffered rows for one transport past ``after_id``, oldest first. + + Every scanned row comes back as ``(id, payload)``, with ``payload`` + None where the JSON would not parse. Dropping those rows here + instead would hide their ids from the caller, and a batch of + entirely unreadable rows would then pin the cursor in place and be + re-read on every run, with everything behind them unreachable. + Reporting them lets the drain skip the row and still move past it. + """ + rows: list[tuple[int, dict[str, Any] | None]] = [] + async with self.db.execute( + "SELECT id, payload FROM channel_observations " + "WHERE channel = ? AND id > ? ORDER BY id LIMIT ?", + (channel, after_id, limit), + ) as cursor: + async for row in cursor: + try: + rows.append((row[0], json.loads(row[1]))) + except (ValueError, TypeError) as e: + logger.warning( + "Skipping unreadable observation %s on %s: %s", + row[0], channel, e, + ) + rows.append((row[0], None)) + return rows + + async def cleanup_expired_channel_observations(self) -> int: + """Delete observations past their TTL. Returns count deleted.""" + now = datetime.now(timezone.utc).isoformat() + result = await self._write( + "DELETE FROM channel_observations WHERE expires_at < ?", (now,), + ) + return result.rowcount or 0 diff --git a/nerve/pairing.py b/nerve/pairing.py index 15ceaae1..2cf6811f 100644 --- a/nerve/pairing.py +++ b/nerve/pairing.py @@ -6,7 +6,7 @@ 1. A code is generated on the server — either automatically at channel startup (fresh install with no ``allowed_users``) or on demand via ``nerve pair``. - 2. The user sends ``/pair `` to the bot. + 2. The user sends ``/pair `` to the bot in a private chat. 3. On a match the user's ID is appended to ``telegram.allowed_users`` in config.local.yaml and authorized immediately. diff --git a/nerve/sources/channel.py b/nerve/sources/channel.py new file mode 100644 index 00000000..d3ceaf38 --- /dev/null +++ b/nerve/sources/channel.py @@ -0,0 +1,137 @@ +"""Drain the channel observation buffer into the source inbox. + +Chat arrives by push; sources are pull, cursor, and cron. Rather than teach +the sources layer about sockets — or poll a chat API that already delivered +the same messages, paying twice in latency and rate limit for a second +cursor to disagree with — the channel buffers what it saw and this source +drains the buffer. + +Everything past that is inherited. :class:`~nerve.sources.runner.SourceRunner` +supplies filtering, condensing, TTL, health, and cursor advance, and the +existing ``poll_source`` / ``read_source`` tools and ``MessagesGate`` work +against the result with no channel-specific code anywhere in this layer. + +The cursor is the buffer's autoincrement id, which is why the buffer exists: +a monotonic integer that survives pruning makes ``WHERE id > cursor`` +trivially correct, where a chat timestamp would not be. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any + +from nerve.sources.base import Source +from nerve.sources.models import FetchResult, SourceRecord + +if TYPE_CHECKING: + from nerve.db import Database + +logger = logging.getLogger(__name__) + +_SUMMARY_PREVIEW = 80 + + +class ChannelSource(Source): + """A source over one channel's observation buffer. + + The source name is ``:observed`` — compound like + ``gmail:``, and distinct on purpose. ``telegram`` is already a + pull source, and sharing a name would mean sharing a job id + (``source:telegram``) and a cursor key: the two runners would evict each + other from the scheduler and then read each other's cursor, one an + integer and the other Telethon's JSON state. So a cron gate reads + ``sources: [slack:observed]``, and observed traffic stays legible as its + own stream rather than blending into a pull source's. + """ + + def __init__(self, channel: str, db: Database): + self.channel = channel + self.source_name = f"{channel}:observed" + self._db = db + + async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: + """Read buffered observations past *cursor*. + + A malformed cursor is treated as "start from the beginning" rather + than an error: the buffer is TTL-bounded, so the worst case is + re-reading a bounded backlog, and ``source_messages`` is keyed + ``(source, id)``, which makes that a no-op instead of a duplicate. + """ + after_id = _as_id(cursor) + try: + rows = await self._db.read_channel_observations( + self.channel, after_id=after_id, limit=limit, + ) + except Exception as e: + logger.error( + "Channel source %s: reading the buffer failed: %s", + self.channel, e, exc_info=True, + ) + return FetchResult(records=[], next_cursor=cursor) + + if not rows: + return FetchResult(records=[], next_cursor=cursor, has_more=False) + + # The cursor tracks what was *scanned*, not what parsed. An + # unreadable row still has to move it, or the drain re-reads it + # every run and never reaches what is behind it. + return FetchResult( + records=[ + _to_record(self.channel, self.source_name, payload) + for _, payload in rows + if isinstance(payload, dict) + ], + next_cursor=str(rows[-1][0]), + has_more=len(rows) >= limit, + ) + + +def _as_id(cursor: str | None) -> int: + """The buffer id a cursor names, or 0.""" + if not cursor: + return 0 + try: + return int(cursor) + except (TypeError, ValueError): + logger.warning("Ignoring unreadable channel cursor %r", cursor) + return 0 + + +def _to_record( + channel: str, source_name: str, payload: dict[str, Any], +) -> SourceRecord: + """Turn one buffered :class:`ObservedMessage` payload into a record.""" + conversation_id = payload.get("conversation_id") or "" + message_id = payload.get("message_id") or "" + text = payload.get("text") or "" + title = payload.get("conversation_title") or conversation_id + sender = payload.get("sender_name") or payload.get("sender_id") or "unknown" + + preview = text[:_SUMMARY_PREVIEW].replace("\n", " ") + if len(text) > _SUMMARY_PREVIEW: + preview += "..." + + # The record id is the transport's own address, not the buffer id, so a + # message observed twice collapses on the inbox's (source, id) key + # instead of arriving twice. + return SourceRecord( + id=f"{conversation_id}:{message_id}", + source=source_name, + record_type=f"{channel}_message", + summary=f"[{title}] {sender}: {preview}", + content=text, + timestamp=payload.get("timestamp") or "", + metadata={ + "conversation_id": conversation_id, + "conversation_title": payload.get("conversation_title") or "", + "sender_id": payload.get("sender_id") or "", + "sender_name": payload.get("sender_name") or "", + "message_id": message_id, + "channel_key": payload.get("channel_key") or "", + **(payload.get("metadata") or {}), + }, + ) + + +__all__ = ["ChannelSource"] diff --git a/nerve/sources/registry.py b/nerve/sources/registry.py index 34446119..fa6ea41b 100644 --- a/nerve/sources/registry.py +++ b/nerve/sources/registry.py @@ -240,4 +240,61 @@ def build_source_runners( gh_repos.batch_size, gh_repos.repos or "none", ) + # Channel source drains. Not a poll: the channel transport already + # delivered and buffered these events. This only moves them into the + # inbox. What to watch is a property of the channel, so the config lives at + # slack.source / telegram.source rather than under sync.*, and the + # runner carries its own schedule instead of being looked up there. + for channel_name, channel_config in ( + ("slack", config.slack), + ("telegram", config.telegram), + ): + source_cfg = getattr(channel_config, "source", None) + if source_cfg is None or not source_cfg.enabled: + continue + if not source_cfg.allow_conversations: + # ChannelSourceConfig.from_dict already warned. Don't build a + # runner whose policy can never approve anything to drain. + continue + from nerve.sources.channel import ChannelSource + from nerve.sources.filters import FieldRule, InboxFilter + + # Re-apply the deny rules at drain time. The channel already ran the + # full policy before buffering, so this is a second net, not the gate: + # it catches rows written before a conversation was added to the deny + # list, which the buffer would otherwise deliver minutes or a TTL + # later under the old policy. + # + # Deny-only on purpose. These rules match the id field, so a + # name-based pattern simply will not match — harmless for a deny rule + # (the row passes, exactly as it does today) but fail-closed for an + # allow rule, which would drop every message from a policy written + # against channel names. The allow decision stays where it can be + # made correctly, at the channel, which knows which of a + # conversation's names are grantable. + source_filter = InboxFilter(rules=[ + FieldRule( + field="conversation_id", deny=list(source_cfg.deny_conversations), + ), + FieldRule(field="sender_id", deny=list(source_cfg.deny_senders)), + ]) + + runners.append(SourceRunner( + source=ChannelSource(channel_name, db), + db=db, + batch_size=source_cfg.batch_size, + condense=source_cfg.condense, + condense_model=condense_model, + condense_client_factory=condense_factory, + ttl_days=ttl_days, + inbox_filter=source_filter, + schedule=source_cfg.schedule, + )) + logger.info( + "Registered source: %s observations (batch=%d, schedule=%s, " + "conversations allow=%s deny=%s)", + channel_name, source_cfg.batch_size, source_cfg.schedule, + source_cfg.allow_conversations, source_cfg.deny_conversations or [], + ) + return runners diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index b122010e..c4c37360 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -6,6 +6,9 @@ consumption is handled separately by consumer tools. Pipeline: fetch → source.preprocess → persist → LLM condense → advance cursor + +A run repeats that pipeline while the source reports ``has_more``, so a backlog +deeper than ``batch_size`` drains in one run rather than one batch per tick. """ from __future__ import annotations @@ -91,6 +94,10 @@ def is_backed_off(self) -> bool: # otherwise drop hundreds at once). _MAX_DROP_LOG_LINES = 10 +# Batches one run drains before leaving the rest for the next tick. Bounds a +# very deep backlog, and a source whose has_more never goes false. +_MAX_PASSES = 20 + _CONDENSE_PROMPT = ( "Extract the essential information from this source record content.\n" "Rules:\n" @@ -105,6 +112,17 @@ def is_backed_off(self) -> bool: ) +@dataclass +class _BatchResult: + """One batch's outcome, plus what the drain loop needs to page onward.""" + + records_ingested: int + records_dropped: int = 0 + next_cursor: str | None = None + has_more: bool = False + error: str | None = None + + class SourceRunner: """Fetches records from a source and persists them to the inbox. @@ -126,6 +144,13 @@ class SourceRunner: persist. Records that don't pass are dropped — never written to the inbox, never seen by the agent. An inactive/None filter is a no-op. See :mod:`nerve.sources.filters`. + schedule: Crontab or interval this runner asks to be scheduled on. + Empty means "look me up in ``config.sync.``", which is how + every pull source works. A source configured somewhere else — + a channel's ``source`` block, say — carries its own cadence + here, because the alternative is a phantom ``config.sync`` + section that duplicates it, or a runner that is silently never + scheduled. See :meth:`CronService._source_schedule`. """ def __init__( @@ -139,6 +164,7 @@ def __init__( condense_client_factory: Callable[[], Any] | None = None, ttl_days: int = 7, inbox_filter: InboxFilter | None = None, + schedule: str = "", ): self.source = source self.db = db @@ -153,6 +179,7 @@ def __init__( self._client_factory = condense_client_factory self.ttl_days = ttl_days self.inbox_filter = inbox_filter + self.schedule = schedule self._lock = asyncio.Lock() self._condense_client: Any | None = None self.health = SourceHealth() @@ -229,8 +256,58 @@ async def run(self) -> IngestResult: return result async def _run_locked(self) -> IngestResult: - """Actual run logic, called under lock.""" + """Drain batches until the source reports no more, or _MAX_PASSES. + + A source arriving faster than one ``batch_size`` per tick never catches + up on one batch per run, and a buffered one is then trimmed from the + old end while the drain is still behind it. + + Each batch persists and advances the cursor before the next starts, so + stopping at the bound leaves progress durable and the next run resumes + there. + """ cursor = await self.db.get_sync_cursor(self.source.source_name) + total_ingested = 0 + total_dropped = 0 + + for pass_number in range(1, _MAX_PASSES + 1): + result = await self._ingest_batch(cursor) + if result.error: + # Earlier batches are persisted and their cursor moved. Report + # the failure, keep the progress. + return IngestResult( + records_ingested=total_ingested, + records_dropped=total_dropped, + error=result.error, + ) + total_ingested += result.records_ingested + total_dropped += result.records_dropped + + if not result.has_more: + break + # More to read but a standing cursor would re-read the same batch. + if result.next_cursor == cursor: + logger.warning( + "Source %s: reports more to read but did not advance its " + "cursor (%s). Stopping this run.", + self.source.source_name, cursor, + ) + break + cursor = result.next_cursor + if pass_number == _MAX_PASSES: + logger.warning( + "Source %s: still has a backlog after %d batches of %d. " + "Stopping until the next run. Raise batch_size or the " + "schedule if this repeats.", + self.source.source_name, _MAX_PASSES, self.batch_size, + ) + + return IngestResult( + records_ingested=total_ingested, records_dropped=total_dropped, + ) + + async def _ingest_batch(self, cursor: str | None) -> _BatchResult: + """One fetch → preprocess → persist → condense → advance cycle.""" logger.info( "Source %s: fetching (cursor=%s, batch_size=%d)", self.source.source_name, cursor, self.batch_size, @@ -240,7 +317,7 @@ async def _run_locked(self) -> IngestResult: result = await self.source.fetch(cursor, limit=self.batch_size) except Exception as e: logger.error("Source %s fetch failed: %s", self.source.source_name, e, exc_info=True) - return IngestResult(records_ingested=0, error=str(e)) + return _BatchResult(records_ingested=0, error=str(e)) if not result.records: # Even with 0 records, advance cursor if it changed @@ -253,7 +330,8 @@ async def _run_locked(self) -> IngestResult: ) else: logger.info("Source %s: no new records", self.source.source_name) - return IngestResult(records_ingested=0) + # No records means nothing to page past, whatever has_more says. + return _BatchResult(records_ingested=0, next_cursor=result.next_cursor) # Ingestion pipeline: # 1. Source-specific cleanup (e.g., Gmail boilerplate stripping) @@ -309,7 +387,12 @@ async def _run_locked(self) -> IngestResult: self.source.source_name, len(records), result.next_cursor, ) - return IngestResult(records_ingested=len(records), records_dropped=dropped_count) + return _BatchResult( + records_ingested=len(records), + records_dropped=dropped_count, + next_cursor=result.next_cursor, + has_more=result.has_more, + ) # ------------------------------------------------------------------ # Inbox persistence diff --git a/tests/slack_live.py b/tests/slack_live.py index 151cb6dc..c300be43 100644 --- a/tests/slack_live.py +++ b/tests/slack_live.py @@ -352,11 +352,22 @@ class RecordingRouter: def __init__(self, reply_text: str | None = None): self.messages: list[Any] = [] + self.observed: list[Any] = [] self.reply_text = reply_text self.channel: Any = None self._sessions: dict[str, str] = {} self._arrived = asyncio.Event() + async def observe(self, msg: Any, **kwargs: Any) -> bool: + """Record what the channel collected for the source inbox. + + The buffer itself is a plain SQLite table with its own unit tests. + What a live run settles is which real events reach here and what + Slack put in their fields. + """ + self.observed.append(msg) + return True + async def handle_message(self, msg: Any) -> str: self._sessions.setdefault(msg.channel_key, f"s{len(self._sessions)}") if self.reply_text and self.channel is not None: @@ -399,6 +410,61 @@ async def wait_for_message( f"within {timeout}s (saw {[m.text for m in self.messages]})", ) + def _matching_observed(self, marker: str) -> list[Any]: + return [m for m in self.observed if marker in (m.text or "")] + + async def wait_for_observation( + self, marker: str, timeout: float = EVENT_TIMEOUT, + ) -> Any: + """Block until an observation carrying *marker* is buffered.""" + deadline = asyncio.get_running_loop().time() + timeout + while asyncio.get_running_loop().time() < deadline: + found = self._matching_observed(marker) + if found: + return found[-1] + await asyncio.sleep(0.1) + raise AssertionError( + f"no observation carrying {marker!r} was buffered within " + f"{timeout}s (saw {[m.text for m in self.observed]})", + ) + + async def expect_no_observation( + self, marker: str, channel, timeout: float = EVENT_TIMEOUT, + ) -> None: + """Assert the source gate saw the event and declined it. + + Same reasoning as :meth:`expect_no_message`: an empty list proves + nothing until the envelope is known to have reached the channel. + """ + await self._await_arrival(marker, channel, timeout) + await asyncio.sleep(REFUSAL_SETTLE_SECONDS) + assert not self._matching_observed(marker), ( + f"the source collected {marker!r}, which the grant should have " + f"declined" + ) + + async def _await_arrival( + self, marker: str, channel, timeout: float, + ) -> None: + """Wait until an envelope carrying *marker* reached the channel.""" + diagnostics = getattr(channel._client, "_live_diagnostics", None) + assert diagnostics is not None, ( + "the live channel has no diagnostics, so a refusal cannot be " + "told apart from an event that never arrived" + ) + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if diagnostics.forwarded(marker): + return + await asyncio.sleep(0.1) + dropped = diagnostics.dropped(marker) + raise AssertionError( + f"no envelope carrying {marker!r} reached the channel within " + f"{timeout}s, so the gate was never exercised " + f"({'the harness dropped it as stale' if dropped else 'Slack never delivered it'})", + ) + async def expect_no_message( self, marker: str, channel, timeout: float = EVENT_TIMEOUT, ) -> None: diff --git a/tests/test_channel_observation.py b/tests/test_channel_observation.py new file mode 100644 index 00000000..f98ece6b --- /dev/null +++ b/tests/test_channel_observation.py @@ -0,0 +1,1426 @@ +"""Channel → source bridge — what gets watched, buffered, and drained. + +Three layers, tested separately because they fail differently: + +* the observation policy, which is the one place a mistake is a security + bug rather than a missing feature; +* the buffer, whose id must stay monotonic across pruning or the drain + silently skips messages; +* :class:`ChannelSource`, which turns buffered rows into inbox records and + hands the rest — filtering, TTL, health, cursor — to ``SourceRunner``. + +What is *not* here is anything that turns on the events Slack actually +sends. These hand ``_observe`` an event dict the test wrote, so they settle +the policy branches and nothing about whether a real message carries the +fields the payload is built from. Those live in +``TestChannelSourceCollectsRealTraffic`` in +:mod:`tests.test_slack_live_inbound`, driven by a real person posting into a +real channel. Kept here: the pure branches, and the conversation kinds a +live test cannot provoke. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nerve.channels.access import Identity, PatternGate +from nerve.channels.base import ObservedMessage +from nerve.channels.observation import ObservationPolicy +from nerve.channels.router import ChannelRouter +from nerve.channels.slack import SlackChannel +from nerve.config import ( + ChannelSourceConfig, + NerveConfig, + SlackConfig, + TelegramConfig, +) +from nerve.db.observations import _TRIM_EVERY +from nerve.sources.channel import ChannelSource +from nerve.sources.models import FetchResult, SourceRecord +from nerve.sources.registry import build_source_runners +from nerve.sources.runner import _MAX_PASSES, SourceRunner + +pytestmark = pytest.mark.asyncio + + +# ---------------------------------------------------------------------- # +# Helpers # +# ---------------------------------------------------------------------- # + + +def _policy(**kwargs) -> ObservationPolicy: + return ObservationPolicy( + enabled=kwargs.pop("enabled", True), + conversations=PatternGate( + "conversation", + allow=kwargs.pop("allow_conversations", ["C0123ABCD"]), + deny=kwargs.pop("deny_conversations", []), + ), + senders=PatternGate( + "sender", + allow=kwargs.pop("allow_senders", []), + deny=kwargs.pop("deny_senders", []), + ), + ) + + +def _slack_channel( + router=None, allow_channels=None, **source_kwargs, +) -> SlackChannel: + """A Slack channel with a stub transport and an observe policy. + + ``allow_channels`` is the *access* grant, deliberately separate from the + observe kwargs — the two policies are independent and the tests here + depend on that. + """ + cfg = NerveConfig() + cfg.slack = SlackConfig( + enabled=True, + bot_token="xoxb-test", + app_token="xapp-test", + allow_channels=list(allow_channels or []), + source=ChannelSourceConfig(**source_kwargs), + ) + channel = SlackChannel(cfg, router=router or MagicMock()) + channel._web = MagicMock() + channel._web.chat_postMessage = AsyncMock(return_value={"ts": "1.1"}) + channel._web.conversations_info = AsyncMock( + return_value={"channel": {"name": "general"}}, + ) + channel._web.users_info = AsyncMock( + return_value={"user": {"name": "alice", "profile": {}}}, + ) + channel._state = "running" + channel._bot_user_id = "U0BOT" + if router is None: + channel.router.observe = AsyncMock(return_value=True) + channel.router.get_last_session = AsyncMock(return_value=None) + channel.router.handle_message = AsyncMock(return_value="done") + return channel + + +def _event(**kwargs) -> dict: + base = { + "type": "message", + "channel": "C0123ABCD", + "channel_type": "channel", + "user": "U0456DEFG", + "ts": "1700000000.000100", + "text": "just chatting", + } + base.update(kwargs) + return base + + +# ---------------------------------------------------------------------- # +# Observation policy — the part where a mistake is a security bug # +# ---------------------------------------------------------------------- # + + +class TestObservationPolicy: + def test_an_allowed_conversation_is_observed(self): + verdict = _policy().check(Identity(id="C0123ABCD"), Identity(id="U1")) + + assert verdict.allowed + + def test_disabled_observes_nothing(self): + verdict = _policy(enabled=False).check( + Identity(id="C0123ABCD"), Identity(id="U1"), + ) + + assert not verdict.allowed + assert "not enabled" in verdict.reason + + def test_an_empty_allow_list_means_nothing_not_everything(self): + # The inverse of PatternGate's own default, and the whole reason + # this policy exists separately. A standing grant to record other + # people's messages must be written down, not inferred from silence. + verdict = _policy(allow_conversations=[]).check( + Identity(id="C0123ABCD"), Identity(id="U1"), + ) + + assert not verdict.allowed + assert "no conversations are approved" in verdict.reason + + def test_an_unlisted_conversation_is_refused(self): + verdict = _policy().check(Identity(id="C0999ZZZZ"), Identity(id="U1")) + + assert not verdict.allowed + + def test_a_denied_conversation_is_refused(self): + verdict = _policy( + allow_conversations=["*"], deny_conversations=["C0999ZZZZ"], + ).check(Identity(id="C0999ZZZZ"), Identity(id="U1")) + + assert not verdict.allowed + assert "deny" in verdict.reason + + def test_a_denied_sender_is_refused_in_an_allowed_conversation(self): + verdict = _policy(deny_senders=["U0BOT"]).check( + Identity(id="C0123ABCD"), Identity(id="U0BOT"), + ) + + assert not verdict.allowed + + def test_active_requires_both_enabled_and_a_grant(self): + assert _policy().active + assert not _policy(enabled=False).active + assert not _policy(allow_conversations=[]).active + + def test_observation_is_not_the_access_policy(self): + # Observing a room the agent takes no orders from is the point. + # If these two were the same object, enabling one would enable the + # other, which is the failure this separation exists to prevent. + cfg = SlackConfig(allow_channels=["C0AAA1111"]) + channel = _slack_channel( + enabled=True, allow_conversations=["C0BBB2222"], + ) + channel.config.slack.allow_channels = cfg.allow_channels + + assert channel.policy.channels.allow == ["C0AAA1111"] + assert channel.observation.conversations.allow == ["C0BBB2222"] + + +# ---------------------------------------------------------------------- # +# Slack hook # +# ---------------------------------------------------------------------- # + + +class TestSlackRouting: + """The two routes are decided independently, then reconciled. + + Live handling needs the message addressed to the agent and its sender + authorized; the source needs its own grant and neither of those. Every + combination is reachable, so every combination is pinned here. + """ + + async def test_source_only_when_the_sender_may_not_drive_the_agent(self): + # Addressed, but access refuses it. The source grant is its own + # question, so the message still reaches the inbox — which is the + # point of watching a channel the agent takes no orders from. + channel = _slack_channel( + allow_channels=["C0OTHER"], + enabled=True, + allow_conversations=["C0123ABCD"], + ) + + await channel._handle_message_event( + _event(text="<@U0BOT> hello", type="app_mention"), + ) + + channel.router.handle_message.assert_not_awaited() + channel.router.observe.assert_awaited_once() + + async def test_channel_only_when_the_source_does_not_want_it(self): + channel = _slack_channel( + allow_channels=["C0123ABCD"], + enabled=True, + allow_conversations=["C0AAA1111"], + ) + + await channel._handle_message_event( + _event(text="<@U0BOT> hello", type="app_mention"), + ) + + channel.router.handle_message.assert_awaited_once() + channel.router.observe.assert_not_awaited() + + async def test_neither_route_takes_an_unaddressed_unwatched_message(self): + channel = _slack_channel( + allow_channels=["C0123ABCD"], + enabled=True, + allow_conversations=["C0AAA1111"], + ) + + await channel._handle_message_event(_event()) + + channel.router.handle_message.assert_not_awaited() + channel.router.observe.assert_not_awaited() + + +class TestSlackObserve: + async def test_observation_off_buffers_nothing(self): + channel = _slack_channel(enabled=False, allow_conversations=["C0123ABCD"]) + + await channel._handle_message_event(_event()) + + channel.router.observe.assert_not_awaited() + + async def test_a_direct_message_is_never_observed(self): + # Declining to answer a DM is a refusal. Filing it away instead is + # not what the silence led the sender to expect. + channel = _slack_channel(enabled=True, allow_conversations=["*"]) + + await channel._handle_message_event( + _event(channel="D0123ABCD", channel_type="im"), + ) + + channel.router.observe.assert_not_awaited() + + async def test_a_group_dm_is_never_observed(self): + # An MPIM arrives as channel_type="mpim" on a `G` id. Checking only + # for `D` would file a group DM away as if it were a channel. + channel = _slack_channel(enabled=True, allow_conversations=["*"]) + + await channel._handle_message_event( + _event(channel="G0123ABCD", channel_type="mpim"), + ) + + channel.router.observe.assert_not_awaited() + + async def test_a_private_channel_is_observed(self): + # The other half of the `G` ambiguity: a real private channel must + # still be watchable. + channel = _slack_channel(enabled=True, allow_conversations=["*"]) + + await channel._handle_message_event( + _event(channel="G0123ABCD", channel_type="group"), + ) + + channel.router.observe.assert_awaited_once() + + async def test_an_ambiguous_g_conversation_is_not_observed(self): + # No channel_type to disambiguate: decline rather than guess. + channel = _slack_channel(enabled=True, allow_conversations=["*"]) + event = _event(channel="G0123ABCD") + event.pop("channel_type") + + await channel._handle_message_event(event) + + channel.router.observe.assert_not_awaited() + + async def test_join_and_leave_noise_is_not_buffered(self): + channel = _slack_channel(enabled=True, allow_conversations=["*"]) + + await channel._handle_message_event(_event(subtype="channel_join")) + + channel.router.observe.assert_not_awaited() + + async def test_another_app_is_not_buffered(self): + channel = _slack_channel(enabled=True, allow_conversations=["*"]) + channel._web.users_info = AsyncMock( + return_value={"user": {"is_bot": True, "profile": {}}}, + ) + + await channel._handle_message_event(_event(bot_id="B0OTHER")) + + channel.router.observe.assert_not_awaited() + + async def test_an_id_only_policy_costs_no_api_call(self): + # Observation sits on the dispatch path of a busy channel. A lookup + # per message would make watching one expensive. + channel = _slack_channel(enabled=True, allow_conversations=["C0123ABCD"]) + + await channel._handle_message_event(_event()) + + channel._web.conversations_info.assert_not_awaited() + channel._web.users_info.assert_not_awaited() + + async def test_a_lowercase_id_matches_at_the_cost_of_a_lookup(self): + # Shaped like both an id and a name, so it is resolved rather than + # assumed. It still matches; it just costs the lookup. + channel = _slack_channel( + enabled=True, + allow_conversations=["c0123abcd"], + deny_senders=["u0999zzzz"], + ) + + await channel._handle_message_event(_event()) + + channel.router.observe.assert_awaited_once() + channel._web.conversations_info.assert_awaited() + channel._web.users_info.assert_awaited() + + async def test_a_name_shaped_like_an_id_can_still_deny(self): + # Read as an id, `buildstatus` is never matched against the resolved + # name and the deny rule silently stops working. + channel = _slack_channel( + enabled=True, + allow_conversations=["C0123ABCD"], + deny_conversations=["buildstatus"], + ) + channel._web.conversations_info = AsyncMock( + return_value={"channel": {"name": "buildstatus"}}, + ) + + await channel._handle_message_event(_event()) + + channel.router.observe.assert_not_awaited() + + async def test_the_thread_is_recorded_for_a_reader_to_expand(self): + channel = _slack_channel(enabled=True, allow_conversations=["C0123ABCD"]) + channel.router.get_last_session = AsyncMock(return_value=None) + + await channel._handle_message_event( + _event(thread_ts="1699999999.000000"), + ) + + observed = channel.router.observe.await_args.args[0] + assert observed.metadata["thread_ts"] == "1699999999.000000" + + async def test_an_unwatched_channel_costs_no_sender_lookup(self): + # The conversation settles this without a lookup. Resolving the + # sender first would cost a users.info call for every message in + # every channel the bot sits in. + channel = _slack_channel( + enabled=True, + allow_conversations=["C0111AAAA"], + deny_senders=["spam-bot"], + ) + + await channel._handle_message_event(_event(channel="C0999BBBB")) + + channel._web.users_info.assert_not_awaited() + channel.router.observe.assert_not_awaited() + + async def test_a_watched_channel_still_resolves_the_sender(self): + channel = _slack_channel( + enabled=True, + allow_conversations=["C0123ABCD"], + deny_senders=["spam-bot"], + ) + + await channel._handle_message_event(_event()) + + channel._web.users_info.assert_awaited_once() + channel.router.observe.assert_awaited_once() + + async def test_an_upload_is_recorded_by_name_not_as_a_blank(self): + channel = _slack_channel(enabled=True, allow_conversations=["C0123ABCD"]) + + await channel._handle_message_event( + _event(text="", files=[{"name": "outage.pdf"}]), + ) + + observed = channel.router.observe.await_args.args[0] + assert observed.text == "[File: outage.pdf]" + + async def test_an_upload_with_a_comment_keeps_both(self): + channel = _slack_channel(enabled=True, allow_conversations=["C0123ABCD"]) + + await channel._handle_message_event( + _event(text="see attached", files=[{"name": "outage.pdf"}]), + ) + + observed = channel.router.observe.await_args.args[0] + assert observed.text == "[File: outage.pdf]\n\nsee attached" + + async def test_a_message_carrying_nothing_at_all_is_not_buffered(self): + # No text and no file name is nothing worth a row. + channel = _slack_channel(enabled=True, allow_conversations=["C0123ABCD"]) + + await channel._handle_message_event(_event(text="")) + + channel.router.observe.assert_not_awaited() + + +# ---------------------------------------------------------------------- # +# Telegram hook # +# ---------------------------------------------------------------------- # + + +def _tg_channel(**source_kwargs): + """A Telegram channel with a stub router and a source policy.""" + from nerve.channels.telegram import TelegramChannel + + cfg = NerveConfig() + cfg.telegram.source = ChannelSourceConfig(**source_kwargs) + cfg.telegram.allowed_users = [999] + channel = TelegramChannel.__new__(TelegramChannel) + channel._config = lambda: cfg + channel._allowed_users = set(cfg.telegram.allowed_users) + channel._last_update_time = 0.0 + channel.router = MagicMock() + channel.router.observe = AsyncMock(return_value=True) + # Every routing test drives _handle_message, whose live branch is long. + # A media group short-circuits it at the first step, so these assert on + # which route was taken without stubbing the whole extraction pipeline. + channel._collect_media_group = AsyncMock() + return channel + + +def _tg_update( + chat_id=-100123, chat_type="supergroup", title="ops-room", + user_id=42, username="mallory", is_bot=False, + sticker=None, document=None, photo=None, +): + """A plain text group message, unless an attachment is asked for. + + The attachment fields have to be set even when absent. A real + ``telegram.Message`` reports ``None`` for the ones it does not carry, + where a MagicMock hands back a truthy stand-in for every attribute. This + double would then look like a message carrying all three at once. + """ + update = MagicMock() + update.effective_chat.id = chat_id + update.effective_chat.type = chat_type + update.effective_chat.title = title + update.effective_chat.username = None + update.effective_user.id = user_id + update.effective_user.username = username + update.effective_user.first_name = "M" + update.effective_user.last_name = None + update.effective_user.full_name = "M" + update.effective_user.is_bot = is_bot + update.message.text = "overheard" + update.message.caption = None + update.message.message_id = 7 + update.message.date = None + update.message.reply_to_message = None + update.message.media_group_id = "album" + update.message.sticker = sticker + update.message.document = document + update.message.photo = photo + return update + + +def _tg_command_application(channel, callback_name="_handle_new_session"): + """Build the real PTB dispatch table with command callbacks isolated.""" + from telegram import User + + channel.config.telegram.bot_token = "123:ABC" + command_callback = AsyncMock() + generic_callback = AsyncMock(wraps=channel._handle_message) + setattr(channel, callback_name, command_callback) + channel._handle_message = generic_callback + app = channel._build_application() + # process_update only needs the initialized flag for these blocking, + # persistence-free handlers. Setting the bot user avoids a getMe request. + app._initialized = True + app.bot._bot_user = User( + id=1000, first_name="Nerve", is_bot=True, username="nerve_bot", + ) + return app, command_callback, generic_callback + + +def _tg_command_update(app, text="/new", user_id=42, *, missing_user=False): + """A real Telegram command update, so PTB chooses the production handler.""" + from telegram import Update + + command = text.split(maxsplit=1)[0] + message = { + "message_id": 7, + "date": 1_700_000_000, + "chat": {"id": -100123, "type": "supergroup", "title": "Room"}, + "text": text, + "entities": [{ + "type": "bot_command", "offset": 0, "length": len(command), + }], + } + if not missing_user: + message["from"] = { + "id": user_id, "is_bot": False, "first_name": "Alice", + } + return Update.de_json({"update_id": 1, "message": message}, app.bot) + + +def _tg_caption_update(app, caption="/pair 654321", user_id=42): + """A real Telegram photo caption, which PTB sends to MessageHandler.""" + from telegram import Update + + command = caption.split(maxsplit=1)[0] + return Update.de_json({ + "update_id": 1, + "message": { + "message_id": 7, + "date": 1_700_000_000, + "chat": {"id": -100123, "type": "supergroup", "title": "Room"}, + "from": {"id": user_id, "is_bot": False, "first_name": "Alice"}, + "photo": [{ + "file_id": "photo-id", + "file_unique_id": "photo-unique-id", + "width": 1, + "height": 1, + "file_size": 1, + }], + "caption": caption, + "caption_entities": [{ + "type": "bot_command", "offset": 0, "length": len(command), + }], + }, + }, app.bot) + + +class TestTelegramRouting: + """Same independent routing as Slack, decided by authorization. + + Telegram has no "addressed to me" test, so authorization alone settles the + live route for each message. + The source grant is asked either way. + """ + + async def test_source_only_for_a_sender_who_may_not_drive_the_agent(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + + await channel._handle_message(_tg_update(user_id=42), None) + + channel._collect_media_group.assert_not_awaited() + channel.router.observe.assert_awaited_once() + + async def test_channel_only_when_the_source_does_not_want_the_chat(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100999"]) + + await channel._handle_message(_tg_update(user_id=999), None) + + channel._collect_media_group.assert_awaited_once() + channel.router.observe.assert_not_awaited() + + async def test_a_handled_message_is_not_also_collected_by_default(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + + await channel._handle_message(_tg_update(user_id=999), None) + + channel._collect_media_group.assert_awaited_once() + channel.router.observe.assert_not_awaited() + + async def test_include_handled_messages_sends_it_to_both(self): + channel = _tg_channel( + enabled=True, + allow_conversations=["-100123"], + include_handled_messages=True, + ) + + await channel._handle_message(_tg_update(user_id=999), None) + + channel._collect_media_group.assert_awaited_once() + channel.router.observe.assert_awaited_once() + + async def test_neither_route_takes_a_refused_sender_in_an_unwatched_chat(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100999"]) + + await channel._handle_message(_tg_update(user_id=42), None) + + channel._collect_media_group.assert_not_awaited() + channel.router.observe.assert_not_awaited() + + async def test_a_message_without_an_effective_user_is_ignored(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + update = _tg_update() + update.effective_user = None + + await channel._handle_message(update, None) + + channel._collect_media_group.assert_not_awaited() + channel.router.observe.assert_not_awaited() + + +class TestTelegramCommandRouting: + async def test_an_unauthorized_command_reaches_the_source(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + app, command_callback, generic_callback = _tg_command_application(channel) + + await app.process_update(_tg_command_update(app, user_id=42)) + + command_callback.assert_awaited_once() + generic_callback.assert_not_awaited() + channel.router.observe.assert_awaited_once() + + async def test_a_handled_command_stays_on_the_live_route_by_default(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + app, command_callback, generic_callback = _tg_command_application(channel) + + await app.process_update(_tg_command_update(app, user_id=999)) + + command_callback.assert_awaited_once() + generic_callback.assert_not_awaited() + channel.router.observe.assert_not_awaited() + + async def test_a_handled_command_can_reach_both_routes(self): + channel = _tg_channel( + enabled=True, + allow_conversations=["-100123"], + include_handled_messages=True, + ) + app, command_callback, generic_callback = _tg_command_application(channel) + + await app.process_update(_tg_command_update(app, user_id=999)) + + command_callback.assert_awaited_once() + generic_callback.assert_not_awaited() + channel.router.observe.assert_awaited_once() + + async def test_a_pairing_code_is_never_collected(self): + channel = _tg_channel( + enabled=True, + allow_conversations=["-100123"], + include_handled_messages=True, + ) + app, command_callback, generic_callback = _tg_command_application( + channel, callback_name="_handle_pair", + ) + + await app.process_update(_tg_command_update( + app, text="/pair 654321", user_id=42, + )) + + command_callback.assert_awaited_once() + generic_callback.assert_not_awaited() + channel.router.observe.assert_not_awaited() + + async def test_a_pairing_code_addressed_to_another_bot_is_not_collected(self): + channel = _tg_channel( + enabled=True, + allow_conversations=["-100123"], + include_handled_messages=True, + ) + app, command_callback, generic_callback = _tg_command_application( + channel, callback_name="_handle_pair", + ) + + await app.process_update(_tg_command_update( + app, text="/pair@other_bot 654321", user_id=42, + )) + + command_callback.assert_not_awaited() + generic_callback.assert_awaited_once() + channel.router.observe.assert_not_awaited() + + async def test_a_pairing_code_in_a_media_caption_is_not_collected(self): + channel = _tg_channel( + enabled=True, + allow_conversations=["-100123"], + include_handled_messages=True, + ) + app, command_callback, generic_callback = _tg_command_application( + channel, callback_name="_handle_pair", + ) + + await app.process_update(_tg_caption_update(app, user_id=42)) + + command_callback.assert_not_awaited() + generic_callback.assert_awaited_once() + channel.router.observe.assert_not_awaited() + + async def test_a_command_without_an_effective_user_is_ignored(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + app, command_callback, generic_callback = _tg_command_application(channel) + + await app.process_update(_tg_command_update(app, missing_user=True)) + + command_callback.assert_not_awaited() + generic_callback.assert_not_awaited() + channel.router.observe.assert_not_awaited() + + +class TestTelegramObserve: + async def test_pairing_is_refused_outside_a_private_human_chat( + self, monkeypatch, + ): + channel = _tg_channel(enabled=True, allow_conversations=["*"]) + update = _tg_update() + update.message.sender_chat = None + update.message.reply_text = AsyncMock() + verify = MagicMock(return_value=True) + monkeypatch.setattr("nerve.pairing.verify_pairing_code", verify) + context = MagicMock(args=["654321"]) + + await channel._handle_pair(update, context) + + verify.assert_not_called() + assert 42 not in channel._allowed_users + update.message.reply_text.assert_not_awaited() + + async def test_a_granted_group_is_observed(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + + await channel._observe(_tg_update()) + + channel.router.observe.assert_awaited_once() + observed = channel.router.observe.await_args.args[0] + assert observed.channel_name == "telegram" + assert observed.conversation_id == "-100123" + assert observed.text == "overheard" + + async def test_a_spoofed_title_does_not_grant_access(self): + # A group's title is set by whoever runs it. If a title could satisfy + # an allow rule, anyone could create a group named "ops-room" and + # walk into a grant meant for a different one. + channel = _tg_channel(enabled=True, allow_conversations=["ops-room"]) + + await channel._observe(_tg_update(chat_id=-100999, title="ops-room")) + + channel.router.observe.assert_not_awaited() + + async def test_a_title_can_still_deny(self): + # Spoofable names may only ever subtract access. + channel = _tg_channel( + enabled=True, + allow_conversations=["*"], + deny_conversations=["ops-room"], + ) + + await channel._observe(_tg_update(title="ops-room")) + + channel.router.observe.assert_not_awaited() + + async def test_a_spoofed_username_does_not_grant_a_sender(self): + channel = _tg_channel( + enabled=True, + allow_conversations=["*"], + allow_senders=["mallory"], + ) + + await channel._observe(_tg_update(username="mallory")) + + channel.router.observe.assert_not_awaited() + + async def test_a_full_profile_name_can_deny_a_sender(self): + channel = _tg_channel( + enabled=True, + allow_conversations=["*"], + deny_senders=["Alice Smith"], + ) + update = _tg_update() + update.effective_user.first_name = "Alice" + update.effective_user.last_name = "Smith" + update.effective_user.full_name = "Alice Smith" + + await channel._observe(update) + + channel.router.observe.assert_not_awaited() + + async def test_a_private_chat_is_never_observed(self): + channel = _tg_channel(enabled=True, allow_conversations=["*"]) + + await channel._observe(_tg_update(chat_type="private")) + + channel.router.observe.assert_not_awaited() + + async def test_another_bot_is_not_observed(self): + # Two agents filling each other's inboxes is no better than two + # agents answering each other. + channel = _tg_channel(enabled=True, allow_conversations=["*"]) + + await channel._observe(_tg_update(is_bot=True)) + + channel.router.observe.assert_not_awaited() + + async def test_the_source_off_collects_nothing(self): + channel = _tg_channel(enabled=False, allow_conversations=["*"]) + + await channel._observe(_tg_update()) + + channel.router.observe.assert_not_awaited() + + async def test_a_sticker_is_recorded_by_its_emoji(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + update = _tg_update(sticker=MagicMock(emoji="🔥")) + update.message.text = None + + await channel._observe(update) + + observed = channel.router.observe.await_args.args[0] + assert observed.text == "[Sticker: 🔥]" + + async def test_an_uncaptioned_photo_is_recorded_as_one(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + update = _tg_update(photo=[MagicMock()]) + update.message.text = None + + await channel._observe(update) + + observed = channel.router.observe.await_args.args[0] + assert observed.text == "[Photo]" + + async def test_a_document_keeps_its_caption(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + update = _tg_update(document=MagicMock(file_name="runbook.md")) + update.message.text = None + update.message.caption = "the one we wrote in March" + + await channel._observe(update) + + observed = channel.router.observe.await_args.args[0] + assert observed.text == "[File: runbook.md]\n\nthe one we wrote in March" + + async def test_a_message_carrying_nothing_at_all_is_not_buffered(self): + channel = _tg_channel(enabled=True, allow_conversations=["-100123"]) + update = _tg_update() + update.message.text = None + + await channel._observe(update) + + channel.router.observe.assert_not_awaited() + + +# ---------------------------------------------------------------------- # +# Router seam # +# ---------------------------------------------------------------------- # + + +def _observed(**kwargs) -> ObservedMessage: + base = dict( + channel_name="slack", + channel_key="slack:C0123ABCD", + conversation_id="C0123ABCD", + sender_id="U0456DEFG", + text="hello", + message_id="1700000000.000100", + timestamp="2023-11-14T22:13:20+00:00", + ) + base.update(kwargs) + return ObservedMessage(**base) + + +class TestRouterObserve: + async def test_an_observation_reaches_the_database(self, db): + engine = MagicMock() + engine.db = db + router = ChannelRouter(engine) + + assert await router.observe(_observed()) + + rows = await db.read_channel_observations("slack") + assert len(rows) == 1 + assert rows[0][1]["text"] == "hello" + + async def test_a_database_failure_does_not_escape(self): + # This runs on the dispatch path of a channel that already decided + # not to answer. A DB hiccup must not take down message handling for + # traffic the agent was never going to act on. + engine = MagicMock() + engine.db.insert_channel_observation = AsyncMock( + side_effect=RuntimeError("boom"), + ) + router = ChannelRouter(engine) + + assert not await router.observe(_observed()) + + +# ---------------------------------------------------------------------- # +# Buffer # +# ---------------------------------------------------------------------- # + + +class TestBuffer: + async def test_rows_come_back_in_order_past_a_cursor(self, db): + for i in range(5): + await db.insert_channel_observation( + "slack", "slack:C1", {"n": i}, + ) + + rows = await db.read_channel_observations("slack", after_id=0, limit=3) + + assert [p["n"] for _, p in rows] == [0, 1, 2] + rest = await db.read_channel_observations("slack", after_id=rows[-1][0]) + assert [p["n"] for _, p in rest] == [3, 4] + + async def test_channels_do_not_see_each_other(self, db): + await db.insert_channel_observation("slack", "slack:C1", {"n": 1}) + await db.insert_channel_observation("telegram", "telegram:9", {"n": 2}) + + rows = await db.read_channel_observations("slack") + + assert [p["n"] for _, p in rows] == [1] + + async def test_ids_keep_climbing_after_a_prune(self, db): + # The reason the migration uses AUTOINCREMENT. A plain rowid is + # reused once the highest row is deleted, so a drained-and-pruned + # buffer would reissue ids the cursor has already passed and the next + # observations would be skipped for good. + first = await db.insert_channel_observation("slack", "k", {"n": 1}) + await db._write("DELETE FROM channel_observations", ()) + + second = await db.insert_channel_observation("slack", "k", {"n": 2}) + + assert second > first + + async def test_the_row_cap_drops_the_oldest(self, db): + # Trimming is amortized, so the cap is a bound the buffer returns to + # rather than one it never crosses: it may overshoot by up to one + # trim interval. That beats a COUNT on the dispatch path. + cap = 5 + written = _TRIM_EVERY * 2 + cap + for i in range(written): + await db.insert_channel_observation( + "slack", "k", {"n": i}, max_rows=cap, + ) + + rows = await db.read_channel_observations("slack", limit=10_000) + + assert len(rows) <= cap + _TRIM_EVERY + assert len(rows) < written + # What survives is the newest end — a stale backlog nobody drained + # is the right thing to lose. + assert rows[-1][1]["n"] == written - 1 + + async def test_an_expired_observation_is_swept(self, db): + await db.insert_channel_observation("slack", "k", {"n": 1}, ttl_days=-1) + await db.insert_channel_observation("slack", "k", {"n": 2}, ttl_days=7) + + deleted = await db.cleanup_expired_channel_observations() + + assert deleted == 1 + rows = await db.read_channel_observations("slack") + assert [p["n"] for _, p in rows] == [2] + + async def test_an_unreadable_payload_is_reported_not_hidden(self, db): + # It comes back with a None payload rather than being dropped, so + # the drain can skip the row and still advance past its id. + await db.insert_channel_observation("slack", "k", {"n": 1}) + await _write_garbage(db) + await db.insert_channel_observation("slack", "k", {"n": 3}) + + rows = await db.read_channel_observations("slack") + + assert [p["n"] if p else None for _, p in rows] == [1, None, 3] + + +# ---------------------------------------------------------------------- # +# ChannelSource # +# ---------------------------------------------------------------------- # + + +async def _write_garbage(db) -> None: + """Put a row in the buffer whose payload will never parse.""" + await db._write( + "INSERT INTO channel_observations " + "(channel, channel_key, payload, created_at, expires_at) " + "VALUES ('slack', 'k', 'not json', '2026-01-01', '2099-01-01')", + (), + ) + + +class TestChannelSource: + async def test_buffered_rows_become_records(self, db): + await db.insert_channel_observation( + "slack", "slack:C0123ABCD", + { + "conversation_id": "C0123ABCD", + "conversation_title": "general", + "sender_id": "U0456DEFG", + "sender_name": "alice", + "text": "ship it", + "message_id": "1700000000.000100", + "timestamp": "2023-11-14T22:13:20+00:00", + "channel_key": "slack:C0123ABCD", + "metadata": {"thread_ts": "1699999999.000000"}, + }, + ) + + result = await ChannelSource("slack", db).fetch(None) + + assert len(result.records) == 1 + record = result.records[0] + assert record.id == "C0123ABCD:1700000000.000100" + assert record.source == "slack:observed" + assert record.record_type == "slack_message" + assert record.summary == "[general] alice: ship it" + assert record.content == "ship it" + assert record.metadata["thread_ts"] == "1699999999.000000" + assert record.metadata["conversation_id"] == "C0123ABCD" + + async def test_the_cursor_is_the_buffer_id(self, db): + last = 0 + for i in range(3): + last = await db.insert_channel_observation( + "slack", "k", {"text": str(i), "message_id": str(i)}, + ) + + result = await ChannelSource("slack", db).fetch(None) + + assert result.next_cursor == str(last) + + async def test_a_cursor_resumes_where_it_left_off(self, db): + source = ChannelSource("slack", db) + for i in range(3): + await db.insert_channel_observation( + "slack", "k", {"text": str(i), "message_id": str(i)}, + ) + first = await source.fetch(None, limit=2) + + second = await source.fetch(first.next_cursor) + + assert [r.content for r in first.records] == ["0", "1"] + assert [r.content for r in second.records] == ["2"] + + async def test_a_full_batch_reports_more(self, db): + for i in range(4): + await db.insert_channel_observation( + "slack", "k", {"text": str(i), "message_id": str(i)}, + ) + + result = await ChannelSource("slack", db).fetch(None, limit=2) + + assert result.has_more + + async def test_an_empty_buffer_holds_the_cursor(self, db): + result = await ChannelSource("slack", db).fetch("42") + + assert result.records == [] + assert result.next_cursor == "42" + assert not result.has_more + + async def test_an_unreadable_cursor_starts_over_rather_than_failing(self, db): + # source_messages is keyed (source, id), so re-reading a bounded, + # TTL-capped buffer re-inserts nothing. Failing closed here would + # instead wedge the source until someone edited the database. + await db.insert_channel_observation( + "slack", "k", {"text": "hi", "message_id": "1"}, + ) + + result = await ChannelSource("slack", db).fetch("not-a-number") + + assert len(result.records) == 1 + + async def test_an_unreadable_row_is_skipped_and_passed(self, db): + # The wedge this guards against: if the cursor only ever advanced to + # the last row that *parsed*, a batch of entirely unreadable rows + # would move it nowhere, be re-read every run, and hide everything + # behind them for good. + source = ChannelSource("slack", db) + await _write_garbage(db) + await _write_garbage(db) + + first = await source.fetch(None) + + assert first.records == [] + assert first.next_cursor == "2" + + await db.insert_channel_observation( + "slack", "k", {"text": "reachable", "message_id": "3"}, + ) + second = await source.fetch(first.next_cursor) + assert [r.content for r in second.records] == ["reachable"] + + async def test_valid_json_that_is_not_an_object_is_skipped(self, db): + # `[]` parses fine and then has no .get(). Letting that raise out of + # fetch would leave the cursor behind the row forever. + await db._write( + "INSERT INTO channel_observations " + "(channel, channel_key, payload, created_at, expires_at) " + "VALUES ('slack', 'k', '[]', '2026-01-01', '2099-01-01')", + (), + ) + await db.insert_channel_observation( + "slack", "k", {"text": "reachable", "message_id": "2"}, + ) + + result = await ChannelSource("slack", db).fetch(None) + + assert [r.content for r in result.records] == ["reachable"] + assert result.next_cursor == "2" + + async def test_the_same_message_observed_twice_lands_once(self, db): + payload = { + "conversation_id": "C1", + "message_id": "1700000000.000100", + "text": "hi", + "timestamp": "2023-11-14T22:13:20+00:00", + } + await db.insert_channel_observation("slack", "k", payload) + await db.insert_channel_observation("slack", "k", dict(payload)) + + result = await ChannelSource("slack", db).fetch(None) + inserted = await db.insert_source_messages(result.records, source="slack") + + assert len(result.records) == 2 + assert inserted == 1 + + +# ---------------------------------------------------------------------- # +# Draining a backlog # +# ---------------------------------------------------------------------- # + + +async def _fill(db, count, channel="slack"): + for i in range(count): + await db.insert_channel_observation( + channel, "slack:C1", + { + "conversation_id": "C1", + "message_id": str(i), + "text": f"message {i}", + }, + ) + + +class TestDrainKeepsUp: + """A backlog deeper than one batch has to drain, not accumulate. + + A watched channel busier than ``batch_size`` per tick would otherwise + fall permanently behind, and the buffer's own row cap then trims from the + old end, dropping the messages the drain had not reached yet. + """ + + async def test_a_backlog_past_one_batch_drains_in_one_run(self, db): + await _fill(db, 120) + runner = SourceRunner( + source=ChannelSource("slack", db), db=db, batch_size=50, + ) + + result = await runner.run() + + assert result.records_ingested == 120 + assert await db.get_sync_cursor("slack:observed") == "120" + + async def test_the_run_stops_at_the_pass_bound_and_keeps_its_place(self, db): + # A bound, not a promise to empty the buffer. The next run resumes + # exactly where this one stopped. + await _fill(db, _MAX_PASSES * 2 + 5) + runner = SourceRunner( + source=ChannelSource("slack", db), db=db, batch_size=2, + ) + + first = await runner.run() + second = await runner.run() + + assert first.records_ingested == _MAX_PASSES * 2 + assert second.records_ingested == 5 + + async def test_a_source_that_never_advances_does_not_spin(self, db): + # has_more with a standing cursor re-reads the same batch, for nothing. + stuck = MagicMock() + stuck.source_name = "stuck:observed" + stuck.fetch = AsyncMock(return_value=FetchResult( + records=[SourceRecord( + id="1", source="stuck:observed", record_type="x", + summary="s", content="c", timestamp="", + )], + next_cursor=None, + has_more=True, + )) + stuck.preprocess = AsyncMock(side_effect=lambda r: r) + runner = SourceRunner(source=stuck, db=db, batch_size=1) + + await runner.run() + + assert stuck.fetch.await_count == 1 + + async def test_a_failure_mid_backlog_keeps_what_landed(self, db): + await _fill(db, 60) + source = ChannelSource("slack", db) + real_fetch = source.fetch + calls = [] + + async def fetch(cursor, limit=100): + calls.append(cursor) + if len(calls) > 1: + raise RuntimeError("buffer went away") + return await real_fetch(cursor, limit=limit) + + source.fetch = fetch + runner = SourceRunner(source=source, db=db, batch_size=50) + + result = await runner.run() + + assert result.error == "buffer went away" + assert result.records_ingested == 50 + # The first batch's cursor is durable, so the retry resumes past it. + assert await db.get_sync_cursor("slack:observed") == "50" + + +# ---------------------------------------------------------------------- # +# Registry wiring # +# ---------------------------------------------------------------------- # + + +class TestRegistry: + def test_an_observing_channel_gets_a_runner(self, tmp_path): + cfg = NerveConfig() + cfg.slack.source = ChannelSourceConfig( + enabled=True, allow_conversations=["C0123ABCD"], schedule="*/7 * * * *", + ) + + runners = build_source_runners(cfg, MagicMock()) + + slack = [r for r in runners if r.source.source_name == "slack:observed"] + assert len(slack) == 1 + # The runner carries its own cadence: the config lives at + # slack.source, and CronService would otherwise find no + # config.sync.slack section and never schedule it. + assert slack[0].schedule == "*/7 * * * *" + + def test_the_carried_schedule_is_what_cron_uses(self): + # Without this the runner is built and then silently never + # scheduled: there is no config.sync.slack section to look up, so + # the lookup returns None and _plan_source_runners drops it. + from nerve.cron.service import CronService + + service = CronService.__new__(CronService) + service.config = NerveConfig() + runner = MagicMock() + runner.source.source_name = "slack:observed" + runner.schedule = "*/7 * * * *" + + assert service._source_schedule(runner) == "*/7 * * * *" + + def test_a_non_string_schedule_is_ignored(self): + # _source_schedule reads a duck-typed attribute, and a bare + # MagicMock answers truthily to anything asked of it — which would + # otherwise put a mock where a crontab expression belongs. + from nerve.cron.service import CronService + + service = CronService.__new__(CronService) + service.config = NerveConfig() + runner = MagicMock() + runner.source.source_name = "github" + + assert service._source_schedule(runner) == NerveConfig().sync.github.schedule + + def test_no_runner_without_a_conversation_grant(self, tmp_path): + cfg = NerveConfig() + cfg.slack.source = ChannelSourceConfig(enabled=True, allow_conversations=[]) + + runners = build_source_runners(cfg, MagicMock()) + + assert not [r for r in runners if r.source.source_name == "slack:observed"] + + def test_no_runner_when_observation_is_off(self, tmp_path): + cfg = NerveConfig() + + runners = build_source_runners(cfg, MagicMock()) + + assert not [r for r in runners if r.source.source_name == "slack:observed"] + + def test_telegram_gets_the_same_treatment(self, tmp_path): + cfg = NerveConfig() + cfg.telegram.source = ChannelSourceConfig( + enabled=True, allow_conversations=["-100123"], + ) + + runners = build_source_runners(cfg, MagicMock()) + + assert [ + r for r in runners if r.source.source_name == "telegram:observed" + ] + + +class TestConfig: + def test_the_source_block_parses(self): + cfg = SlackConfig.from_dict({ + "source": { + "enabled": True, + "allow_channels": ["C1"], + "deny_senders": ["U0BOT"], + "schedule": "*/9 * * * *", + }, + }) + + assert cfg.source.enabled + assert cfg.source.allow_conversations == ["C1"] + assert cfg.source.deny_senders == ["U0BOT"] + assert cfg.source.schedule == "*/9 * * * *" + + def test_the_source_is_off_by_default(self): + cfg = SlackConfig.from_dict({}) + + assert not cfg.source.enabled + assert cfg.source.allow_conversations == [] + + def test_handled_messages_are_left_to_the_live_route_by_default(self): + assert not SlackConfig.from_dict( + {"source": {}}, + ).source.include_handled_messages + assert not TelegramConfig.from_dict( + {"source": {}}, + ).source.include_handled_messages + + @pytest.mark.parametrize("subject", ["slack", "telegram"]) + def test_include_handled_messages_parses_on_both_transports(self, subject): + parse = SlackConfig.from_dict if subject == "slack" else ( + TelegramConfig.from_dict + ) + cfg = parse({"source": {"include_handled_messages": True}}) + + assert cfg.source.include_handled_messages + + def test_a_mapping_allow_list_does_not_become_a_wildcard(self): + # {"*": false} reads like a disabled wildcard, and list() of it + # yields its keys. Guessing at intent would turn a config that looks + # switched off into one that grants everything. + cfg = SlackConfig.from_dict({ + "source": {"enabled": True, "allow_channels": {"*": False}}, + }) + + assert cfg.source.allow_conversations == [] + assert not cfg.source.active if hasattr( + cfg.source, "active", + ) else True + + @pytest.mark.parametrize("bad", [{"C1": True}, {"*": False}, None]) + def test_a_mapping_or_missing_allow_list_grants_nothing(self, bad): + cfg = SlackConfig.from_dict({ + "source": {"enabled": True, "allow_channels": bad}, + }) + + assert cfg.source.allow_conversations == [] + + def test_a_bare_string_is_one_pattern_not_many(self): + # This is how `allow_conversations: ${OBSERVE_ROOMS}` arrives after + # interpolation. Dropping it would break the documented env-var + # idiom; splitting it would invent patterns nobody wrote. + cfg = SlackConfig.from_dict({ + "source": {"allow_channels": "C0123ABCD"}, + }) + + assert cfg.source.allow_conversations == ["C0123ABCD"] + + def test_scalar_entries_are_kept_and_junk_dropped(self): + cfg = SlackConfig.from_dict({ + "source": {"allow_channels": [{"a": 1}, "C1", " ", "C2"]}, + }) + + assert cfg.source.allow_conversations == ["C1", "C2"] + + @pytest.mark.parametrize("bad", [5, "", " ", None]) + def test_an_unusable_schedule_falls_back(self, bad): + # There is no sync.slack section to fall back to, so an ignored + # schedule would leave the buffer filling with nothing draining it. + cfg = SlackConfig.from_dict({"source": {"schedule": bad}}) + + assert cfg.source.schedule == "*/5 * * * *" + + @pytest.mark.parametrize("bad", [0, -1, 5]) + def test_a_row_target_that_would_empty_the_buffer_is_refused(self, bad): + # Zero reads like "no limit" and means the opposite: the trim keeps + # no rows. SQLite reads a negative OFFSET as zero, so it does the same. + cfg = SlackConfig.from_dict({"source": {"max_stored_messages": bad}}) + + assert cfg.source.max_stored_messages == 100 + + def test_a_usable_row_target_is_kept(self): + cfg = SlackConfig.from_dict({"source": {"max_stored_messages": 250}}) + + assert cfg.source.max_stored_messages == 250 + + def test_each_transport_uses_its_own_noun(self): + # On Telegram a "channel" is a specific entity type distinct from a + # group, so one shared noun could not name both correctly. + slack = SlackConfig.from_dict({"source": {"allow_channels": ["C1"]}}) + tg = TelegramConfig.from_dict({"source": {"allow_chats": ["-100"]}}) + + assert slack.source.allow_conversations == ["C1"] + assert tg.source.allow_conversations == ["-100"] + + def test_the_other_transports_key_grants_nothing(self): + # Fail closed on a key this transport does not read, rather than + # quietly accepting it and collecting nothing anyway. + cfg = SlackConfig.from_dict({"source": {"allow_chats": ["C1"]}}) + + assert cfg.source.allow_conversations == [] + + @pytest.mark.parametrize( + "written,matched", + [ + ("#eng-backend", "eng-backend"), + ("@alice", "alice"), + ("eng-backend", "eng-backend"), + ("*@example.com", "*@example.com"), + ], + ) + def test_a_pasted_display_name_still_matches(self, written, matched): + # A chat client renders "#eng-backend"; the API name has no sigil and + # matching is literal, so left alone the paste would match nothing + # and an allow list would leave the inbox silently empty. + cfg = SlackConfig.from_dict({"source": {"allow_channels": [written]}}) + + assert cfg.source.allow_conversations == [matched] + + def test_enabling_without_a_grant_warns(self, caplog): + with caplog.at_level("WARNING"): + SlackConfig.from_dict({"source": {"enabled": True}}) + + assert "allow_channels" in caplog.text diff --git a/tests/test_slack_channel.py b/tests/test_slack_channel.py index a0c8d524..555f8ae4 100644 --- a/tests/test_slack_channel.py +++ b/tests/test_slack_channel.py @@ -1063,12 +1063,26 @@ def test_real_slack_ids_are_recognised(self): assert is_slack_id("C0456DEF") assert is_slack_id("W01ABCDEFGH") + def test_a_lowercase_id_is_not_treated_as_an_id(self): + # Not a typo. `c0456def` is a legal channel name, so a lowercase + # pattern is ambiguous and only a lookup settles it. Matching stays + # case-insensitive, so a lowercase id still matches. + assert not is_slack_id("u0123abc") + assert not is_slack_id("c0456def") + def test_an_uppercase_name_is_not_an_id(self): # This is the bug: a case heuristic read ALICE as an id, skipped the # users.info lookup, and let deny_users=["ALICE"] admit her. assert not is_slack_id("ALICE") assert not is_slack_id("ENGINEERING") + def test_a_lowercase_name_shaped_like_an_id_is_not_an_id(self): + # The same bug in the direction the eye misses. Ordinary names, but + # read as ids they skip the lookup and deny_users admits them. + assert not is_slack_id("beckyjones") + assert not is_slack_id("buildstatus") + assert not is_slack_id("dataplatform") + def test_a_handle_or_email_is_not_an_id(self): assert not is_slack_id("alex.soffronow") assert not is_slack_id("a@b.com") @@ -1094,6 +1108,38 @@ async def test_an_uppercase_deny_name_still_forces_a_lookup(self): assert not await channel._authorize("U999", "C0456DEF", "channel") channel._web.users_info.assert_awaited() + @pytest.mark.asyncio + async def test_a_lowercase_deny_name_still_forces_a_lookup(self): + # Read as an id, this handle is never matched against the resolved + # name and the deny rule admits her. + channel = _channel(deny_users=["beckyjones"], allow_channels=["C0456DEF"]) + channel._web.users_info = AsyncMock( + return_value={ + "user": { + "id": "U999", + "name": "beckyjones", + "profile": {"email": "becky@x.com"}, + }, + } + ) + channel._web.conversations_info = AsyncMock( + return_value={"channel": {"id": "C0456DEF", "name": "eng"}} + ) + assert not await channel._authorize("U999", "C0456DEF", "channel") + channel._web.users_info.assert_awaited() + + @pytest.mark.asyncio + async def test_a_lowercase_id_grants_without_needing_its_name(self): + # The other half: a pasted lowercase id still grants, via the lookup. + channel = _channel(allow_users=["u0123abc"], allow_channels=["c0456def"]) + channel._web.users_info = AsyncMock( + return_value={"user": {"id": "U0123ABC", "name": "alice", "profile": {}}} + ) + channel._web.conversations_info = AsyncMock( + return_value={"channel": {"id": "C0456DEF", "name": "eng"}} + ) + assert await channel._authorize("U0123ABC", "C0456DEF", "channel") + @pytest.mark.asyncio async def test_a_missing_email_refuses_an_email_deny_rule(self): # users.info answers 200 without profile.email when the token lacks diff --git a/tests/test_slack_live_inbound.py b/tests/test_slack_live_inbound.py index 5169e40a..120f9e5b 100644 --- a/tests/test_slack_live_inbound.py +++ b/tests/test_slack_live_inbound.py @@ -39,7 +39,14 @@ import pytest import pytest_asyncio -from nerve.channels.slack import format_target +from types import SimpleNamespace + +from nerve.channels.router import ChannelRouter +from nerve.channels.slack import format_target, slack_ts_to_iso +from nerve.config import ChannelSourceConfig +from nerve.db import Database +from nerve.sources.channel import ChannelSource +from nerve.sources.runner import SourceRunner from tests.slack_live import ( BOT_TOKEN, EVENT_TIMEOUT, @@ -47,6 +54,7 @@ USER_TOKEN, Posted, RecordingRouter, + REFUSAL_SETTLE_SECONDS, SOCKET_DRAIN_SECONDS, build_channel, direct_message_guardrails, @@ -167,6 +175,11 @@ async def _use(router: RecordingRouter, **slack_kwargs): ("commands", None), ): setattr(config.slack, field, slack_kwargs.get(field, default)) + # The source grant is reset like the rest: left standing it would + # collect the next test's traffic under this test's policy. + config.slack.source = slack_kwargs.get( + "source", ChannelSourceConfig(), + ) channel.apply_config(config) channel.router = router router.channel = channel @@ -441,3 +454,249 @@ async def test_the_watchdog_restores_a_dropped_socket(self): # Let Slack drop this connection before module fixture cleanup # relies on the shared socket receiving every deletion event. await asyncio.sleep(SOCKET_DRAIN_SECONDS) + + +# ---------------------------------------------------------------------- # +# Channel source — what a watched conversation feeds the inbox # +# ---------------------------------------------------------------------- # + + +@requires_inbound +class TestChannelSourceCollectsRealTraffic: + """The source gate, driven by messages a real person actually sent. + + The unit tests hand ``_observe`` an event dict the test wrote, so they + settle the policy branches and nothing about the events Slack sends. The + interesting claims here are that ordinary chatter carries the fields the + payload is built from, that ``channel_type`` on a real public-channel + message clears the DM guard, and that the live route and the source route + really are decided independently of one another. + """ + + async def test_ordinary_chatter_is_collected_but_not_answered( + self, live_channel, human, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter() + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + source=ChannelSourceConfig( + enabled=True, allow_conversations=[TEST_CHANNEL], + ), + ) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"team chatter {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + + observed = await router.wait_for_observation(marker) + assert observed.channel_name == "slack" + assert observed.conversation_id == TEST_CHANNEL + assert observed.sender_id == auth["user_id"] + assert observed.message_id == sent["ts"] + assert observed.channel_key == f"slack:{TEST_CHANNEL}:{sent['ts']}" + # The stamp Slack wrote, not the moment we read it. + assert observed.timestamp == slack_ts_to_iso(sent["ts"]) + # No mention, so the live route declined it. That is the whole point + # of the source: it reaches the inbox without starting a turn. + assert not router.messages + + async def test_a_conversation_off_the_grant_is_not_collected( + self, live_channel, human, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter() + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + source=ChannelSourceConfig( + enabled=True, allow_conversations=["C0NOTTHISONE"], + ), + ) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"unwatched chatter {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + + await router.expect_no_observation(marker, channel) + + async def test_a_channel_name_grant_resolves_the_real_name( + self, live_channel, human, bot, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + info = await bot.conversations_info(channel=TEST_CHANNEL) + name = info["channel"]["name"] + router = RecordingRouter() + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + source=ChannelSourceConfig( + enabled=True, allow_conversations=[name], + ), + ) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"named grant {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + + observed = await router.wait_for_observation(marker) + # Resolved for the grant, so it is recorded rather than left empty. + assert observed.conversation_title == name + + async def test_a_sender_denied_by_handle_is_not_collected( + self, live_channel, human, posted, + ): + # The deny pattern is a handle, so the gate has to resolve the real + # sender through users.info to find out it matches. + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter() + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + source=ChannelSourceConfig( + enabled=True, + allow_conversations=[TEST_CHANNEL], + deny_senders=[auth["user"]], + ), + ) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"denied sender {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + + await router.expect_no_observation(marker, channel) + + async def test_an_answered_mention_is_not_collected_by_default( + self, live_channel, human, bot, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter(reply_text="ack") + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + source=ChannelSourceConfig( + enabled=True, allow_conversations=[TEST_CHANNEL], + ), + ) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, + text=f"<@{channel._bot_user_id}> handled {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + + await router.wait_for_message(marker) + replies = await bot.conversations_replies( + channel=TEST_CHANNEL, ts=sent["ts"], + ) + for reply in replies["messages"]: + if reply.get("user") == channel._bot_user_id: + posted.note_bot(TEST_CHANNEL, reply["ts"]) + # Answered live, so the source leaves it alone: one message should + # not arrive twice. + assert not router._matching_observed(marker) + + async def test_include_handled_messages_sends_a_mention_to_both( + self, live_channel, human, bot, posted, + ): + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter(reply_text="ack") + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + source=ChannelSourceConfig( + enabled=True, + allow_conversations=[TEST_CHANNEL], + include_handled_messages=True, + ), + ) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, + text=f"<@{channel._bot_user_id}> both routes {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + + await router.wait_for_message(marker) + await router.wait_for_observation(marker) + replies = await bot.conversations_replies( + channel=TEST_CHANNEL, ts=sent["ts"], + ) + for reply in replies["messages"]: + if reply.get("user") == channel._bot_user_id: + posted.note_bot(TEST_CHANNEL, reply["ts"]) + + async def test_the_bots_own_post_is_not_collected( + self, live_channel, bot, posted, + ): + marker = unique_marker() + router = RecordingRouter() + channel, _ = await live_channel( + router, + source=ChannelSourceConfig( + enabled=True, allow_conversations=[TEST_CHANNEL], + ), + ) + + sent = await bot.chat_postMessage( + channel=TEST_CHANNEL, text=f"the agent talking {marker}", + ) + posted.note_bot(TEST_CHANNEL, sent["ts"]) + + # An agent reading its own output back is a loop, not a source. + await asyncio.sleep(REFUSAL_SETTLE_SECONDS) + assert not router._matching_observed(marker) + + async def test_a_collected_message_drains_into_the_inbox( + self, live_channel, human, posted, tmp_path, + ): + # The whole bridge, on one real message: Slack event → buffer row → + # ChannelSource → an inbox record a consumer tool would read. + marker = unique_marker() + auth = await human.auth_test() + router = RecordingRouter() + channel, _ = await live_channel( + router, + allow_users=[auth["user_id"]], + source=ChannelSourceConfig( + enabled=True, allow_conversations=[TEST_CHANNEL], + ), + ) + + sent = await human.chat_postMessage( + channel=TEST_CHANNEL, text=f"drain me {marker}", + ) + posted.note_user(TEST_CHANNEL, sent["ts"]) + observed = await router.wait_for_observation(marker) + + db = Database(tmp_path / "observations.db") + await db.connect() + try: + real_router = ChannelRouter(engine=SimpleNamespace(db=db)) + assert await real_router.observe(observed) + + result = await SourceRunner( + source=ChannelSource("slack", db), db=db, + ).run() + + assert result.records_ingested == 1 + rows, _ = await db.list_source_messages(source="slack:observed") + assert len(rows) == 1 + record_id = f"{TEST_CHANNEL}:{sent['ts']}" + assert rows[0]["id"] == record_id + stored = await db.get_source_message("slack:observed", record_id) + assert marker in stored["content"] + assert stored["metadata"]["sender_id"] == auth["user_id"] + finally: + await db.close()