diff --git a/config.example.yaml b/config.example.yaml index fb8f3e6b7..e29c748f9 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -371,6 +371,22 @@ sync: # close of the PRs it reported # deny_authors: [] # case-insensitive globs on the author login batch_size: 50 + # Linear issue events of the agent's Linear connector in the MCP catalog. + # The channel gateway polls Linear with the connector's credential and + # project scope and streams the records (needs channels.hosted and a + # gateway that knows generic sources). Each record names the catalog + # server whose Linear tools act on it. Switching it on needs a restart. + # See docs/sources.md "Linear (hosted)". + # linear: + # enabled: false + # mode: hosted # the only mode; poll refuses to load + # events: [assigned, commented, mentioned] # default: every event + # inbox: true # also mentions of the credential's Linear user + # include_bots: false # also events of bots and integrations + # drop_own_actions: true # drop the credential user's own events + # rules: # JSON Pointers into the record data; + # - path: /issue/team/key # applied by the gateway and again here + # allow: ["ENG*"] # Memory (memU) memory: diff --git a/docs/config.md b/docs/config.md index 8992bbabd..bb766cebb 100644 --- a/docs/config.md +++ b/docs/config.md @@ -242,7 +242,7 @@ A reload is always explicit. Two things cause one: | Cron jobs (`config/cron/*.yaml`) | ✅ | | Custom cron gate plugins (new, edited *and* deleted `.py` files) | ✅ the registry is rebuilt from the directory | | Cron file locations (`cron.jobs_file`, `system_file`, `gate_plugins_dir`) | ✅ re-read from the new config | -| Cron sources: `sync.telegram`, `.gmail`, `.github`, `.github_events`, `.github_repos`, `.message_ttl_days`, and each source's `schedule` | ✅ runners are rebuilt and rescheduled, all or nothing: a `schedule` the scheduler will not take refuses the source reload with the running sources on their old triggers, and reports the error rather than `ok`. In hosted mode the GitHub filters, `pr_watch`, `inbox` and the repository watch (`sync.github_repos`) are published to the gateway again. **`sync.codex` and `sync.github.mode` are not among these**; see the restart table | +| Cron sources: `sync.telegram`, `.gmail`, `.github`, `.github_events`, `.github_repos`, `.linear`, `.message_ttl_days`, and each source's `schedule` | ✅ runners are rebuilt and rescheduled, all or nothing: a `schedule` the scheduler will not take refuses the source reload with the running sources on their old triggers, and reports the error rather than `ok`. In hosted mode the GitHub filters, `pr_watch`, `inbox` and the repository watch (`sync.github_repos`) are published to the gateway again, and so are the Linear settings and rules. **`sync.codex` and `sync.github.mode` are not among these**; see the restart table | | `cron.delivery_debounce_seconds` | ✅ read for each delivery | | MCP servers (`mcp_servers`) | ✅ new sessions get the new set | | Skills (`skills/`, and the platform skills in `skills/` in the config directory) | ✅ re-scanned | @@ -298,6 +298,7 @@ reload cannot inspect, and are documented here only. | `mcp_gateway_url` | the catalog reader is built at startup. See [MCP servers in external mode](#mcp-servers-in-external-mode) | | `slack.mode`, `channels.hosted.*` | the hosted channels, the email channel included, their stream endpoint, and their token verifier are built at startup. A reload reads the file that `gateway_jwks_file` names again, but not a changed path. Changing `mode` to `socket` reports a Slack reload error until a restart, because the hosted channel still holds the `slack` name. Changing it to `hosted` stops the Socket Mode connection at once, and the stream endpoint opens after a restart | | `sync.github.mode` | the GitHub sources (`github` and the repository watch, `github_repos`), and in hosted mode the stream endpoint and the consumers of the gateway's records, are chosen at startup. Until a restart the sources keep the mode they started with, and so does the `github_pr_activity` gate. The filters, `pr_watch`, `inbox` and `sync.github_repos` do follow a reload: in hosted mode the reload publishes them to the gateway again. To go back to `poll`, set `enabled: false` (and `sync.github_repos.enabled: false`) and reload first (see [sources.md](sources.md#hosted-github-source)) | +| Turning `sync.linear.enabled` **on** | the consumer of the gateway's Linear records is built at startup, only while the source is on. Turning it **off** is hot: the reload tells the gateway that Nerve takes no Linear records. The events, the other settings and the rules follow a reload | | `auth.jwt_secret` | pinned at startup for every consumer, web gateway and MCP endpoint alike. A reload that changes or removes it is reported and changes nothing live: removing the key must not reopen the instance, and rotating it must not swap the key under live sessions half-way. The next restart applies it (with the key removed, the secret generated into `nerve.db` takes over) | | `workflows.enabled`, `workflows.review_loop.enabled` | each service is created at startup and only when its flag is on. Turning one **off** does not stop the service already running, and turning it **on** creates nothing for a reload to reach | | `workflows.poll_interval_seconds`, `workflows.review_loop.reconcile_interval_seconds` | both loops were handed their interval when they started. Everything else under `workflows.*` is read per use (see the table above) | @@ -1457,13 +1458,15 @@ notifications, and with `sync.github_repos` the repositories it watches; the `channels.hosted` settings above apply unchanged, and they are enough on their own: Slack need not be hosted too. See [Hosted GitHub source](sources.md#hosted-github-source) and -[Hosted repository watch](sources.md#hosted-repository-watch). They also +[Hosted repository watch](sources.md#hosted-repository-watch). With +`sync.linear.enabled` the gateway also polls Linear with the agent's Linear +connector (see [Linear](sources.md#linear-hosted)). They also carry the mail that members send to the agent: the usable `channels.hosted` settings give the email channel, which `email.enabled` switches on (see [Email](#email)). -When Slack is hosted as well, its switch also holds GitHub and member mail -back. Nerve reads the events of every hosted provider from one ordered +When Slack is hosted as well, its switch also holds GitHub, Linear and member +mail back. Nerve reads the events of every hosted provider from one ordered inbox, and it reads nothing while hosted Slack is switched off (`slack.enabled: false`): the GitHub records wait at the gateway until Slack is switched on again, and past the gateway's limit of stored records the @@ -2008,6 +2011,28 @@ reload: | `sync.github_repos.updates` | bool | `false` | Also report a new head commit, ready for review, and the merge or close of each pull request the watch reported | | `sync.github_repos.deny_authors` | list | `[]` | Case-insensitive globs on the author's login, with the guardrail rules; what they match is never reported, releases included | +**Linear-specific** (issue events of the agent's Linear connector in the MCP +catalog; see [sources.md](sources.md#linear-hosted)). The channel gateway +polls Linear with the connector's credential and scope and streams the +records, so the source needs the `channels.hosted` settings (see +[Hosted mode](#hosted-mode)) and a gateway that knows generic sources. The +settings and the rules go to the gateway in the subscription, and the rules +are applied here again. Everything follows a reload, except that switching +the source on needs a restart: + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `sync.linear.enabled` | bool | `false` | Take the Linear records from the channel gateway. Switching it on takes effect at the next restart; switching it off follows a reload | +| `sync.linear.mode` | string | `hosted` | `hosted`, the only mode. `poll` and any other value refuse to load | +| `sync.linear.events` | list | every event | What is reported: `issue_created`, `assigned`, `unassigned`, `delegated`, `state_changed`, `priority_changed`, `moved_in`, `commented`, `mentioned`. Case does not matter; an unknown event or an empty list refuses to load. With the default, an event that a later gateway adds is reported too | +| `sync.linear.inbox` | bool | `true` | Also report the mentions of the credential's Linear user, which the gateway reads from its Linear inbox (`mentioned`, `via: inbox`). Only a credential that belongs to the agent (an agent API key, the agent's sign-in) has one; assignments, comments and the other changes come from the issues of the scope either way | +| `sync.linear.include_bots` | bool | `false` | Also report the events of bots and integrations | +| `sync.linear.drop_own_actions` | bool | `true` | Drop the events of the credential's Linear user. With a tenant API key they are kept and marked with `actor.credential_user: true` | +| `sync.linear.rules` | list | `[]` | At most 16 rules, each a `path` (a JSON Pointer into the record data, such as `/issue/team/key`) and exactly one non-empty `allow` or `deny` list of globs. A rule the gateway would refuse refuses to load. The subscription is never sent without its rules: rules too large for the gateway's frame leave Linear out (`nerve doctor` reports it). See [sources.md](sources.md#rules-of-generic-hosted-sources) | +| `sync.linear.schedule` | cron | `*/15 * * * *` | Fallback cadence of the drain, which also runs on every delivery | +| `sync.linear.batch_size` | int | `50` | Max records per drain | +| `sync.linear.condense` | bool | `false` | LLM-condense long records via `memory.fast_model` | + `cron.delivery_debounce_seconds` (int, default `60`) spaces the job runs that a hosted delivery starts; a negative value reads as `0`, and a value above `600` waits `600` seconds (`nerve config validate` warns). See diff --git a/docs/cron.md b/docs/cron.md index 6566c20c7..c5dfc14b4 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -707,8 +707,9 @@ See [sources.md](sources.md) for full documentation. ### Runs on delivery -A hosted source — GitHub with `sync.github.mode: hosted`, see -[Hosted GitHub source](sources.md#hosted-github-source) — does not wait for its +A hosted source (GitHub with `sync.github.mode: hosted`, see +[Hosted GitHub source](sources.md#hosted-github-source), or Linear, see +[Linear (hosted)](sources.md#linear-hosted)) does not wait for its schedule. When the channel gateway delivers records, the source's runner runs at once, after any run of it already in progress (which may have fetched before the records arrived); deliveries during that drain make it run once diff --git a/docs/sources.md b/docs/sources.md index 6b50d6083..af9d73c6c 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -428,6 +428,110 @@ machine for them. platform that parks the machine does not wake it for that run. The mode follows `sync.github.mode`, which is restart-only. +### Linear (hosted) +- **Adapter:** `nerve/sources/hosted.py` (`HostedSource`). It drains the records that the channel gateway delivered. This process makes no Linear calls and needs no Linear key. + +With `sync.linear.enabled: true`, the shared channel gateway (the one of +[hosted Slack](config.md#hosted-mode) and the +[hosted GitHub source](#hosted-github-source), set up with the same +`channels.hosted` settings) polls Linear for the agent and streams the records +to Nerve. It uses the agent's Linear connector in the MCP catalog: its +credential, its project scope, and its GitHub filter. A record holds only what +the connector would show the agent. The gateway keeps the records until Nerve +reads them, also while Nerve is not running. + +- **The connector is the switch.** The gateway polls Linear for an agent only + while the agent has a Linear connector in the MCP catalog and Nerve + subscribes, which it does while `enabled` is true. An agent with more than + one Linear connector gets the records of each one. +- **What arrives:** `linear_issue_event` records, one for each event: + `issue_created`, `assigned`, `unassigned`, `delegated`, `state_changed`, + `priority_changed`, `moved_in` (moved into a project of the scope), + `commented` and `mentioned`. The summary is one line, such as + `[ENG-12] Alice assigned you: Fix the login page`, and the content is + Markdown for the agent. Assignments, comments and the other changes come + from the issues of the scope. `mentioned` comes from the Linear inbox of + the credential's Linear user: only a credential that belongs to the agent + (an agent API key, or the agent's sign-in) has one, and the gateway reads + only the mentions from it. An item that is too large for the gateway to + read is left out, with no record. +- **The record keeps its schema and its data.** The gateway sends the data of + a record as a JSON object of the record schema `linear.issue_event/1`, and + Nerve stores both as the record's metadata, `{"schema": ..., "data": ...}`, + with every member of the data, also the members that this release does not + know. The poll tools show them to the agent as a `Schema:` line and a + `Data:` line of JSON. The members of the data (optional members are left + out, never `null`): + - `event`, and `via`: `watch` for an issue of the scope, `inbox` for a + mention from the Linear inbox (only with `mentioned`); + - `server`: the `id` of the Linear server in the MCP catalog (its catalog + server ID, such as `linear`) and its `name`; + - `issue`: `id`, `identifier`, `title`, `url`, `team` (`key`), `project` + (`id`, `name`), `state` (`name`, `type`), `priority` (0 to 4), + `assignee` and `delegate` (`id`, `name`); + - `comment`: `id`, `url`, `body` (at most 4,000 characters; `truncated` is + true when it was cut), `parent_id`; + - `actor`: `id`, `name`, `kind` (`user`, `app`, `bot`, `integration`, + `external`, or `system` for a change without an actor), and + `credential_user`; + - `changes`: a list of `{field, from, to}`, where `from` and `to` are a + name or a priority. +- **One record for each event.** The record ID is + `::`. A changed event, such as a comment that was + edited, keeps it, so the new version replaces the stored one and is unread + again. +- **Act on a record with the Linear tools of the server that it names** + (`server.id`). That MCP catalog server reaches the same issues, with the + same scope. +- **The settings go to the gateway.** `events`, `inbox`, `include_bots` and + `drop_own_actions` go in the subscription as the provider settings. Only the + values that differ from the defaults are sent, so with the default `events` + an event that a later gateway adds arrives too. `inbox` switches the + mentions of the credential's Linear user on or off; it has an effect only + with a credential that belongs to the agent. `drop_own_actions` drops the + events of the credential's Linear user. A tenant API key belongs to a + person, so the events of that person are kept and marked with + `actor.credential_user: true`, whatever the setting is. To drop them, add the + rule `{path: /actor/credential_user, deny: ["true"]}`. +- **The rules go to the gateway too.** The gateway applies `rules` before it + stores a record, and the guardrail here applies them again; see + [Rules of generic hosted sources](#rules-of-generic-hosted-sources). A rule + that the gateway would refuse refuses to load. The subscription is never + sent without its rules: if the subscriptions frame with the rules is larger + than the gateway takes, Nerve leaves the Linear subscription out, logs an + error, and the gateway stores no Linear records until the rules are shorter. + `nerve doctor` reports this as an error. +- **The gateway must know generic sources.** A gateway that predates them + closes a stream whose subscriptions carry `settings`, `schemas` or `rules`, + hosted Slack traffic included. Update the gateway before you switch Linear on. +- **Same runner model as hosted GitHub.** The records go to a local buffer, + and the `source:linear` runner drains it as soon as records are delivered. + Its scheduled run is a fallback that is not the instance's next scheduled + task while nothing waits. Consumer cursors, `messages` gates, `condense` and + the Sources page work as for the other sources; the Sources page shows each + record with a Linear card. +- **A switched-off hosted Slack holds Linear back**, as it holds hosted GitHub + back (see [Hosted GitHub source](#hosted-github-source)). +- **Hosted only.** `mode` is `hosted`, also when it is not set. `mode: poll` + refuses to load: this release has no Linear source that polls from this + process. +- **Switching it on needs a restart.** The consumer of the gateway's Linear + records is built at startup, so `enabled: true` takes effect at the next + restart. Switching it off follows a reload: Nerve tells the gateway that it + takes no Linear records. The other settings follow a reload. + +```yaml +sync: + linear: + enabled: true + events: [assigned, commented, mentioned] # default: every event + rules: + - path: /issue/team/key + allow: ["ENG*"] # applied by the gateway and again here +``` + +`nerve doctor` reports the source and checks the hosted settings. + ### 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 @@ -668,6 +772,18 @@ sync: deny_authors: [] # Globs on the author's login; never reported batch_size: 50 condense: false + + linear: + enabled: false # The channel gateway polls Linear with the agent's Linear connector + mode: hosted # The only mode; poll refuses to load + events: [issue_created, assigned, unassigned, delegated, state_changed, priority_changed, moved_in, commented, mentioned] + inbox: true # Also the mentions of the credential's Linear user (agent credentials only) + include_bots: false # Also the events of bots and integrations + drop_own_actions: true # Drop the events of the credential's Linear user (not with a tenant API key) + rules: [] # Allow/deny globs on JSON Pointers into the data. See "Rules of generic hosted sources" + schedule: "*/15 * * * *" # Fallback cadence of the drain + batch_size: 50 + condense: false ``` ### Config Fields @@ -875,11 +991,13 @@ drop them; `deny_repos` and an `allow_repos` list that the repo does not match d ### Debugging drops Dropped records are never persisted, so the run summary count is backed by per-record log -lines naming the rule that fired: +lines naming the rule that fired. They name the record ID, the rule's field and the list +that dropped it, and never a value or the summary of the dropped record, so what the +guardrail keeps from the inbox does not reach the log either: ``` -Source github: guardrail dropped 3/5 records (e.g. '[owner/repo] CI workflow run failed …') -Source github: dropped 19542… on ci_branch='main' — [owner/repo] CI workflow run failed … +Source github: guardrail dropped 3/5 records +Source github: dropped 19542… by the deny list of the rule on ci_branch ``` ### Drop hook @@ -961,6 +1079,49 @@ Rules a hook can rely on: |-------|------|---------|-------------| | `github.drop_hook` | list | `[]` | argv of a command that gets each dropped record as a JSON line on stdin. Passed as is: no shell, nothing stripped. Empty = off | +### Rules of generic hosted sources + +A generic hosted source, such as [Linear](#linear-hosted), stores the data of +each record (a JSON object) as the record's metadata. Its rules +(`sync.linear.rules`) name a value in that data with a JSON Pointer (RFC 6901) +and match it with an `allow` or a `deny` list of glob patterns. The channel +gateway applies the same rules before it stores a record, so the two sides +drop the same records, and the guardrail here (`DataRule` in +`nerve/sources/filters.py`) applies them again. + +```yaml +sync: + linear: + rules: + - path: /issue/team/key + allow: ["ENG*", "OPS"] + - path: /actor/kind + deny: [bot, integration] +``` + +- **The path** starts with `/`, has 1 to 8 parts and at most 256 bytes, and + writes `~` as `~0` and `/` as `~1`. A part names the member of an object, or + the element of an array at its index (`/changes/0/field`). +- **The values at the path:** a string as it is; a number as its shortest + JSON text (`2` for `2.0`, `1.5`); `true`, `false` and `null` as those words; + and each such element of an array. An object, an array or an object in an + array, an empty array, and a path that names nothing give no value. +- **Each rule has exactly one non-empty list**, `allow` or `deny`, of at most + 64 patterns of at most 256 bytes without control characters. Patterns match + as the other guardrail patterns do: case-insensitive shell globs on the + stripped value. A `deny` rule drops a record when one of its values matches. + An `allow` rule drops a record when none of its values matches, also a record + without a value at its path. A record stays only when every rule keeps it. +- **At most 16 rules.** A rule outside these bounds, or with unknown keys, + refuses to load, so the gateway never closes the stream over one. A + subscription is never sent without its rules (see [Linear](#linear-hosted)). +- **The paths are into the data.** The record's schema, which Nerve keeps next + to the data, is not part of it. + +The runner's drop log names the record ID, the path of the rule that dropped +it, and whether its `allow` or its `deny` list did, and nothing of the record's +data. + ### Extending to other sources Adding a guardrail to another source is a config field plus one registry line. In @@ -1173,7 +1334,11 @@ web/src/components/Sources/ ├── index.ts ← Registry: source name → renderer type mapping ├── MarkdownRenderer.tsx ← Default: ReactMarkdown with prose styles ├── EmailRenderer.tsx ← Gmail: sandboxed iframe for HTML + text toggle - └── GitHubRenderer.tsx ← GitHub: repo/PR header card + markdown body + ├── GitHubRenderer.tsx ← GitHub: repo/PR header card + markdown body + ├── LinearRenderer.tsx ← Records of schema linear.issue_event/1: issue card + markdown body + the record data + ├── DataRenderer.tsx ← A record of another schema: markdown body + the record data + ├── RecordData.tsx ← The record data as an expandable JSON view, with its schema + └── storedRecord.ts ← The schema and the data of a stored generic record (metadata {schema, data}) ``` ### Steps @@ -1212,7 +1377,9 @@ export function MyRenderer({ content, metadata, summary }: Props) { | `nerve/sources/models.py` | `SourceRecord`, `FetchResult`, `IngestResult` dataclasses | | `nerve/sources/base.py` | `Source` abstract base class | | `nerve/sources/runner.py` | `SourceRunner` — pure ingestion pipeline (fetch → preprocess → guardrail → persist → condense → drop hook → advance) | -| `nerve/sources/filters.py` | `InboxFilter` / `FieldRule` — declarative allow/deny guardrails applied before persist | +| `nerve/sources/filters.py` | `InboxFilter` / `FieldRule` / `DataRule`: declarative allow/deny guardrails applied before persist | +| `nerve/sources/source_json.py` | The JSON data of generic hosted sources: checks, canonical form (RFC 8785), JSON Pointers | +| `nerve/sources/hosted.py` | `HostedSource`: drain the records of a generic hosted source (Linear) that the channel gateway delivered | | `nerve/sources/processor.py` | Legacy agent prompt building (unused by runner, may be used by tools) | | `nerve/sources/registry.py` | Config → `list[SourceRunner]` factory | | `nerve/sources/telegram.py` | Telegram adapter (Telethon) | @@ -1222,7 +1389,7 @@ export function MyRenderer({ content, metadata, summary }: Props) { | `nerve/sources/github_events.py` | GitHub user events adapter (gh CLI) | | `nerve/sources/github_repos.py` | GitHub repository watch adapter (gh CLI), record builder, its guardrail | | `nerve/agent/tools.py` | Consumer tools: `list_sources`, `poll_source`, `read_source` | -| `web/src/components/Sources/` | Source message renderers (email, GitHub, default markdown) | +| `web/src/components/Sources/` | Source message renderers (email, GitHub, Linear, record data, default markdown) | ### Cursor Design diff --git a/nerve/agent/tools/handlers/sources.py b/nerve/agent/tools/handlers/sources.py index ee062387a..36a591d1f 100644 --- a/nerve/agent/tools/handlers/sources.py +++ b/nerve/agent/tools/handlers/sources.py @@ -7,6 +7,7 @@ from __future__ import annotations +import json import logging from datetime import datetime, timezone @@ -19,6 +20,7 @@ SYNC_STATUS_SCHEMA, ) from nerve.sources.models import cursor_text +from nerve.sources.source_json import record_data logger = logging.getLogger(__name__) @@ -75,7 +77,13 @@ def _format_source_batch(messages: list[dict], source: str | None = None) -> str parts.append(f"**Type:** {record_type} | **Time:** {timestamp} ({relative}) | **seq:** {rowid}") metadata = m.get("metadata") - if metadata and isinstance(metadata, dict): + generic = record_data(metadata) + if generic is not None: + # The data of a generic hosted source, every member as it came. + schema, data = generic + parts.append(f"**Schema:** {schema}") + parts.append(f"**Data:** {json.dumps(data, ensure_ascii=False, separators=(',', ':'))}") + elif metadata and isinstance(metadata, dict): interesting = {k: v for k, v in metadata.items() if v and k not in ("message_id",)} if interesting: meta_str = ", ".join(f"{k}={v}" for k, v in interesting.items()) diff --git a/nerve/channels/hosted/__init__.py b/nerve/channels/hosted/__init__.py index fdbaadbbd..095bb3ff5 100644 --- a/nerve/channels/hosted/__init__.py +++ b/nerve/channels/hosted/__init__.py @@ -6,6 +6,7 @@ gateway's inbox over them. Hosted sources use the same streams: the gateway polls a source (GitHub, with -``sync.github.mode: hosted``) under the filters that Nerve publishes, and -Nerve reads the records like any other event (see :mod:`.sources`). +``sync.github.mode: hosted``; Linear, with ``sync.linear.enabled``) under the +filters that Nerve publishes, and Nerve reads the records like any other +event (see :mod:`.sources`). """ diff --git a/nerve/channels/hosted/contract/__init__.py b/nerve/channels/hosted/contract/__init__.py index b5a36e569..c4c01b940 100644 --- a/nerve/channels/hosted/contract/__init__.py +++ b/nerve/channels/hosted/contract/__init__.py @@ -36,6 +36,7 @@ RepoWatch, SourceFilters, SourceRecordPayload, + SourceRule, SourceSubscription, StreamLimits, Subscriptions, @@ -79,6 +80,7 @@ "RepoWatch", "SourceFilters", "SourceRecordPayload", + "SourceRule", "SourceSubscription", "StreamLimits", "Subscriptions", diff --git a/nerve/channels/hosted/contract/model.py b/nerve/channels/hosted/contract/model.py index 73d856847..5c49f13e7 100644 --- a/nerve/channels/hosted/contract/model.py +++ b/nerve/channels/hosted/contract/model.py @@ -13,6 +13,7 @@ import re import unicodedata import uuid +from collections.abc import Iterable from datetime import datetime, timezone from typing import Any @@ -24,8 +25,10 @@ omit_empty, parse_json, reject, + source_json, struct, ) +from nerve.sources.source_json import SourceJSONError, check_source_json_object, pointer_tokens _record = dataclasses.dataclass(frozen=True) @@ -81,6 +84,29 @@ # A provider name is a lowercase token of at most this many bytes. MAX_PROVIDER_BYTES = 64 _PROVIDER_CHARACTERS = frozenset("abcdefghijklmnopqrstuvwxyz0123456789_-.") +# The provider of GitHub notifications and pull request activity. +GITHUB_PROVIDER = "github" +# The providers with a closed vocabulary: their records carry metadata from a +# closed set of keys, and their subscriptions have members of their own. +# Every other provider is a generic source: its records carry a schema and +# data that the provider defines, and its subscriptions carry settings, +# record schemas, and rules (see SourceSubscription). +CLOSED_PROVIDERS = frozenset({GITHUB_PROVIDER, REPO_WATCH_PROVIDER, EMAIL_PROVIDER}) +# The conversation kind of a tracked item, such as an issue. +CONVERSATION_KIND_ITEM = "item" +# Bounds of generic sources. The data of a record and the settings of a +# subscription count the UTF-8 bytes of their canonical form. +MAX_SOURCE_DATA_BYTES = 32 * 1024 +MAX_SOURCE_SETTINGS_BYTES = 8 * 1024 +MAX_RECORD_TYPE_BYTES = 64 +MAX_RECORD_SCHEMA_MAJOR_DIGITS = 6 +MAX_SUBSCRIPTION_SCHEMAS = 16 +MAX_SOURCE_RULES = 16 +MAX_SOURCE_RULE_PATTERNS = 64 +# The name of a generic record type after its provider and "_". +_RECORD_TYPE_NAME = re.compile(r"[a-z0-9_]+") +# The major version of a record schema: decimal, without a leading zero. +_MAJOR_VERSION = re.compile(rf"[1-9][0-9]{{0,{MAX_RECORD_SCHEMA_MAJOR_DIGITS - 1}}}") # How Nerve answers the turns of a connection's events (ConnectionBehaviour). ANSWER_POST = "post" ANSWER_TOOL = "tool" @@ -155,6 +181,54 @@ def valid_provider(provider: str) -> bool: ) +def generic_provider(provider: str) -> bool: + """Whether *provider* is a generic source (not in :data:`CLOSED_PROVIDERS`).""" + return provider not in CLOSED_PROVIDERS + + +def valid_record_type(provider: str, record_type: str) -> bool: + """Whether a generic *provider* can produce *record_type*. + + That is the provider, ``_``, and a name of lowercase ASCII letters, + digits and ``_``, at most :data:`MAX_RECORD_TYPE_BYTES` in total, such as + ``linear_issue_event``. + """ + prefix = f"{provider}_" + return ( + record_type.startswith(prefix) + and byte_length(record_type) <= MAX_RECORD_TYPE_BYTES + and _RECORD_TYPE_NAME.fullmatch(record_type[len(prefix):]) is not None + ) + + +def schema_record_type(provider: str, schema: str) -> str: + """The record type that a record schema of a generic *provider* names. + + A schema is the provider, ``.``, the record type's name, ``/``, and its + major version (up to :data:`MAX_RECORD_SCHEMA_MAJOR_DIGITS` digits, + without a leading zero): ``linear.issue_event/1`` names the record type + ``linear_issue_event``. Anything else is malformed. + """ + prefix = f"{provider}." + name, slash, major = schema[len(prefix):].partition("/") + record_type = f"{provider}_{name}" + if not ( + schema.startswith(prefix) and slash and generic_provider(provider) + and valid_record_type(provider, record_type) and _MAJOR_VERSION.fullmatch(major) + ): + malformed(f"record schema is not a schema of provider {provider!r}") + return record_type + + +def _validate_source_json(name: str, value: dict[str, Any], maximum_bytes: int) -> None: + """Check a source JSON object (the data of a record, the settings of a subscription).""" + try: + check_source_json_object(value, maximum_bytes) + except SourceJSONError as error: + reason = RejectionReason.LIMIT_EXCEEDED if error.limit else RejectionReason.MALFORMED_FRAME + reject(reason, f"{name} {error}") + + # ---------------------------------------------------------------------- # # References and content # # ---------------------------------------------------------------------- # @@ -363,22 +437,45 @@ class SourceRecordPayload: ``id`` is the source's own key for the record. A changed record keeps its ID, and Nerve stores it in place of the earlier copy. + + A record of a provider in :data:`CLOSED_PROVIDERS` carries ``metadata`` + from the closed vocabulary of its record type, and no ``schema`` or + ``data``. A record of a generic source carries ``schema`` and ``data`` + and no metadata: ``data`` is a JSON object that the provider defines and + that the schema describes. Nerve keeps every member of it, also the ones + that it does not know. """ id: str = "" record_type: str = "" + schema: str = omit_empty() summary: str = "" content: str = omit_empty() timestamp: datetime = ZERO_TIME metadata: tuple[MetadataEntry, ...] = omit_empty(()) + data: dict[str, Any] | None = source_json() - def validate(self) -> None: + def validate(self, provider: str) -> None: + """Check the record of a source of *provider*, as the gateway does.""" if not self.id or not self.record_type or not self.summary: malformed("source record has no ID, record type, or summary") if max(byte_length(self.id), byte_length(self.record_type)) > MAX_IDENTIFIER_BYTES: reject(RejectionReason.LIMIT_EXCEEDED, "source record ID or type exceeds the identifier limit") if max(byte_length(self.summary), byte_length(self.content)) > MAX_TEXT_BYTES: reject(RejectionReason.LIMIT_EXCEEDED, "source record text exceeds the text limit") + if not generic_provider(provider): + if self.schema or self.data is not None: + malformed(f"a {provider} source record carries a schema or data") + return + if not valid_record_type(provider, self.record_type): + reject(RejectionReason.KIND_UNSUPPORTED, f"record type is not valid for provider {provider!r}") + if schema_record_type(provider, self.schema) != self.record_type: + malformed("source record schema names another record type") + if self.metadata: + malformed("a generic source record carries metadata") + if self.data is None: + malformed("a generic source record has no data") + _validate_source_json("source record data", self.data, MAX_SOURCE_DATA_BYTES) @_record @@ -426,7 +523,7 @@ def validate(self) -> None: if self.kind == EVENT_KIND_SOURCE_RECORD: if self.record is None: malformed("source record event has no record") - self.record.validate() + self.record.validate(self.provider) # ---------------------------------------------------------------------- # @@ -858,6 +955,43 @@ class RepoWatch: deny_authors: tuple[str, ...] = () +@_record +class SourceRule: + """Keeps or drops the records of a generic source by a value of their data. + + ``path`` is a JSON Pointer into the data, such as ``/actor/kind``. A rule + has exactly one non-empty list of glob patterns, ``allow`` or ``deny``, + which match as the filter patterns do. A deny rule drops a record when + one of the values at its path matches; an allow rule drops a record when + none does, also when the record has no value there. The gateway applies + the rules before it stores a record, and Nerve's inbox guardrail applies + them again (:class:`nerve.sources.filters.DataRule`, which also defines + the values at a path). + """ + + path: str = "" + allow: tuple[str, ...] = omit_empty(()) + deny: tuple[str, ...] = omit_empty(()) + + def validate(self) -> None: + """Check the rule as the gateway does. + + The path is a bounded JSON Pointer (see + :func:`nerve.sources.source_json.pointer_tokens`), and the rule has + one list of 1 to :data:`MAX_SOURCE_RULE_PATTERNS` patterns that a + filter list takes. + """ + if pointer_tokens(self.path) is None: + malformed("subscription rule path is not a bounded JSON Pointer") + if bool(self.allow) == bool(self.deny): + malformed("a subscription rule needs exactly one of allow and deny") + patterns = self.deny or self.allow + if len(patterns) > MAX_SOURCE_RULE_PATTERNS: + reject(RejectionReason.LIMIT_EXCEEDED, "a subscription rule has too many patterns") + if not all(valid_pattern(pattern) for pattern in patterns): + malformed("a subscription rule pattern is empty, too long, or has a control character") + + @_record class SourceSubscription: """What the agent takes from one provider's hosted source. @@ -874,6 +1008,16 @@ class SourceSubscription: is switched on, so every other frame is unchanged. The ``email`` subscription carries only its provider and empty filters. + + A subscription of a generic source names its record ``schemas``: the + versions of the data that Nerve reads, at least one. For each record + type, the gateway sends the highest major version that ``schemas`` names + and that it produces. It then carries ``settings``, a JSON object that + the provider defines (absent settings are the provider's defaults), and + ``rules`` on the data of its records (see :class:`SourceRule`), and none + of the members above other than empty filters. A subscription of a + provider outside :data:`CLOSED_PROVIDERS` without schemas is not generic, + and carries no settings or rules either. """ provider: str = "" @@ -881,6 +1025,9 @@ class SourceSubscription: pr_watch: bool = omit_empty(False) inbox: str = omit_empty("") repos: tuple[RepoWatch, ...] = omit_empty(()) + settings: dict[str, Any] | None = source_json() + schemas: tuple[str, ...] = omit_empty(()) + rules: tuple[SourceRule, ...] = omit_empty(()) @_record @@ -899,9 +1046,11 @@ def validate(self) -> None: The version, the number of sources, each provider name and that no provider appears twice, each inbox, the bounds of every filter list, - the watched repositories, which only ``github_repos`` names (see - :func:`_validate_repo_watches`), and that an ``email`` subscription - carries nothing but its provider. + the watched repositories (see :func:`_validate_repo_entries`), which + ``github``, ``email`` and a generic source do not name and + ``github_repos`` must name, that an ``email`` subscription + carries nothing but its provider, and the members of a generic + source (see :func:`_validate_generic_members`). """ if self.version != SUBSCRIPTIONS_VERSION: reject(RejectionReason.VERSION_UNSUPPORTED, f"unknown subscriptions version {self.version!r}") @@ -922,21 +1071,25 @@ def validate(self) -> None: malformed("source subscription inbox is not 'participating' or 'all'") for patterns in source.filters.pattern_lists().values(): _validate_patterns(patterns, "source filter") + _validate_repo_entries(source.repos) if source.provider == REPO_WATCH_PROVIDER: - _validate_repo_watches(source) - elif source.repos: - malformed(f"only a {REPO_WATCH_PROVIDER} subscription names repositories") + _validate_repo_watch(source) + elif source.repos and not generic_provider(source.provider): + malformed(f"a {source.provider} subscription names repositories") if source.provider == EMAIL_PROVIDER and ( any(source.filters.pattern_lists().values()) or source.pr_watch or source.inbox ): malformed(f"an {EMAIL_PROVIDER} subscription carries filters, pr_watch or an inbox") + _validate_generic_members(source) def without_filters(self) -> Subscriptions: """The same sources with no filter patterns. The gateway then keeps every record of those sources, and Nerve's own inbox filter is the only one that drops records. A watched - repository's ``deny_authors`` are filter patterns too. + repository's ``deny_authors`` are filter patterns too. The rules of a + generic source stay: it is never sent wider (see + :meth:`without_providers`). """ return dataclasses.replace(self, sources=tuple( dataclasses.replace( @@ -947,6 +1100,63 @@ def without_filters(self) -> Subscriptions: for source in self.sources )) + def ruled_providers(self) -> tuple[str, ...]: + """The providers whose subscriptions carry rules, in frame order. + + Such a subscription is never sent without its rules: the gateway + would then store the records that the rules drop. + """ + return tuple(source.provider for source in self.sources if source.rules) + + def without_providers(self, providers: Iterable[str]) -> Subscriptions: + """The same subscriptions without the sources of *providers*. + + The gateway stores nothing for a provider that the frame leaves out. + """ + left_out = set(providers) + return dataclasses.replace( + self, sources=tuple(source for source in self.sources if source.provider not in left_out), + ) + + +def _validate_generic_members(source: SourceSubscription) -> None: + """Check ``settings``, ``schemas`` and ``rules`` as the gateway does. + + Their counts are bounded (:data:`MAX_SUBSCRIPTION_SCHEMAS`, + :data:`MAX_SOURCE_RULES`). A subscription is generic when its provider + is outside :data:`CLOSED_PROVIDERS` and it names at least one schema. + Then it has no filter pattern, ``pr_watch``, ``inbox`` or ``repos``; its + settings are a source JSON object within + :data:`MAX_SOURCE_SETTINGS_BYTES`; its schemas are distinct schemas of its + provider (see :func:`schema_record_type`); and its rules are valid. Any + other subscription names none of the three. A member with its zero value + counts as absent, except the settings, which count whenever they are + present. + """ + if len(source.schemas) > MAX_SUBSCRIPTION_SCHEMAS: + reject(RejectionReason.LIMIT_EXCEEDED, "a subscription names too many record schemas") + if len(source.rules) > MAX_SOURCE_RULES: + reject(RejectionReason.LIMIT_EXCEEDED, "a subscription has too many rules") + if not (generic_provider(source.provider) and source.schemas): + if source.settings is not None or source.schemas or source.rules: + malformed( + "a subscription of a closed provider, or without record schemas, names " + "settings, record schemas, or rules" + ) + return + if any(source.filters.pattern_lists().values()) or source.pr_watch or source.inbox or source.repos: + malformed("a generic subscription names filters, a pull request watch, an inbox, or repositories") + if source.settings is not None: + _validate_source_json("subscription settings", source.settings, MAX_SOURCE_SETTINGS_BYTES) + seen: set[str] = set() + for schema in source.schemas: + schema_record_type(source.provider, schema) + if schema in seen: + malformed("subscription schemas repeat a schema") + seen.add(schema) + for rule in source.rules: + rule.validate() + def _validate_patterns(patterns: tuple[str, ...], what: str) -> None: """The bounds of one pattern list: its length and each pattern.""" @@ -959,31 +1169,24 @@ def _validate_patterns(patterns: tuple[str, ...], what: str) -> None: ) -def _validate_repo_watches(source: SourceSubscription) -> None: - """Check a ``github_repos`` subscription as the gateway does. +def _validate_repo_entries(repos: tuple[RepoWatch, ...]) -> None: + """Check the watched repositories of any subscription, as the gateway does. - It names 1 to :data:`MAX_REPO_WATCHES` repositories, each an - ``owner/repo`` name (:func:`valid_repo_name`) and each once, ignoring - case; every one reports one or more of :data:`REPO_WATCH_TYPES`, each - once, and has at most :data:`MAX_SUBSCRIPTION_PATTERNS` ``deny_authors`` - that a filter list would take. It carries no filters, ``pr_watch`` or - ``inbox``. + At most :data:`MAX_REPO_WATCHES`, each an ``owner/repo`` name + (:func:`valid_repo_name`) and each once, ignoring case; every one + reports one or more of :data:`REPO_WATCH_TYPES`, each once, and has at + most :data:`MAX_SUBSCRIPTION_PATTERNS` ``deny_authors`` that a filter + list would take. Which providers may name repositories at all is + checked by the caller. """ - if any(source.filters.pattern_lists().values()) or source.pr_watch or source.inbox: - malformed(f"a {REPO_WATCH_PROVIDER} subscription carries filters, pr_watch or an inbox") - if not source.repos: - malformed(f"a {REPO_WATCH_PROVIDER} subscription names no repository") - if len(source.repos) > MAX_REPO_WATCHES: - reject( - RejectionReason.LIMIT_EXCEEDED, - f"a {REPO_WATCH_PROVIDER} subscription names too many repositories", - ) + if len(repos) > MAX_REPO_WATCHES: + reject(RejectionReason.LIMIT_EXCEEDED, "a subscription names too many repositories") names: set[str] = set() - for repo in source.repos: + for repo in repos: if not valid_repo_name(repo.name): malformed("a watched repository is not an owner/repo name") if repo.name.lower() in names: - malformed(f"a {REPO_WATCH_PROVIDER} subscription repeats a repository") + malformed("a subscription repeats a repository") names.add(repo.name.lower()) for kind in repo.types: if kind not in REPO_WATCH_TYPES: @@ -999,6 +1202,18 @@ def _validate_repo_watches(source: SourceSubscription) -> None: ) +def _validate_repo_watch(source: SourceSubscription) -> None: + """Check what only a ``github_repos`` subscription must hold. + + It names at least one repository, and no filters, ``pr_watch`` or + ``inbox``. + """ + if any(source.filters.pattern_lists().values()) or source.pr_watch or source.inbox: + malformed(f"a {REPO_WATCH_PROVIDER} subscription carries filters, pr_watch or an inbox") + if not source.repos: + malformed(f"a {REPO_WATCH_PROVIDER} subscription names no repository") + + # ---------------------------------------------------------------------- # # Envelope # # ---------------------------------------------------------------------- # diff --git a/nerve/channels/hosted/contract/wire.py b/nerve/channels/hosted/contract/wire.py index 24f7674f2..18174ff98 100644 --- a/nerve/channels/hosted/contract/wire.py +++ b/nerve/channels/hosted/contract/wire.py @@ -6,6 +6,14 @@ members without a change in Nerve. A missing or null member takes the field's default. Encoding leaves out a field marked with :func:`omit_empty` when it holds its zero value. + +A field marked with :func:`source_json` holds a JSON object that a provider +defines, such as the data of a generic source record. Decoding keeps it as +the parsed object, with every member, and tells an absent member (``None``) +from a present one: a present ``null`` or another JSON type is malformed. + +The parser refuses what the gateway's parser refuses: a JSON object that +repeats a member name, and ``NaN`` or ``Infinity``. """ from __future__ import annotations @@ -63,6 +71,15 @@ def struct(factory: type) -> Any: return dataclasses.field(default_factory=factory) +def source_json() -> Any: + """A JSON object that a provider defines: ``None`` when the member is absent. + + It is left out of the JSON when it is ``None``. Any other value, an empty + object too, is sent. + """ + return dataclasses.field(default=None, metadata={"source_json": True}) + + def format_time(value: datetime) -> str: """An RFC 3339 UTC timestamp, as the contract sends it.""" utc = value.astimezone(timezone.utc) @@ -73,15 +90,30 @@ def format_time(value: datetime) -> str: def parse_json(text: str) -> Any: - """Parse one JSON value, or raise :class:`Rejected`.""" + """Parse one JSON value, or raise :class:`Rejected`. + + An object that repeats a member name, and the literals ``NaN``, + ``Infinity`` and ``-Infinity``, are malformed, as for the gateway. + """ try: - return json.loads(text) + return json.loads(text, object_pairs_hook=_unique_members, parse_constant=_not_json) except RecursionError: reject(RejectionReason.LIMIT_EXCEEDED, "JSON nesting exceeds the parser's depth") except ValueError as error: malformed(f"decode frame: {error}") +def _unique_members(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + members = dict(pairs) + if len(members) != len(pairs): + malformed("decode frame: a JSON object repeats a member name") + return members + + +def _not_json(literal: str) -> typing.NoReturn: + malformed(f"decode frame: {literal} is not a JSON value") + + _HINTS: dict[type, dict[str, Any]] = {} @@ -96,9 +128,15 @@ def decode_record(cls: type, raw: Any, path: str = "") -> Any: hints = _HINTS[cls] = typing.get_type_hints(cls) values: dict[str, Any] = {} for field in dataclasses.fields(cls): + member_path = f"{path}.{field.name}" if path else field.name + if field.metadata.get("source_json"): + if field.name in raw: + if not isinstance(raw[field.name], dict): + malformed(f"decode frame: {member_path} is not a JSON object") + values[field.name] = raw[field.name] + continue value = raw.get(field.name) if value is not None: - member_path = f"{path}.{field.name}" if path else field.name values[field.name] = _decode_value(hints[field.name], value, member_path) return cls(**values) @@ -149,6 +187,10 @@ def encode_record(record: Any) -> dict[str, Any]: value = getattr(record, field.name) if field.metadata.get("omitempty") and value in (None, False, 0, "", ()): continue + if field.metadata.get("source_json"): + if value is not None: + out[field.name] = value + continue out[field.name] = _encode_value(value) return out @@ -183,5 +225,6 @@ def dumps(record: Any) -> str: "omit_empty", "parse_json", "reject", + "source_json", "struct", ] diff --git a/nerve/channels/hosted/runtime.py b/nerve/channels/hosted/runtime.py index 91c4fbe4e..09dee094b 100644 --- a/nerve/channels/hosted/runtime.py +++ b/nerve/channels/hosted/runtime.py @@ -74,9 +74,10 @@ def hosted_channel_providers(config: NerveConfig) -> list[str]: def hosted_providers(config: NerveConfig) -> list[str]: """Every provider whose traffic comes from the gateway: chat, then sources. - A hosted source that is switched off is included too: the gateway is - told that the agent subscribes to nothing, and the records already on - their way are kept for when it is switched on. + A hosted GitHub source that is switched off is included too: the gateway + is told that the agent subscribes to nothing, and the records already on + their way are kept for when it is switched on. Linear is included while + it is switched on (see :func:`~nerve.channels.hosted.sources.hosted_sources`). """ return hosted_channel_providers(config) + hosted_sources(config) diff --git a/nerve/channels/hosted/sources.py b/nerve/channels/hosted/sources.py index 20724ecc2..45495f2f8 100644 --- a/nerve/channels/hosted/sources.py +++ b/nerve/channels/hosted/sources.py @@ -4,16 +4,25 @@ It applies the filters that Nerve publishes before it stores a record, keeps the records in the agent's inbox until Nerve reads them, also while Nerve is not running, and Nerve reads them over the same streams that carry hosted -channel traffic. GitHub is the hosted source, as a whole: with -``sync.github.mode: hosted`` both the notifications (provider ``github``) and -the repository watch (provider ``github_repos``) come from the gateway. +channel traffic. The hosted sources: + +- GitHub, as a whole: with ``sync.github.mode: hosted`` both the + notifications (provider ``github``) and the repository watch (provider + ``github_repos``) come from the gateway. Their records carry metadata from + the contract's closed vocabulary. +- Linear (provider ``linear``), with ``sync.linear.enabled``: a generic + source. Its records carry ``data``, a JSON object that the provider + defines (record schema ``linear.issue_event/1``), and its subscription + carries provider ``settings``, the record ``schemas`` that Nerve reads, + and ``rules`` on the data. :class:`HostedSourceConsumer` takes a provider's records from the inbox reader and writes each one to the local ``channel_observations`` buffer under the provider's name. The provider's source runner drains that buffer into the -inbox with the guardrail, drop hook, and condensing of the polling source -(:mod:`nerve.sources.github_hosted`). A record is acknowledged only after its -buffer row is written. +inbox with the guardrail and condensing of every source, and for GitHub the +drop hook of the polling source (:mod:`nerve.sources.github_hosted`, +:mod:`nerve.sources.hosted`). A record is acknowledged only after its buffer +row is written. :func:`subscriptions_for` builds the ``subscriptions`` frame from the same settings that build the inbox filter, so the gateway and Nerve drop the same @@ -32,10 +41,14 @@ from nerve.channels.hosted.contract import ( EVENT_KIND_SOURCE_RECORD, + PROTOCOL_VERSION, SUBSCRIPTIONS_VERSION, + Envelope, Event, + Payload, RepoWatch, SourceFilters, + SourceRule, SourceSubscription, Subscriptions, ) @@ -46,14 +59,16 @@ REPO_WATCH_PROVIDER, REPO_WATCH_TYPES, ZERO_TIME, + byte_length, valid_pattern, valid_repo_name, ) -from nerve.channels.hosted.contract.wire import format_time +from nerve.channels.hosted.contract.wire import dumps, format_time from nerve.channels.hosted.intake import Disposition +from nerve.config import LINEAR_EVENTS if TYPE_CHECKING: - from nerve.config import GitHubReposSyncConfig, GitHubSyncConfig, NerveConfig + from nerve.config import GitHubReposSyncConfig, GitHubSyncConfig, LinearSyncConfig, NerveConfig from nerve.db import Database logger = logging.getLogger(__name__) @@ -61,6 +76,10 @@ GITHUB = "github" # The repository watch: the repositories that sync.github_repos names. GITHUB_REPOS = REPO_WATCH_PROVIDER +# Linear issue events of the agent's Linear connector. +LINEAR = "linear" +# The record schemas of Linear records that Nerve reads. +LINEAR_SCHEMAS = ("linear.issue_event/1",) # Row target of a provider's buffer. The gateway keeps at most this many # source records for an agent, and the runner drains the buffer after every @@ -72,9 +91,14 @@ def hosted_sources(config: NerveConfig) -> list[str]: """The source providers whose records come from the gateway. GitHub is hosted as a whole: ``sync.github.mode: hosted`` makes the - repository watch (``github_repos``) hosted too. + repository watch (``github_repos``) hosted too, switched on or not. + Linear is hosted only (``sync.linear.mode``), and is a hosted source + while ``sync.linear.enabled`` is on when the process starts. """ - return [GITHUB, GITHUB_REPOS] if config.sync.github.mode == "hosted" else [] + providers = [GITHUB, GITHUB_REPOS] if config.sync.github.mode == "hosted" else [] + if config.sync.linear.enabled: + providers.append(LINEAR) + return providers # ---------------------------------------------------------------------- # @@ -99,9 +123,24 @@ def subscriptions_for(config: NerveConfig, providers: Iterable[str]) -> Subscrip watch = github_repos_subscription(config.sync.github_repos) if watch is not None: sources.append(watch) + if LINEAR in providers and config.sync.linear.enabled: + sources.append(linear_subscription(config.sync.linear)) return Subscriptions(version=SUBSCRIPTIONS_VERSION, sources=tuple(sources)) +def subscriptions_frame_bytes(body: Subscriptions) -> int: + """The UTF-8 bytes of the ``subscriptions`` frame that carries *body*. + + The request ID of a stream's frame (``o-``) can be a few bytes longer + than the one used here. + """ + envelope = Envelope( + version=PROTOCOL_VERSION, kind="subscriptions", request_id="o-1", + payload=Payload(subscriptions=body), + ) + return byte_length(dumps(envelope)) + + def chat_subscription(provider: str) -> SourceSubscription: """The subscription of a hosted chat provider: its name with empty filters. @@ -191,6 +230,36 @@ def github_repos_subscription(settings: GitHubReposSyncConfig) -> SourceSubscrip ) +def linear_subscription(settings: LinearSyncConfig) -> SourceSubscription: + """The Linear subscription: provider settings, record schemas, and rules. + + The settings carry only the values that differ from the provider's + defaults (every event, the inbox on, bot events off, own actions + dropped), so that absent settings mean the defaults and an event kind + that a later gateway adds reaches an agent that keeps the default. The + config refuses to load rules that the gateway would refuse, so they go + as they are, and the inbox guardrail applies them again. + """ + provider_settings: dict[str, Any] = {} + if set(settings.events) != set(LINEAR_EVENTS): + provider_settings["events"] = [event for event in LINEAR_EVENTS if event in settings.events] + if not settings.inbox: + provider_settings["inbox"] = False + if settings.include_bots: + provider_settings["include_bots"] = True + if not settings.drop_own_actions: + provider_settings["drop_own_actions"] = False + return SourceSubscription( + provider=LINEAR, + settings=provider_settings or None, + schemas=LINEAR_SCHEMAS, + rules=tuple( + SourceRule(path=rule.path, allow=tuple(rule.allow), deny=tuple(rule.deny)) + for rule in settings.rules + ), + ) + + def _patterns(values: object) -> list[str]: """The non-blank patterns of a setting, stripped as the inbox filter strips them.""" if not isinstance(values, (list, tuple)): @@ -234,25 +303,30 @@ def buffered_record(event: Event) -> dict[str, Any]: ``record.metadata`` keeps the wire entries (``key`` with ``value`` or ``values``); the source that drains the buffer knows which keys hold - lists and which hold flags. + lists and which hold flags. A record of a generic source adds its + ``schema`` and its ``data``, with every member. """ record = event.record timestamp = record.timestamp if record.timestamp != ZERO_TIME else event.occurred_at + fields: dict[str, Any] = { + "id": record.id, + "record_type": record.record_type, + "summary": record.summary, + "content": record.content, + "timestamp": format_time(timestamp) if timestamp != ZERO_TIME else "", + "metadata": [ + {"key": entry.key, "value": entry.value, "values": list(entry.values)} + for entry in record.metadata + ], + } + if record.data is not None: + fields["schema"] = record.schema + fields["data"] = record.data return { "event_id": event.event_id, "connection_id": str(event.connection_id), "conversation_id": event.conversation.id, - "record": { - "id": record.id, - "record_type": record.record_type, - "summary": record.summary, - "content": record.content, - "timestamp": format_time(timestamp) if timestamp != ZERO_TIME else "", - "metadata": [ - {"key": entry.key, "value": entry.value, "values": list(entry.values)} - for entry in record.metadata - ], - }, + "record": fields, } @@ -349,11 +423,15 @@ async def stop(self) -> None: "BUFFER_MAX_ROWS", "GITHUB", "GITHUB_REPOS", + "LINEAR", + "LINEAR_SCHEMAS", "HostedSourceConsumer", "buffered_record", "chat_subscription", "github_repos_subscription", "github_subscription", "hosted_sources", + "linear_subscription", "subscriptions_for", + "subscriptions_frame_bytes", ] diff --git a/nerve/channels/hosted/stream.py b/nerve/channels/hosted/stream.py index 5fd310ecc..d361cbab5 100644 --- a/nerve/channels/hosted/stream.py +++ b/nerve/channels/hosted/stream.py @@ -645,9 +645,17 @@ async def publish_subscriptions(self, *, bounded: bool = True) -> bool: def _subscriptions_frame(self, request_id: str) -> str | None: """The ``subscriptions`` frame from the current settings, encoded. + Subscriptions that the gateway would refuse, or whose frame exceeds + its frame limit, go narrower in two steps. First without the filter + patterns of the GitHub sources, which the inbox guardrail still + applies. Then also without each subscription that carries rules (a + generic source, such as Linear): it is never sent without them, + because the gateway would then store the records that they drop, so + it leaves the frame and the gateway stores nothing for it. + ``None`` when there is nothing to send: no hosted sources, settings that cannot be read, or subscriptions that the gateway would refuse - even without their filter patterns. + even so. """ try: body = self._subscriptions() @@ -656,37 +664,42 @@ def _subscriptions_frame(self, request_id: str) -> str | None: return None if body is None: return None - try: - body.validate() - except Rejected as error: - wide = body.without_filters() - try: - wide.validate() - except Rejected as still: - logger.warning( - "Channel stream %d: source subscriptions not sent: the gateway would " - "refuse them (%s)", self.id, still.detail, - ) - return None + text, problem = self._encoded_subscriptions(body, request_id) + if text is not None: + return text + wide = body.without_filters() + text, still = self._encoded_subscriptions(wide, request_id) + if text is not None: logger.warning( - "Channel stream %d: source subscriptions exceed the gateway's bounds (%s); " - "sending them without filter patterns", self.id, error.detail, + "Channel stream %d: source subscriptions exceed the gateway's bounds or frame " + "limit (%s); sending them without filter patterns", self.id, problem, ) - body = wide - try: - try: - return self._encode("subscriptions", body, None, request_id) - except Rejected as error: - if error.reason != RejectionReason.FRAME_TOO_LARGE: - raise - logger.warning( - "Channel stream %d: source subscriptions exceed the gateway's frame " - "limit; sending them without filter patterns", self.id, + return text + ruled = wide.ruled_providers() + if ruled: + text, still = self._encoded_subscriptions(wide.without_providers(ruled), request_id) + if text is not None: + logger.error( + "Channel stream %d: the subscriptions of %s are not sent: with their rules " + "the subscriptions exceed the gateway's bounds or frame limit (%s), and a " + "source with rules is never sent without them. The gateway stores nothing " + "for these providers until %s are shorter; nerve doctor checks them", + self.id, ", ".join(ruled), problem, ", ".join(f"sync.{p}.rules" for p in ruled), ) - return self._encode("subscriptions", body.without_filters(), None, request_id) + return text + logger.warning( + "Channel stream %d: source subscriptions not sent: the gateway would refuse them " + "(%s)", self.id, still, + ) + return None + + def _encoded_subscriptions(self, body: Subscriptions, request_id: str) -> tuple[str | None, str]: + """The frame of *body*, or ``None`` and why the gateway would refuse it.""" + try: + body.validate() + return self._encode("subscriptions", body, None, request_id), "" except Rejected as error: - logger.warning("Channel stream %d: source subscriptions not sent: %s", self.id, error) - return None + return None, error.detail def _publish_done(self, task: asyncio.Task) -> None: """Log a failure of the publish that negotiation started, when it happens.""" diff --git a/nerve/cli.py b/nerve/cli.py index e6a30c2a3..c28933736 100644 --- a/nerve/cli.py +++ b/nerve/cli.py @@ -1771,6 +1771,61 @@ def _github_repos_lines(config, warnings: list[str]) -> list[str]: return [line] +def _linear_lines(config, errors: list[str], warnings: list[str]) -> list[str]: + """Doctor lines for the Linear source (``sync.linear``). + + Linear records come only from the channel gateway, so the source needs + usable ``channels.hosted`` settings and a key file that can be read. A + switched-off hosted Slack holds the records back too, as for hosted + GitHub. + """ + linear = config.sync.linear + if not linear.enabled: + return ["[--] Linear source off (sync.linear.enabled is false)"] + from nerve.channels.hosted.auth import KeyFileError, read_key_file + from nerve.config import LINEAR_EVENTS + + hosted = config.channels.hosted + problems = hosted.problems() + if not problems: + try: + read_key_file(hosted.gateway_jwks_file) + except KeyFileError as error: + problems = [f"channels.hosted.gateway_jwks_file: {error}"] + if problems: + errors.extend(f"[ERR] Linear is on but {problem}" for problem in problems) + return [] + # The stream leaves a subscription with rules out of the frame when the + # rules do not fit, rather than send it without them. + from nerve.channels.hosted.contract import MAX_FRAME_BYTES + from nerve.channels.hosted.sources import hosted_sources, subscriptions_for, subscriptions_frame_bytes + + frame = subscriptions_frame_bytes(subscriptions_for(config, hosted_sources(config)).without_filters()) + if frame > MAX_FRAME_BYTES: + errors.append( + f"[ERR] Linear: with sync.linear.rules the subscriptions frame has about {frame} " + f"bytes, above the {MAX_FRAME_BYTES} that the channel gateway takes, so Nerve does " + "not send the Linear subscription and the gateway stores no Linear records. " + "Shorten sync.linear.rules" + ) + return [] + if config.slack.mode == "hosted" and not config.slack.enabled: + warnings.append( + "[WARN] Slack is hosted but switched off (slack.enabled is false): Nerve " + "reads nothing from the channel gateway until it is switched on, so the " + "Linear records wait there too" + ) + events = "all" if len(linear.events) == len(LINEAR_EVENTS) else ", ".join(linear.events) + count = len(linear.rules) + return [ + "[OK] Linear records come from the channel gateway (the agent's Linear " + f"connector): events {events}; inbox {'on' if linear.inbox else 'off'}; bots " + f"{'included' if linear.include_bots else 'left out'}; own actions " + f"{'dropped' if linear.drop_own_actions else 'kept'}; " + f"{count} {'rule' if count == 1 else 'rules'}" + ] + + def _source_suffix(config, path: str) -> str: """`` (environment)`` / `` (config-repo)`` after a configured connection line when the value did not come from a file on this machine; ``""`` for @@ -2262,6 +2317,7 @@ def doctor_report(config, config_source: str = "", check_api: bool = False) -> s lines.append("[--] Slack disabled") lines.extend(_hosted_github_lines(config, errors, warnings)) lines.extend(_github_repos_lines(config, warnings)) + lines.extend(_linear_lines(config, errors, warnings)) lines.extend(_email_lines(config, errors, warnings)) # Check SSL diff --git a/nerve/config.py b/nerve/config.py index aa5f81b17..6b442fbea 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -10,6 +10,7 @@ import logging import os import re +import unicodedata import uuid from dataclasses import dataclass, field, replace from pathlib import Path @@ -23,6 +24,7 @@ from nerve.coerce import coerced as _coerced from nerve.coerce import lenient_int as _lenient_int from nerve.inference_principal import direct_client_headers +from nerve.sources.source_json import pointer_tokens import yaml @@ -2300,6 +2302,194 @@ def from_dict(cls, d: dict) -> GitHubReposSyncConfig: ) +# How Linear issue events reach the inbox: the channel gateway polls Linear +# with the agent's Linear connector and streams the records here. "poll", a +# Linear source that this process runs, does not exist. +LINEAR_SYNC_MODES: tuple[str, ...] = ("hosted",) +LINEAR_POLL_MODE = "poll" +# The Linear issue events that the channel gateway reports +# (LinearSyncConfig.events): the names of the records' data.event. +LINEAR_EVENTS: tuple[str, ...] = ( + "issue_created", "assigned", "unassigned", "delegated", "state_changed", + "priority_changed", "moved_in", "commented", "mentioned", +) +# The rules of a generic hosted source (LinearSyncConfig.rules): at most this +# many rules, each with at most this many patterns of at most this many +# UTF-8 bytes. The channel gateway's bounds too. +SOURCE_RULES_MAX = 16 +SOURCE_RULE_PATTERNS_MAX = 64 +SOURCE_RULE_PATTERN_MAX_BYTES = 256 +_SOURCE_RULE_KEYS = frozenset({"path", "allow", "deny"}) + + +@dataclass +class SourceRuleConfig: + """One rule of a generic hosted source (``sync.linear.rules``). + + ``path`` is a JSON Pointer into the record's data, such as + ``/actor/kind``, and exactly one of ``allow`` and ``deny`` holds glob + patterns; the other is empty. :func:`_source_rules` builds them from the + config and refuses a rule that the channel gateway would refuse, so this + has no ``from_dict`` of its own. + """ + path: str = "" + allow: list[str] = field(default_factory=list) + deny: list[str] = field(default_factory=list) + + +def _linear_mode(value: object) -> str: + """``sync.linear.mode``: ``hosted``, also when unset; anything else refuses to load.""" + mode = _setting_str(value, "hosted").lower() + if mode == LINEAR_POLL_MODE: + raise ValueError( + "sync.linear.mode 'poll' is not available: Linear records come only from the " + "hosted channel gateway, so the mode is 'hosted'" + ) + if mode not in LINEAR_SYNC_MODES: + raise ValueError(f"sync.linear.mode must be one of {LINEAR_SYNC_MODES}, got {mode!r}") + return mode + + +def _linear_events(value: object) -> list[str]: + """``sync.linear.events``, checked against :data:`LINEAR_EVENTS`. + + Unset (missing, null or blank) is every event. Case does not matter, an + event named twice counts once, and they are kept in the order of + :data:`LINEAR_EVENTS`. An unknown event, or an empty list, refuses to + load: switch the source off with ``enabled: false`` instead. + """ + if value is None or (isinstance(value, str) and not value.strip()): + return list(LINEAR_EVENTS) + chosen: set[str] = set() + for entry in value if isinstance(value, (list, tuple)) else [value]: + name = str(entry or "").strip().lower() + if name not in LINEAR_EVENTS: + raise ValueError(f"sync.linear.events must be one or more of {LINEAR_EVENTS}, got {name!r}") + chosen.add(name) + if not chosen: + raise ValueError(f"sync.linear.events must be one or more of {LINEAR_EVENTS}, got none") + return [event for event in LINEAR_EVENTS if event in chosen] + + +def _source_rules(value: object, setting: str) -> list[SourceRuleConfig]: + """The rules of a generic hosted source, such as ``sync.linear.rules``. + + Each rule is a mapping with a ``path``, a JSON Pointer into the record's + data such as ``/actor/kind``, and exactly one non-empty ``allow`` or + ``deny`` list of glob patterns. Patterns are stripped and blank ones + dropped; a single value is a list of one. A rule that the channel gateway + would refuse refuses to load, so the gateway never closes the stream over + one. + """ + if value is None: + return [] + if not isinstance(value, list): + raise ValueError(f"{setting} must be a list of rules, each with a path and an allow or deny list") + if len(value) > SOURCE_RULES_MAX: + raise ValueError(f"{setting} must have at most {SOURCE_RULES_MAX} rules, got {len(value)}") + rules: list[SourceRuleConfig] = [] + for index, entry in enumerate(value): + where = f"{setting}[{index}]" + if not isinstance(entry, dict): + raise ValueError(f"{where} must be a mapping with a path and an allow or deny list") + unknown = sorted(str(key) for key in set(entry) - _SOURCE_RULE_KEYS) + if unknown: + raise ValueError(f"{where} has unknown keys {unknown}; a rule has path, and allow or deny") + path = entry.get("path") + if not isinstance(path, str) or pointer_tokens(path) is None: + raise ValueError( + f"{where}.path must be a JSON Pointer into the record data, such as /actor/kind " + "(it starts with '/', has 1 to 8 parts and at most 256 bytes without control " + f"characters, and writes '~' as '~0' and '/' as '~1'), got {path!r}" + ) + lists: dict[str, list[str]] = {} + for key in ("allow", "deny"): + patterns = _source_rule_patterns(entry.get(key), f"{where}.{key}") + if patterns: + lists[key] = patterns + if len(lists) != 1: + raise ValueError(f"{where} needs exactly one non-empty list: allow or deny") + rules.append(SourceRuleConfig(path=path, **lists)) + return rules + + +def _source_rule_patterns(value: object, where: str) -> list[str]: + """The patterns of one rule list, stripped, without the blank ones (none when unset).""" + if value is None: + return [] + entries = value if isinstance(value, (list, tuple)) else [value] + patterns = [text for text in (str(entry).strip() for entry in entries if entry is not None) if text] + if len(patterns) > SOURCE_RULE_PATTERNS_MAX: + raise ValueError(f"{where} must have at most {SOURCE_RULE_PATTERNS_MAX} patterns, got {len(patterns)}") + for pattern in patterns: + if ( + len(pattern.encode("utf-8", "replace")) > SOURCE_RULE_PATTERN_MAX_BYTES + or any(unicodedata.category(character) in ("Cc", "Cs") for character in pattern) + ): + raise ValueError( + f"{where} patterns must have at most {SOURCE_RULE_PATTERN_MAX_BYTES} bytes and no " + f"control characters, got {pattern[:40]!r}" + ) + return patterns + + +@dataclass +class LinearSyncConfig: + """Config for the Linear source: issue events of the agent's Linear connector. + + The channel gateway polls Linear for the agent with the credential of the + agent's Linear connector in the MCP catalog, and only within the + connector's project scope, then streams the records to Nerve (``mode: + hosted``, the only mode). Each record names the catalog server, whose + Linear tools act on it. The settings below go to the gateway as the + provider settings of the subscription; the rules go too, and the inbox + guardrail applies them again. All of it follows a reload, but switching + the source on needs a restart. See docs/sources.md "Linear". + """ + enabled: bool = False + # "hosted" only, also when unset. "poll" and any other value refuse to + # load. Restart-only. + mode: str = "hosted" + # The events to report (LINEAR_EVENTS); every one by default. An unknown + # event or an empty list refuses to load. + events: list[str] = field(default_factory=lambda: list(LINEAR_EVENTS)) + # Also report the mentions of the credential's Linear user, which the + # gateway reads from its Linear inbox (event mentioned, via inbox). Only a + # credential of the agent's own (an agent API key, the agent's sign-in) + # has an inbox; a credential of the tenant ignores it. Assignments, + # comments and the other changes come from the issues of the scope. + inbox: bool = True + # Also report the events of bots and integrations. + include_bots: bool = False + # Drop the events of the credential's own Linear user. A tenant API key + # belongs to a person, so its events are kept and marked + # (actor.credential_user) whatever this says. + drop_own_actions: bool = True + # Allow/deny glob lists on JSON Pointer paths into the record data, such + # as {"path": "/issue/team/key", "allow": ["ENG*"]}. At most + # SOURCE_RULES_MAX rules; a rule the gateway would refuse refuses to load. + rules: list[SourceRuleConfig] = field(default_factory=list) + schedule: str = "*/15 * * * *" + condense: bool = False + batch_size: int = 50 + + @classmethod + @_coerced + def from_dict(cls, d: dict) -> LinearSyncConfig: + return cls( + enabled=d.get("enabled", False), + mode=_linear_mode(d.get("mode")), + events=_linear_events(d.get("events")), + inbox=d.get("inbox", True), + include_bots=d.get("include_bots", False), + drop_own_actions=d.get("drop_own_actions", True), + rules=_source_rules(d.get("rules"), "sync.linear.rules"), + schedule=d.get("schedule", "*/15 * * * *"), + condense=d.get("condense", False), + batch_size=d.get("batch_size", 50), + ) + + @dataclass class CodexOriginConfig: """A single Codex thread sync origin. @@ -2402,6 +2592,7 @@ class SyncConfig: github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) github_events: GitHubEventsSyncConfig = field(default_factory=GitHubEventsSyncConfig) github_repos: GitHubReposSyncConfig = field(default_factory=GitHubReposSyncConfig) + linear: LinearSyncConfig = field(default_factory=LinearSyncConfig) codex: CodexSyncConfig = field(default_factory=CodexSyncConfig) message_ttl_days: int = 7 # How long to keep source messages in the inbox consumer_cursor_ttl_days: int = 2 # Consumer cursors expire after N days of inactivity @@ -2416,6 +2607,7 @@ def from_dict(cls, d: dict) -> SyncConfig: github=GitHubSyncConfig.from_dict(d.get("github", {})), github_events=GitHubEventsSyncConfig.from_dict(d.get("github_events", {})), github_repos=GitHubReposSyncConfig.from_dict(d.get("github_repos", {})), + linear=LinearSyncConfig.from_dict(d.get("linear", {})), codex=CodexSyncConfig.from_dict(d.get("codex", {})), message_ttl_days=d.get("message_ttl_days", 7), consumer_cursor_ttl_days=d.get("consumer_cursor_ttl_days", 2), diff --git a/nerve/sources/filters.py b/nerve/sources/filters.py index 4c2380526..a1bc9a9ab 100644 --- a/nerve/sources/filters.py +++ b/nerve/sources/filters.py @@ -29,6 +29,12 @@ (``ClickHouse/*``). List-valued metadata (e.g. Gmail ``labels``) matches if *any* element matches. Non-string scalars (e.g. Telegram ``chat_id``) are coerced to ``str`` before matching. + +A generic hosted source (Linear, for example) uses :class:`DataRule`: the +record's metadata holds the JSON data that the provider sent, and each rule +names a value in it with a JSON Pointer. The channel gateway applies the same +rules before it stores a record, so these semantics are the channel +contract's, and the two sides drop the same records. """ from __future__ import annotations @@ -36,21 +42,34 @@ import dataclasses import fnmatch import logging +from collections.abc import Iterable, Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any + +from nerve.sources.source_json import ( + SourceJSONError, + canonical_number, + pointer_tokens, + record_data, + resolve_pointer, +) if TYPE_CHECKING: from nerve.sources.models import SourceRecord logger = logging.getLogger(__name__) +# The names of a rule's two lists, as a log line names the one that fired. +ALLOW = "allow" +DENY = "deny" + def _norm(value: object) -> str: """Normalize a value for case-insensitive matching.""" return str(value).strip().lower() -def matches_any(value: object, patterns: list[str]) -> bool: +def matches_any(value: object, patterns: Iterable[str]) -> bool: """Case-insensitive shell-glob match of *value* against any pattern. The guardrail rules, for a source that applies them itself (the GitHub @@ -101,34 +120,114 @@ def _values(self, record: SourceRecord) -> list[str]: return [str(v) for v in raw if v is not None] return [str(raw)] + def fired_list(self, record: SourceRecord) -> str: + """``deny`` or ``allow``: the list that drops *record*, for a log line.""" + return DENY if self.deny and any(matches_any(v, self.deny) for v in self._values(record)) else ALLOW + def passes(self, record: SourceRecord) -> bool: """Return True if *record* satisfies this rule (should be kept).""" if record.record_type in self.exempt_record_types: return True - values = self._values(record) + return _passes(self._values(record), self.allow, self.deny) + + +def _passes(values: list[str], allow: Sequence[str], deny: Sequence[str]) -> bool: + """The decision of one rule on its values, with the semantics at the top of this module.""" + # Deny wins: drop on any deny match. + if deny and any(matches_any(v, deny) for v in values): + return False - # deny wins — drop on any deny match. - if self.deny and any(matches_any(v, self.deny) for v in values): + # Allow gate: when set, require a match. An absent field fails closed. + if allow: + if not values: return False + return any(matches_any(v, allow) for v in values) - # allow gate — when set, require a match. An absent field fails closed. - if self.allow: - if not values: - return False - return any(matches_any(v, self.allow) for v in values) + return True + + +@dataclass(frozen=True) +class DataRule: + """Allow/deny matching on a value in the data of a generic hosted record. + + ``path`` is a JSON Pointer (RFC 6901) into the record's data, the JSON + object that the provider sent, such as ``/actor/kind``. The record's + metadata holds it next to its schema (see + :func:`nerve.sources.source_json.record_data`); a record without data + has no value at any path. The values at the path are: + + * a string, as it is; + * a number, as its canonical JSON text: ``2`` for ``2.0``, ``1.5``; + * ``true``, ``false`` and ``null``, as those words; + * each element of an array that is one of the above. + + An object, an array or object in an array, an empty array, and a path + that names nothing in the data give no value. ``allow`` and ``deny`` + match those values as :class:`FieldRule` matches: a deny match drops the + record, and a non-empty allow list drops a record without a match, also + one without a value. A path that is not a valid JSON Pointer names + nothing. + """ - return True + path: str + allow: tuple[str, ...] = () + deny: tuple[str, ...] = () + + @property + def field(self) -> str: + """The path, as the runner's drop log and the drop hook name a rule.""" + return self.path + + @property + def active(self) -> bool: + return bool(self.allow or self.deny) + + def _resolved(self, record: SourceRecord) -> tuple[bool, Any]: + tokens = pointer_tokens(self.path) + stored = record_data(record.metadata) + if tokens is None or stored is None: + return False, None + return resolve_pointer(stored[1], tokens) + + def values(self, record: SourceRecord) -> list[str]: + """The values that the rule judges in *record*.""" + found, value = self._resolved(record) + if not found: + return [] + values: list[str] = [] + for element in value if isinstance(value, (list, tuple)) else [value]: + if isinstance(element, str): + values.append(element) + elif element is None: + values.append("null") + elif isinstance(element, bool): + values.append("true" if element else "false") + elif isinstance(element, (int, float)): + try: + values.append(canonical_number(element)) + except SourceJSONError: # a number that no JSON data can hold + continue + return values + + def fired_list(self, record: SourceRecord) -> str: + """``deny`` or ``allow``: the list of the rule, which is the one that drops *record*.""" + return DENY if self.deny else ALLOW + + def passes(self, record: SourceRecord) -> bool: + """Return True if *record* satisfies this rule (should be kept).""" + return _passes(self.values(record), self.allow, self.deny) @dataclass class InboxFilter: - """A set of :class:`FieldRule` applied to source records (logical AND). + """A set of rules applied to source records (logical AND). - An inactive filter (no rules, or all rules empty) is a pass-through — + Each rule is a :class:`FieldRule` or a :class:`DataRule`. An inactive + filter (no rules, or all rules empty) is a pass-through: :meth:`partition` returns every record as kept with nothing dropped. """ - rules: list[FieldRule] = dataclasses.field(default_factory=list) + rules: list[FieldRule | DataRule] = dataclasses.field(default_factory=list) @property def active(self) -> bool: @@ -139,7 +238,7 @@ def passes(self, record: SourceRecord) -> bool: """Return True if *record* passes all rules (should be kept).""" return all(r.passes(record) for r in self.rules) - def rejects(self, record: SourceRecord) -> FieldRule | None: + def rejects(self, record: SourceRecord) -> FieldRule | DataRule | None: """The first rule that drops *record*, or None if it passes. Dropped records are never persisted, so this is the only way to tell diff --git a/nerve/sources/github_hosted.py b/nerve/sources/github_hosted.py index 178cef892..61d7f1d33 100644 --- a/nerve/sources/github_hosted.py +++ b/nerve/sources/github_hosted.py @@ -31,17 +31,14 @@ 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 +from nerve.sources.hosted import HostedSource, record_fields +from nerve.sources.models import SourceRecord if TYPE_CHECKING: from nerve.db import Database -logger = logging.getLogger(__name__) - SOURCE_NAME = "github" # The buffer channel that the hosted source consumer writes for this source. BUFFER_CHANNEL = "github" @@ -53,7 +50,7 @@ _FLAG_KEYS = frozenset({"unread"}) -class HostedGitHubSource(Source): +class HostedGitHubSource(HostedSource): """A GitHub source over the hosted records buffer of its provider. *provider* is ``github`` (the default) or ``github_repos``: the buffer @@ -61,77 +58,14 @@ class HostedGitHubSource(Source): """ source_name = SOURCE_NAME - push_fed = True def __init__(self, db: Database, provider: str = SOURCE_NAME): if provider not in PROVIDERS: raise ValueError(f"no hosted GitHub source for provider {provider!r}") - self._db = db - self.source_name = provider - self._buffer = provider - - async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: - """Read buffered records past *cursor*, oldest first. - - A failed read raises, so the runner reports it and its health checks - count it. - """ - rows = await self._db.read_channel_observations( - self._buffer, after_id=self._position(cursor), limit=limit, - ) - if not rows: - return FetchResult(records=[], next_cursor=cursor, has_more=False) - - records: list[SourceRecord] = [] - for row_id, payload in rows: - record = ( - record_from_buffer(payload, source=self.source_name) - if isinstance(payload, dict) else None - ) - if record is None: - logger.warning( - "Hosted %s source: skipping unreadable buffer row %s", self.source_name, row_id, - ) - continue - records.append(record) - # The cursor tracks what was scanned, so an unreadable row is passed - # and does not stop the drain. - return FetchResult( - records=records, - next_cursor=str(rows[-1][0]), - has_more=len(rows) >= limit, - ) - - async def has_undrained(self, cursor: str | None) -> bool: - """Whether a record waits in the buffer past *cursor*. - - Normally none does: the cron service drains the buffer as records are - delivered. One row is enough to know; a failed read answers no, so a - forecast never fails for it. - """ - try: - rows = await self._db.read_channel_observations( - self._buffer, after_id=self._position(cursor, quiet=True), limit=1, - ) - except Exception: - logger.debug("Hosted %s source: checking the buffer failed", self.source_name, exc_info=True) - return False - return bool(rows) - - def _position(self, cursor: str | None, *, quiet: bool = False) -> int: - """The buffer row ID that *cursor* names, or 0.""" - if not cursor: - return 0 - try: - return int(cursor) - except ValueError: - if not quiet: - logger.info( - "Hosted %s source: cursor %.80r is not a buffer position (a cursor of " - "the polling source); draining from the start of the buffer", - self.source_name, cursor, - ) - return 0 + super().__init__(db, provider) + + def record(self, payload: dict[str, Any]) -> SourceRecord | None: + return record_from_buffer(payload, source=self.source_name) def record_from_buffer(payload: dict[str, Any], source: str = SOURCE_NAME) -> SourceRecord | None: @@ -139,21 +73,16 @@ def record_from_buffer(payload: dict[str, Any], source: str = SOURCE_NAME) -> So *source* is the provider whose buffer held it, and the record's source. """ - record = payload.get("record") - if not isinstance(record, dict): - return None - record_id = record.get("id") - record_type = record.get("record_type") - summary = record.get("summary") - if not all(isinstance(value, str) and value for value in (record_id, record_type, summary)): + record = record_fields(payload) + if record is None: return None content = record.get("content") timestamp = record.get("timestamp") return SourceRecord( - id=record_id, + id=record["id"], source=source, - record_type=record_type, - summary=summary, + record_type=record["record_type"], + summary=record["summary"], content=content if isinstance(content, str) else "", timestamp=timestamp if isinstance(timestamp, str) else "", metadata=_metadata(record.get("metadata")), diff --git a/nerve/sources/hosted.py b/nerve/sources/hosted.py new file mode 100644 index 000000000..d0f0a238e --- /dev/null +++ b/nerve/sources/hosted.py @@ -0,0 +1,160 @@ +"""Hosted sources: drain the records that the channel gateway delivered. + +In hosted mode the channel gateway polls a provider for the agent and +streams the records to Nerve. The hosted source consumer writes every record +to the ``channel_observations`` buffer under the provider's name (see +:mod:`nerve.channels.hosted.sources`), and a :class:`HostedSource` turns the +buffered payloads back into :class:`SourceRecord` objects. It makes no +provider calls. + +The source keeps the provider's name, so it runs as ``source:`` +with the guardrail, condensing, health checks, consumer cursors and +``messages`` gates of every source. The cursor is the buffer's row ID, as for +the chat channel drains. + +:class:`HostedSource` drains a generic source, such as Linear: the record's +metadata holds its record ``schema`` and its ``data``, the JSON object that +the provider defines, with every member. GitHub has a closed vocabulary of its +own (:class:`~nerve.sources.github_hosted.HostedGitHubSource`). + +Buffered payload of a generic record (one row per record):: + + {"event_id": ..., "connection_id": ..., "conversation_id": ..., + "record": {"id", "record_type", "schema", "summary", "content", + "timestamp": RFC 3339, "metadata": [], "data": {...}}} +""" + +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 +from nerve.sources.source_json import record_metadata + +if TYPE_CHECKING: + from nerve.db import Database + +logger = logging.getLogger(__name__) + + +class HostedSource(Source): + """A source over the hosted records buffer of one provider. + + *provider* is the buffer channel that the hosted source consumer writes, + and the source name. + """ + + push_fed = True + + def __init__(self, db: Database, provider: str): + self._db = db + self.source_name = provider + self._buffer = provider + + def record(self, payload: dict[str, Any]) -> SourceRecord | None: + """The inbox record of one buffered payload, or ``None`` when it is unusable.""" + return record_from_buffer(payload, source=self.source_name) + + async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: + """Read buffered records past *cursor*, oldest first. + + A failed read raises, so the runner reports it and its health checks + count it. + """ + rows = await self._db.read_channel_observations( + self._buffer, after_id=self._position(cursor), limit=limit, + ) + if not rows: + return FetchResult(records=[], next_cursor=cursor, has_more=False) + + records: list[SourceRecord] = [] + for row_id, payload in rows: + record = self.record(payload) if isinstance(payload, dict) else None + if record is None: + logger.warning( + "Hosted %s source: skipping unreadable buffer row %s", self.source_name, row_id, + ) + continue + records.append(record) + # The cursor tracks what was scanned, so an unreadable row is passed + # and does not stop the drain. + return FetchResult( + records=records, + next_cursor=str(rows[-1][0]), + has_more=len(rows) >= limit, + ) + + async def has_undrained(self, cursor: str | None) -> bool: + """Whether a record waits in the buffer past *cursor*. + + Normally none does: the cron service drains the buffer as records are + delivered. One row is enough to know; a failed read answers no, so a + forecast never fails for it. + """ + try: + rows = await self._db.read_channel_observations( + self._buffer, after_id=self._position(cursor, quiet=True), limit=1, + ) + except Exception: + logger.debug("Hosted %s source: checking the buffer failed", self.source_name, exc_info=True) + return False + return bool(rows) + + def _position(self, cursor: str | None, *, quiet: bool = False) -> int: + """The buffer row ID that *cursor* names, or 0.""" + if not cursor: + return 0 + try: + return int(cursor) + except ValueError: + if not quiet: + logger.info( + "Hosted %s source: cursor %.80r is not a buffer position (a cursor of " + "the polling source); draining from the start of the buffer", + self.source_name, cursor, + ) + return 0 + + +def record_fields(payload: dict[str, Any]) -> dict[str, Any] | None: + """The record of a buffered payload, or ``None`` without its ID, type and summary.""" + record = payload.get("record") + if not isinstance(record, dict): + return None + if not all(isinstance(record.get(key), str) and record[key] for key in ("id", "record_type", "summary")): + return None + return record + + +def record_from_buffer(payload: dict[str, Any], source: str) -> SourceRecord | None: + """The inbox record of one buffered generic payload, or ``None`` when it is unusable. + + *source* is the provider whose buffer held it, and the record's source. + The record's metadata holds its ``schema`` and its ``data``, with every + member (:func:`~nerve.sources.source_json.record_metadata`); a record + without data has no metadata. + """ + record = record_fields(payload) + if record is None: + return None + content = record.get("content") + timestamp = record.get("timestamp") + data = record.get("data") + schema = record.get("schema") + return SourceRecord( + id=record["id"], + source=source, + record_type=record["record_type"], + summary=record["summary"], + content=content if isinstance(content, str) else "", + timestamp=timestamp if isinstance(timestamp, str) else "", + metadata=( + record_metadata(schema if isinstance(schema, str) else "", dict(data)) + if isinstance(data, dict) else {} + ), + ) + + +__all__ = ["HostedSource", "record_fields", "record_from_buffer"] diff --git a/nerve/sources/registry.py b/nerve/sources/registry.py index 55275e9d4..521993ac6 100644 --- a/nerve/sources/registry.py +++ b/nerve/sources/registry.py @@ -277,6 +277,36 @@ def build_source_runners( gh_repos.deny_authors or [], ) + # Linear: hosted only. The channel gateway polls Linear with the agent's + # Linear connector and streams the records; the runner drains them with + # the rules of sync.linear as the guardrail, which the gateway applied + # before it stored them. + linear = config.sync.linear + if linear.enabled: + from nerve.channels.hosted.sources import LINEAR + from nerve.sources.filters import DataRule, InboxFilter + from nerve.sources.hosted import HostedSource + + runners.append(SourceRunner( + source=HostedSource(db, LINEAR), + db=db, + batch_size=linear.batch_size, + condense=linear.condense, + condense_model=condense_model, + condense_client_factory=condense_factory, + ttl_days=ttl_days, + inbox_filter=InboxFilter(rules=[ + DataRule(rule.path, allow=tuple(rule.allow), deny=tuple(rule.deny)) + for rule in linear.rules + ]), + )) + logger.info( + "Registered source: linear (hosted) (batch=%d, events=%s, inbox=%s, " + "include_bots=%s, drop_own_actions=%s, rules=%d)", + linear.batch_size, linear.events, linear.inbox, linear.include_bots, + linear.drop_own_actions, len(linear.rules), + ) + # 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 diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index 39cd338ea..a5336cd31 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -402,20 +402,19 @@ async def _ingest_batch(self, cursor: str | None) -> _BatchResult: kept, dropped = self.inbox_filter.partition(records) if dropped: logger.info( - "Source %s: guardrail dropped %d/%d records (e.g. %r)", + "Source %s: guardrail dropped %d/%d records", self.source.source_name, len(dropped), len(records), - dropped[0].summary, ) - # Name the rule that fired — the summary alone can't tell a - # repo drop from an actor or CI-branch drop, and the record - # itself is gone after this point. Capped to keep logs sane. + # Name the rule that fired, and only the rule: the record is + # gone after this point, and what the guardrail kept from the + # inbox must not reach the log either (no summary, no value). + # Capped to keep logs sane. for record in dropped[:_MAX_DROP_LOG_LINES]: rule = self.inbox_filter.rejects(record) - field = rule.field if rule else "?" logger.info( - "Source %s: dropped %s on %s=%r — %s", - self.source.source_name, record.id, field, - (record.metadata or {}).get(field), record.summary, + "Source %s: dropped %s by the %s list of the rule on %s", + self.source.source_name, record.id, + rule.fired_list(record) if rule else "?", rule.field if rule else "?", ) records = kept dropped_count = len(dropped) diff --git a/nerve/sources/source_json.py b/nerve/sources/source_json.py new file mode 100644 index 000000000..574535da0 --- /dev/null +++ b/nerve/sources/source_json.py @@ -0,0 +1,281 @@ +"""The JSON values of generic hosted sources. + +A generic source of the hosted channel gateway, such as Linear, sends the data +of each record as a JSON object that the provider defines, and Nerve sends +the settings of a subscription to it as a JSON object too. The channel +contract calls both a *source JSON object*: a JSON object without a repeated +member name, of valid Unicode (no unpaired surrogate), with at most +:data:`MAX_SOURCE_JSON_DEPTH` levels of objects and arrays (the object itself +is the first level), and with numbers that IEEE 754 binary64 can hold. Its +byte bounds count the UTF-8 bytes of its canonical form (RFC 8785, the JSON +Canonicalization Scheme), which does not depend on how a sender encodes it. + +The rules of a subscription keep or drop records by the value at a JSON +Pointer (RFC 6901) into their data. + +This module holds what the channel contract (which checks these values on +the wire) and the inbox guardrail (which applies the rules again) share. It +works on the values that :func:`json.loads` returns, and it imports nothing +from Nerve. +""" + +from __future__ import annotations + +import json +import math +import re +import unicodedata +from decimal import Decimal +from typing import Any + +# The metadata of a stored generic record holds its record schema and its +# data under these two members, so that the data keeps every member of its +# own and the schema stays known after the drain (see record_data). +RECORD_SCHEMA_KEY = "schema" +RECORD_DATA_KEY = "data" +# The levels of objects and arrays in a source JSON object, the object itself +# as the first. A JSON Pointer into one has at most this many tokens. +MAX_SOURCE_JSON_DEPTH = 8 +# The UTF-8 bytes of a JSON Pointer in a subscription rule. +MAX_POINTER_BYTES = 256 +# Plain decimal notation of a canonical number: from 1e-6 up to but not +# including 1e21 (ECMAScript Number.prototype.toString). +_PLAIN_EXPONENT_MIN = -6 +_PLAIN_EXPONENT_MAX = 21 +# A "~" that does not start the escape "~0" or "~1". +_BAD_ESCAPE = re.compile(r"~(?![01])") +# An array index in a JSON Pointer: decimal, without a leading zero. +_ARRAY_INDEX = re.compile(r"0|[1-9][0-9]*") + + +class SourceJSONError(ValueError): + """A value that breaks a rule of a source JSON object. + + ``limit`` is true when the value is above a bound (its depth or its + size), and false when it is not valid. + """ + + def __init__(self, detail: str, *, limit: bool = False) -> None: + super().__init__(detail) + self.limit = limit + + +def check_source_json_object(value: Any, maximum_bytes: int) -> None: + """Check that *value* is a source JSON object within *maximum_bytes*. + + The bound counts the UTF-8 bytes of the canonical form. Raises + :class:`SourceJSONError`. A repeated member name cannot be seen here, so + the parser refuses it (see ``nerve.channels.hosted.contract.wire``). + """ + if not isinstance(value, dict): + raise SourceJSONError("is not a JSON object") + _check(value, 1) + if len(canonical_json(value).encode("utf-8")) > maximum_bytes: + raise SourceJSONError("exceeds its byte limit", limit=True) + + +def _check(value: Any, depth: int) -> None: + if isinstance(value, (dict, list, tuple)): + if depth > MAX_SOURCE_JSON_DEPTH: + raise SourceJSONError("exceeds its depth limit", limit=True) + if isinstance(value, dict): + for name, member in value.items(): + if not isinstance(name, str): + raise SourceJSONError("has a member name that is not a string") + _check_text(name) + _check(member, depth + 1) + else: + for element in value: + _check(element, depth + 1) + elif isinstance(value, str): + _check_text(value) + elif isinstance(value, bool) or value is None: + pass + elif isinstance(value, (int, float)): + _binary64(value) + else: + raise SourceJSONError(f"holds a {type(value).__name__}, which is not a JSON value") + + +def _check_text(text: str) -> None: + try: + text.encode("utf-8") + except UnicodeEncodeError: + raise SourceJSONError("has a string with an unpaired surrogate") from None + + +def _binary64(number: int | float) -> float: + """*number* as binary64, or :class:`SourceJSONError` when it does not fit. + + A number below the smallest binary64 value is zero, as the parsers of + both sides read it. + """ + try: + value = float(number) + except OverflowError: + raise SourceJSONError("has a number outside the binary64 range") from None + if not math.isfinite(value): + raise SourceJSONError("has a number outside the binary64 range") + return value + + +def record_metadata(schema: str, data: dict[str, Any]) -> dict[str, Any]: + """The metadata of a stored generic record: its record schema and its data.""" + return {RECORD_SCHEMA_KEY: schema, RECORD_DATA_KEY: data} + + +def record_data(metadata: Any) -> tuple[str, dict[str, Any]] | None: + """The record schema and the data of a generic record's metadata. + + ``None`` for the metadata of any other record: it must hold exactly the + two members of :func:`record_metadata`, a string and a JSON object. + """ + if not isinstance(metadata, dict) or set(metadata) != {RECORD_SCHEMA_KEY, RECORD_DATA_KEY}: + return None + schema, data = metadata[RECORD_SCHEMA_KEY], metadata[RECORD_DATA_KEY] + if not isinstance(schema, str) or not isinstance(data, dict): + return None + return schema, data + + +# ---------------------------------------------------------------------- # +# Canonical form (RFC 8785) # +# ---------------------------------------------------------------------- # + + +def canonical_json(value: Any) -> str: + """The RFC 8785 canonical form of a value that passes the checks. + + Object members sorted by the UTF-16 code units of their names, no white + space, the shortest ECMAScript form of each number, and the minimal + escapes of each string: only the quotation mark, the reverse solidus and + the characters below U+0020 are escaped. + """ + parts: list[str] = [] + _write(parts, value) + return "".join(parts) + + +def _write(parts: list[str], value: Any) -> None: + if value is None: + parts.append("null") + elif value is True: + parts.append("true") + elif value is False: + parts.append("false") + elif isinstance(value, (int, float)): + parts.append(canonical_number(value)) + elif isinstance(value, str): + # The escapes of json.dumps without ensure_ascii are those of RFC 8785 + # section 3.2.2.2, with lowercase hexadecimal digits. + parts.append(json.dumps(value, ensure_ascii=False)) + elif isinstance(value, dict): + parts.append("{") + names = sorted(value, key=lambda name: name.encode("utf-16-be", "surrogatepass")) + for index, name in enumerate(names): + if index: + parts.append(",") + parts.append(json.dumps(name, ensure_ascii=False)) + parts.append(":") + _write(parts, value[name]) + parts.append("}") + else: + parts.append("[") + for index, element in enumerate(value): + if index: + parts.append(",") + _write(parts, element) + parts.append("]") + + +def canonical_number(number: int | float) -> str: + """The ECMAScript ``Number.prototype.toString`` form of a binary64 number. + + RFC 8785 section 3.2.2.3: the fewest digits that read back as the same + number, plain decimal notation from 1e-6 up to but not including 1e21, + exponent notation otherwise, and ``0`` for negative zero. An integer is + read as binary64 first, so ``12345678901234567890`` is + ``12345678901234567000`` and ``2.0`` is ``2``. + """ + value = _binary64(number) + if value == 0: + return "0" + sign = "-" if value < 0 else "" + # repr gives the shortest digits that read back as the same binary64. + _, digit_tuple, exponent = Decimal(repr(abs(value))).normalize().as_tuple() + digits = "".join(map(str, digit_tuple)) + count = len(digits) + point = int(exponent) + count # the decimal point is after this many digits + if count <= point <= _PLAIN_EXPONENT_MAX: + return sign + digits + "0" * (point - count) + if 0 < point <= _PLAIN_EXPONENT_MAX: + return sign + digits[:point] + "." + digits[point:] + if _PLAIN_EXPONENT_MIN < point <= 0: + return sign + "0." + "0" * -point + digits + power = point - 1 + mantissa = digits if count == 1 else f"{digits[0]}.{digits[1:]}" + return f"{sign}{mantissa}e{'+' if power > 0 else '-'}{abs(power)}" + + +# ---------------------------------------------------------------------- # +# JSON Pointer (RFC 6901) # +# ---------------------------------------------------------------------- # + + +def pointer_tokens(path: str) -> list[str] | None: + """The reference tokens of *path*, unescaped, or ``None`` when it is not valid. + + A valid path starts with ``/``, has 1 to :data:`MAX_SOURCE_JSON_DEPTH` + tokens and at most :data:`MAX_POINTER_BYTES` bytes of UTF-8 without + control characters, and writes ``~`` only in the escapes ``~0`` (``~``) + and ``~1`` (``/``). + """ + if not isinstance(path, str) or not path.startswith("/"): + return None + try: + size = len(path.encode("utf-8")) + except UnicodeEncodeError: + return None + if size > MAX_POINTER_BYTES or any(unicodedata.category(c) == "Cc" for c in path): + return None + tokens = path[1:].split("/") + if len(tokens) > MAX_SOURCE_JSON_DEPTH or any(_BAD_ESCAPE.search(t) for t in tokens): + return None + # RFC 6901 section 4: ~1 becomes / before ~0 becomes ~. + return [token.replace("~1", "/").replace("~0", "~") for token in tokens] + + +def resolve_pointer(value: Any, tokens: list[str]) -> tuple[bool, Any]: + """Whether the *tokens* name a value in *value*, and that value. + + A token names the member of an object, or the element of an array at the + index that it writes in decimal without a leading zero. + """ + for token in tokens: + if isinstance(value, dict): + if token not in value: + return False, None + value = value[token] + elif isinstance(value, (list, tuple)): + if _ARRAY_INDEX.fullmatch(token) is None or int(token) >= len(value): + return False, None + value = value[int(token)] + else: + return False, None + return True, value + + +__all__ = [ + "MAX_POINTER_BYTES", + "MAX_SOURCE_JSON_DEPTH", + "RECORD_DATA_KEY", + "RECORD_SCHEMA_KEY", + "SourceJSONError", + "canonical_json", + "canonical_number", + "check_source_json_object", + "pointer_tokens", + "record_data", + "record_metadata", + "resolve_pointer", +] diff --git a/nerve/templates/personal/TOOLS.md b/nerve/templates/personal/TOOLS.md index 06305e83f..3195a2f00 100644 --- a/nerve/templates/personal/TOOLS.md +++ b/nerve/templates/personal/TOOLS.md @@ -47,7 +47,8 @@ Call tools as `mcp__nerve__`. The system prompt lists what is live; this i - `send_file`: deliver a workspace file as an attachment or download card - `send_channel_message`: post to a Slack conversation you name; offered only with Slack enabled and `slack.allow_outbound: true`, limited to `slack.allow_channels` -**Sync sources (inbox)**: `telegram`, `gmail:`, `imap: