From 349be4f164b36400ef5fb0039f954586cfea8db0 Mon Sep 17 00:00:00 2001
From: Tee8z <tee8z@protonmail.com>
Date: Tue, 10 Nov 2026 13:52:10 -0500
Subject: [PATCH 1/2] feat(integrations): ingest verified provider activity
 into Buzz threads

Signed-off-by: Tee8z <tee8z@protonmail.com>
---
 .github/workflows/voltage-agent-images.yml    |  25 +-
 Cargo.lock                                    |  23 +
 Cargo.toml                                    |   1 +
 crates/buzz-core/src/integration.rs           | 210 ++++++
 crates/buzz-core/src/kind.rs                  |  14 +
 crates/buzz-core/src/lib.rs                   |   2 +
 crates/buzz-ingest/Cargo.toml                 |  29 +
 crates/buzz-ingest/README.md                  | 105 +++
 crates/buzz-ingest/src/config.rs              | 122 ++++
 crates/buzz-ingest/src/http.rs                | 286 ++++++++
 crates/buzz-ingest/src/main.rs                |  66 ++
 crates/buzz-ingest/src/normalize.rs           | 471 ++++++++++++
 crates/buzz-ingest/src/plan.rs                | 273 +++++++
 crates/buzz-ingest/src/relay.rs               | 111 +++
 crates/buzz-ingest/src/tests.rs               | 689 ++++++++++++++++++
 crates/buzz-relay/src/handlers/ingest.rs      |   4 +-
 crates/buzz-workflow/src/integration_tests.rs | 101 +++
 crates/buzz-workflow/src/lib.rs               |  39 +-
 crates/buzz-workflow/src/schema.rs            |  41 ++
 docs/nips/NIP-IG.md                           |  75 ++
 mobile/lib/shared/relay/nostr_models.dart     |   1 +
 voltage/images/README.md                      |   7 +-
 voltage/images/buzz-ingest/Dockerfile         |  16 +
 .../buzz-ingest/Dockerfile.dockerignore       |  82 +++
 voltage/images/buzz-ingest/test-image.sh      |  31 +
 25 files changed, 2815 insertions(+), 9 deletions(-)
 create mode 100644 crates/buzz-core/src/integration.rs
 create mode 100644 crates/buzz-ingest/Cargo.toml
 create mode 100644 crates/buzz-ingest/README.md
 create mode 100644 crates/buzz-ingest/src/config.rs
 create mode 100644 crates/buzz-ingest/src/http.rs
 create mode 100644 crates/buzz-ingest/src/main.rs
 create mode 100644 crates/buzz-ingest/src/normalize.rs
 create mode 100644 crates/buzz-ingest/src/plan.rs
 create mode 100644 crates/buzz-ingest/src/relay.rs
 create mode 100644 crates/buzz-ingest/src/tests.rs
 create mode 100644 crates/buzz-workflow/src/integration_tests.rs
 create mode 100644 docs/nips/NIP-IG.md
 create mode 100644 voltage/images/buzz-ingest/Dockerfile
 create mode 100644 voltage/images/buzz-ingest/Dockerfile.dockerignore
 create mode 100755 voltage/images/buzz-ingest/test-image.sh

diff --git a/.github/workflows/voltage-agent-images.yml b/.github/workflows/voltage-agent-images.yml
index c482d3e2c12..6021ab8b436 100644
--- a/.github/workflows/voltage-agent-images.yml
+++ b/.github/workflows/voltage-agent-images.yml
@@ -9,9 +9,10 @@ name: Voltage images
 #   buzz-agent           voltage/images/buzz-agent/Dockerfile → voltage-buzz-agent-staging
 #   buzz-sandbox-reaper  voltage/images/buzz-sandbox-reaper/Dockerfile → voltage-buzz-sandbox-reaper-staging
 #   buzz-relay           the upstream root Dockerfile (target runtime) → buzz-relay
+#   buzz-ingest          voltage/images/buzz-ingest/Dockerfile → buzz-ingest
 #
 # The changes job selects only the images whose inputs changed. Manual runs
-# (workflow_dispatch) build all three, or only buzz-agent with agent_only.
+# (workflow_dispatch) build all images, or only buzz-agent with agent_only.
 #
 # Trust model: this repository and its logs are public. Pull requests run
 # without AWS access. Only the publish and merge jobs request an OIDC token,
@@ -131,6 +132,15 @@ jobs:
               - 'examples/**'
               - 'voltage/images/buzz-sandbox-reaper/**'
               - '.github/workflows/voltage-agent-images.yml'
+            ingest:
+              - '.cargo/**'
+              - 'Cargo.toml'
+              - 'Cargo.lock'
+              - 'rust-toolchain.toml'
+              - 'crates/**'
+              - 'examples/**'
+              - 'voltage/images/buzz-ingest/**'
+              - '.github/workflows/voltage-agent-images.yml'
             relay:
               - 'Dockerfile'
               - '.dockerignore'
@@ -157,16 +167,19 @@ jobs:
           AGENT: ${{ steps.filter.outputs.agent }}
           REAPER: ${{ steps.filter.outputs.reaper }}
           RELAY: ${{ steps.filter.outputs.relay }}
+          INGEST: ${{ steps.filter.outputs.ingest }}
         run: |
           set -euo pipefail
           selected=()
           if [ "$ALL" = true ] || [ "$AGENT_ONLY" = true ] || [ "$AGENT" = true ]; then selected+=(buzz-agent); fi
           if [ "$ALL" = true ] || [ "$REAPER" = true ]; then selected+=(buzz-sandbox-reaper); fi
           if [ "$ALL" = true ] || [ "$RELAY" = true ]; then selected+=(buzz-relay); fi
+          if [ "$ALL" = true ] || [ "$INGEST" = true ]; then selected+=(buzz-ingest); fi
           images='{
             "buzz-agent": {"file": "voltage/images/buzz-agent/Dockerfile", "target": "", "repository": "voltage-buzz-agent-staging"},
             "buzz-sandbox-reaper": {"file": "voltage/images/buzz-sandbox-reaper/Dockerfile", "target": "", "repository": "voltage-buzz-sandbox-reaper-staging"},
-            "buzz-relay": {"file": "Dockerfile", "target": "runtime", "repository": "buzz-relay"}
+            "buzz-relay": {"file": "Dockerfile", "target": "runtime", "repository": "buzz-relay"},
+            "buzz-ingest": {"file": "voltage/images/buzz-ingest/Dockerfile", "target": "", "repository": "buzz-ingest"}
           }'
           platforms='[{"arch": "amd64", "runner": "ubuntu-latest"}, {"arch": "arm64", "runner": "ubuntu-24.04-arm"}]'
           build=$(jq -cn --argjson images "$images" --argjson platforms "$platforms" \
@@ -208,7 +221,7 @@ jobs:
         run: |
           set -euo pipefail
           sh voltage/images/buzz-agent/select-skills.sh --none "$RUNNER_TEMP/engineering-skills"
-          for file in voltage/images/buzz-agent/Dockerfile voltage/images/buzz-sandbox-reaper/Dockerfile Dockerfile; do
+          for file in voltage/images/buzz-agent/Dockerfile voltage/images/buzz-sandbox-reaper/Dockerfile voltage/images/buzz-ingest/Dockerfile Dockerfile; do
             echo "==> $file"
             docker buildx build --check --build-context engineering-skills="$RUNNER_TEMP/engineering-skills" -f "$file" .
           done
@@ -261,6 +274,9 @@ jobs:
           docker run --rm "$IMAGE" --help
           docker run --rm "$IMAGE" reconcile --help
           docker run --rm "$IMAGE" serve --help | grep -- '--mention-launch'
+      - name: Ingest smoke test
+        if: matrix.image == 'buzz-ingest'
+        run: bash voltage/images/buzz-ingest/test-image.sh "$IMAGE"
       - name: Relay smoke test
         if: matrix.image == 'buzz-relay'
         run: bash voltage/images/test-relay-image.sh "$IMAGE"
@@ -414,6 +430,9 @@ jobs:
         run: |
           docker run --rm "$IMAGE" --help
           docker run --rm "$IMAGE" serve --help | grep -- '--mention-launch'
+      - name: Ingest smoke test
+        if: matrix.image == 'buzz-ingest'
+        run: bash voltage/images/buzz-ingest/test-image.sh "$IMAGE"
       - name: Relay smoke test
         if: matrix.image == 'buzz-relay'
         run: bash voltage/images/test-relay-image.sh "$IMAGE"
diff --git a/Cargo.lock b/Cargo.lock
index b7e7e325c85..d1270c3a11e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1812,6 +1812,29 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "buzz-ingest"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "axum",
+ "base64 0.22.1",
+ "buzz-core",
+ "chrono",
+ "hex",
+ "hmac 0.13.0",
+ "nostr 0.44.7",
+ "reqwest 0.13.4",
+ "serde",
+ "serde_json",
+ "sha2 0.11.0",
+ "tokio",
+ "tokio-util",
+ "tower 0.5.3",
+ "url",
+ "uuid",
+]
+
 [[package]]
 name = "buzz-media"
 version = "0.1.0"
diff --git a/Cargo.toml b/Cargo.toml
index 90fbc6d96ff..bb19b20e68f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -27,6 +27,7 @@ members = [
     "crates/git-credential-nostr",
     "crates/git-sign-nostr",
     "crates/buzz-pair-relay",
+    "crates/buzz-ingest",
     "crates/buzz-relay-mesh",
     "crates/buzz-mesh-smoke",
     "crates/buzz-dev-mcp",
diff --git a/crates/buzz-core/src/integration.rs b/crates/buzz-core/src/integration.rs
new file mode 100644
index 00000000000..1ef6b3dc0ed
--- /dev/null
+++ b/crates/buzz-core/src/integration.rs
@@ -0,0 +1,210 @@
+//! NIP-IG integration event tags — the shared contract between integration
+//! publishers (such as `buzz-ingest`) and consumers (the workflow engine).
+//!
+//! An integration event is an ordinary channel event (a kind:9 message, or a
+//! kind:30623 [`crate::kind::KIND_INTEGRATION_RECORD`]) that mirrors an
+//! occurrence in an external system: a Linear issue comment, a GitHub check
+//! run, a Slack message. Its typed metadata lives in `ext-*` tags:
+//!
+//! ```text
+//! ["ext-source", "linear"]           provider
+//! ["ext-action", "issue.labeled"]    normalized "<object>.<verb>"
+//! ["ext-id", "<provider object id>"]
+//! ["ext-url", "https://linear.app/..."]
+//! ["ext-actor", "<provider login or id>"]
+//! ["ext-thread", "<thread key>"]
+//! ["ext-key", "<dedup key>"]
+//! ["ext-labels", "agent"]             repeatable
+//! ```
+//!
+//! Consumers read these generically: every `ext-<name>` tag becomes the field
+//! `<name>` (dashes folded to underscores), and repeated tags join with `,`.
+//! Nothing here knows about any provider. See `docs/nips/NIP-IG.md`.
+//!
+//! The tags are signer-controlled metadata. A consumer that acts on them must
+//! first check that the signer is a trusted integration publisher.
+
+use std::collections::BTreeMap;
+
+/// Prefix shared by every integration metadata tag.
+pub const EXT_TAG_PREFIX: &str = "ext-";
+/// Provider name, for example `github`, `linear`, or `slack`.
+pub const TAG_SOURCE: &str = "ext-source";
+/// Normalized action, `<object>.<verb>`, for example `pull_request.opened`.
+pub const TAG_ACTION: &str = "ext-action";
+/// Provider-side identifier of the object the occurrence is about.
+pub const TAG_ID: &str = "ext-id";
+/// Canonical provider URL of the object.
+pub const TAG_URL: &str = "ext-url";
+/// Provider-side actor (login, user id) who caused the occurrence.
+pub const TAG_ACTOR: &str = "ext-actor";
+/// Key of the external thread (issue, pull request, Slack thread).
+pub const TAG_THREAD: &str = "ext-thread";
+/// Stable dedup key of the occurrence; equal to the record's `d` tag.
+pub const TAG_KEY: &str = "ext-key";
+/// Repository, as `owner/name` (GitHub).
+pub const TAG_REPO: &str = "ext-repo";
+/// Team or workspace key (Linear team key, Slack team id).
+pub const TAG_TEAM: &str = "ext-team";
+/// Human reference, for example `ENG-123` or `owner/repo#42`.
+pub const TAG_REF: &str = "ext-ref";
+/// Title of the object the thread mirrors.
+pub const TAG_TITLE: &str = "ext-title";
+/// Current workflow state (Linear state name, GitHub open/closed/merged).
+pub const TAG_STATE: &str = "ext-state";
+/// A label currently on the object. Repeatable.
+pub const TAG_LABEL: &str = "ext-labels";
+/// A label this occurrence added. Repeatable.
+pub const TAG_ADDED_LABEL: &str = "ext-added-labels";
+/// CI conclusion (`success`, `failure`, ...).
+pub const TAG_CONCLUSION: &str = "ext-conclusion";
+/// Review state (`approved`, `changes_requested`, `commented`).
+pub const TAG_REVIEW_STATE: &str = "ext-review-state";
+/// Provider channel id (Slack channel).
+pub const TAG_CHANNEL: &str = "ext-channel";
+
+/// On a kind:30623 record: the id of the kind:9 message that renders it.
+///
+/// Deliberately not an `e` tag. The record is written before its message
+/// (so a retry can never orphan a message), and an `e` tag would make the
+/// relay resolve NIP-10 thread metadata against an event that does not exist
+/// yet.
+pub const TAG_MESSAGE: &str = "message";
+
+/// Most `ext-*` fields read from one event. Bounds work on a hostile event.
+pub const MAX_FIELDS: usize = 32;
+
+/// Most bytes kept per field value, after joining repeated tags.
+pub const MAX_FIELD_BYTES: usize = 4096;
+
+/// Read every `ext-*` tag of `tags` into a field map.
+///
+/// - `ext-added-labels` becomes the key `added_labels`.
+/// - Repeated tags join their values with `,` in tag order.
+/// - Names that would not be a plain identifier (ASCII letters, digits,
+///   underscore after folding) are skipped, so every key is usable as an
+///   evalexpr variable suffix.
+/// - At most [`MAX_FIELDS`] keys; values are cut at [`MAX_FIELD_BYTES`] on a
+///   character boundary.
+pub fn ext_fields(tags: &nostr::Tags) -> BTreeMap<String, String> {
+    let mut fields: BTreeMap<String, String> = BTreeMap::new();
+    for tag in tags.iter() {
+        let parts = tag.as_slice();
+        let (Some(name), Some(value)) = (parts.first(), parts.get(1)) else {
+            continue;
+        };
+        let Some(raw_key) = name.strip_prefix(EXT_TAG_PREFIX) else {
+            continue;
+        };
+        let key = raw_key.replace('-', "_");
+        if key.is_empty() || !key.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+            continue;
+        }
+        if !fields.contains_key(&key) && fields.len() >= MAX_FIELDS {
+            continue;
+        }
+        let slot = fields.entry(key).or_default();
+        if !slot.is_empty() {
+            slot.push(',');
+        }
+        slot.push_str(value);
+        truncate_on_char_boundary(slot, MAX_FIELD_BYTES);
+    }
+    fields
+}
+
+/// True when `tags` carry both an `ext-source` and an `ext-action` value.
+pub fn is_integration_event(tags: &nostr::Tags) -> bool {
+    let has = |name: &str| {
+        tags.iter().any(|t| {
+            let parts = t.as_slice();
+            parts.first().map(String::as_str) == Some(name)
+                && parts.get(1).is_some_and(|v| !v.is_empty())
+        })
+    };
+    has(TAG_SOURCE) && has(TAG_ACTION)
+}
+
+fn truncate_on_char_boundary(s: &mut String, max: usize) {
+    if s.len() <= max {
+        return;
+    }
+    let mut cut = max;
+    while !s.is_char_boundary(cut) {
+        cut -= 1;
+    }
+    s.truncate(cut);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn tags(raw: &[&[&str]]) -> nostr::Tags {
+        nostr::Tags::from_list(
+            raw.iter()
+                .map(|t| nostr::Tag::parse(t.iter().copied()).unwrap())
+                .collect(),
+        )
+    }
+
+    #[test]
+    fn ext_fields_folds_names_and_joins_repeats() {
+        let t = tags(&[
+            &["h", "chan"],
+            &["ext-source", "linear"],
+            &["ext-action", "issue.labeled"],
+            &["ext-labels", "agent"],
+            &["ext-labels", "bug"],
+            &["ext-added-labels", "agent"],
+            &["message", "abc"],
+        ]);
+        let f = ext_fields(&t);
+        assert_eq!(f.get("source").map(String::as_str), Some("linear"));
+        assert_eq!(f.get("action").map(String::as_str), Some("issue.labeled"));
+        assert_eq!(f.get("labels").map(String::as_str), Some("agent,bug"));
+        assert_eq!(f.get("added_labels").map(String::as_str), Some("agent"));
+        assert!(!f.contains_key("h"));
+        assert!(!f.contains_key("message"));
+    }
+
+    #[test]
+    fn ext_fields_skips_non_identifier_names_and_bounds_output() {
+        let mut raw: Vec<Vec<String>> = vec![
+            vec!["ext-".into(), "empty".into()],
+            vec!["ext-a.b".into(), "dot".into()],
+            vec!["ext-bad name".into(), "space".into()],
+        ];
+        for i in 0..(MAX_FIELDS + 5) {
+            raw.push(vec![format!("ext-f{i}"), "v".into()]);
+        }
+        raw.push(vec!["ext-big".into(), "é".repeat(MAX_FIELD_BYTES)]);
+        let t = nostr::Tags::from_list(
+            raw.iter()
+                .map(|t| nostr::Tag::parse(t.iter().map(String::as_str)).unwrap())
+                .collect(),
+        );
+        let f = ext_fields(&t);
+        assert_eq!(f.len(), MAX_FIELDS);
+        assert!(f.keys().all(|k| !k.is_empty() && !k.contains('.')));
+
+        let big = tags(&[&["ext-big", &"é".repeat(MAX_FIELD_BYTES)]]);
+        let f = ext_fields(&big);
+        let v = f.get("big").unwrap();
+        assert!(v.len() <= MAX_FIELD_BYTES);
+        assert!(v.chars().all(|c| c == 'é'));
+    }
+
+    #[test]
+    fn is_integration_event_requires_source_and_action() {
+        assert!(is_integration_event(&tags(&[
+            &["ext-source", "github"],
+            &["ext-action", "pull_request.opened"],
+        ])));
+        assert!(!is_integration_event(&tags(&[&["ext-source", "github"]])));
+        assert!(!is_integration_event(&tags(&[
+            &["ext-source", "github"],
+            &["ext-action", ""],
+        ])));
+    }
+}
diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs
index 9f66666467f..1a221868125 100644
--- a/crates/buzz-core/src/kind.rs
+++ b/crates/buzz-core/src/kind.rs
@@ -505,6 +505,18 @@ pub const KIND_WORKFLOW_DEF: u32 = 30620;
 /// `hidden_at` per viewer; this is the only Nostr-visible projection of it.
 pub const KIND_DM_VISIBILITY: u32 = 30622;
 
+/// NIP-IG: integration record (parameterized replaceable, d=dedup key).
+///
+/// Published by an integration publisher such as `buzz-ingest` for every
+/// external occurrence (a Linear comment, a GitHub check run, a Slack message)
+/// and every external thread it mirrors. The `d` tag is the stable dedup key,
+/// so a redelivered provider webhook resolves to the same record instead of a
+/// second message. The `message` tag names the kind:9 event that renders the
+/// occurrence in the channel. The record content holds a bounded base64
+/// write-ahead plan; ordinary message searches select kind 9. Requires an `h` tag. See
+/// `docs/nips/NIP-IG.md`.
+pub const KIND_INTEGRATION_RECORD: u32 = 30623;
+
 /// Lower bound of the NIP-33 parameterized replaceable range (30000–39999).
 pub const PARAM_REPLACEABLE_KIND_MIN: u32 = 30000;
 /// Upper bound of the NIP-33 parameterized replaceable range (30000–39999).
@@ -842,6 +854,7 @@ pub const ALL_KINDS: &[u32] = &[
     KIND_GIT_STATUS_CLOSED,
     KIND_GIT_STATUS_DRAFT,
     KIND_PROJECT,
+    KIND_INTEGRATION_RECORD,
 ];
 
 /// Returns `true` if `kind` is in the ephemeral range (20000–29999).
@@ -947,6 +960,7 @@ const _: () = assert!(is_parameterized_replaceable(KIND_WORKFLOW_DEF)); // 30620
 const _: () = assert!(is_parameterized_replaceable(KIND_EVENT_REMINDER)); // 30300 ∈ 30000–39999
 const _: () = assert!(is_parameterized_replaceable(KIND_DM_VISIBILITY)); // 30622 ∈ 30000–39999
 const _: () = assert!(is_parameterized_replaceable(KIND_PROJECT)); // 30621 ∈ 30000–39999
+const _: () = assert!(is_parameterized_replaceable(KIND_INTEGRATION_RECORD)); // 30623 ∈ 30000–39999
 const _: () = assert!(is_parameterized_replaceable(KIND_THREAD_SUMMARY)); // 39005 ∈ 30000–39999
 const _: () = assert!(is_parameterized_replaceable(KIND_WINDOW_BOUNDS)); // 39006 ∈ 30000–39999
 
diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs
index 99b4aa68032..43e8f7fcf6b 100644
--- a/crates/buzz-core/src/lib.rs
+++ b/crates/buzz-core/src/lib.rs
@@ -20,6 +20,8 @@ pub mod event;
 pub mod filter;
 /// Git permission types — ref patterns, protection rules, policy evaluation.
 pub mod git_perms;
+/// NIP-IG integration event tags (`ext-*`) shared by publishers and consumers.
+pub mod integration;
 /// Shared invite-link contract constants.
 pub mod invite;
 /// Buzz kind number registry — custom event type constants.
diff --git a/crates/buzz-ingest/Cargo.toml b/crates/buzz-ingest/Cargo.toml
new file mode 100644
index 00000000000..3b94174fcb0
--- /dev/null
+++ b/crates/buzz-ingest/Cargo.toml
@@ -0,0 +1,29 @@
+[package]
+name = "buzz-ingest"
+description = "Verified provider webhooks published as relay-backed Buzz threads"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+axum = { workspace = true }
+base64 = { workspace = true }
+buzz-core = { path = "../buzz-core" }
+chrono = { workspace = true }
+hex = { workspace = true }
+hmac = { workspace = true }
+nostr = { workspace = true }
+reqwest = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+sha2 = { workspace = true }
+tokio = { workspace = true }
+tokio-util = { workspace = true }
+url = { workspace = true }
+uuid = { workspace = true }
+
+[dev-dependencies]
+tower = { workspace = true }
diff --git a/crates/buzz-ingest/README.md b/crates/buzz-ingest/README.md
new file mode 100644
index 00000000000..789c12a383d
--- /dev/null
+++ b/crates/buzz-ingest/README.md
@@ -0,0 +1,105 @@
+# buzz-ingest
+
+`buzz-ingest` verifies GitHub, Linear, and public Slack webhooks, then publishes
+ordinary thread messages to Buzz. It runs beside the relay, with its own member
+key. The relay remains the durable store. Slack DMs and private channels are excluded.
+
+| Source | Supported activity |
+| --- | --- |
+| GitHub | Pull requests, PR comments/reviews/review comments, and check runs. Checks target at most four linked PRs, or a commit thread. |
+| Linear | Issues, labels added through issue updates, and comments. |
+| Slack | Public-channel messages and message edits; signed URL verification. |
+
+GitHub actions derive from authenticated body fields. The unsigned event header
+must match the payload shape. Unsupported or unmapped activity returns an explicit
+ignored response. Mapped publication returns 200 only after the relay accepts its
+plans, anchor, and reply; incomplete writes return 503 for provider retry.
+
+See [NIP-IG](../../docs/nips/NIP-IG.md) for the event and idempotency contract.
+Single-replica Recreate deployment is required. No PVC or local journal is needed.
+
+## Configuration
+
+Set `BUZZ_INGEST_CONFIG` to this JSON file:
+
+```json
+{
+  "relay_url": "https://buzz.voltage.cloud",
+  "github_repos": {"voltagecloud/buzz": "<channel UUID>"},
+  "linear_teams": {"<team UUID>": "<channel UUID>"},
+  "linear_projects": {"<project UUID>": "<channel UUID>"},
+  "linear_webhooks": {"<webhook UUID>": "<channel UUID>"},
+  "linear_labels": {"<label UUID>": "agent"},
+  "slack_channels": {"<Slack team ID>/<public channel ID>": "<channel UUID>"},
+  "slack_workspace_url": "https://voltagecloud.slack.com"
+}
+```
+
+Every mapping is optional; configure between one and 256 total entries.
+Linear project mappings precede team mappings. Comments retain a previously saved
+issue route; a signed `webhookId` mapping supplies the fallback for unseen issues.
+Use `linear_webhooks` when comments on existing issues omit project/team fields.
+Previously mirrored issues also retain a relay-backed routing record.
+No Linear API credential or data mutation is required.
+[Linear issue payloads carry label UUIDs](https://github.com/linear/linear-solutions/blob/main/integration_guides/scenarios/03-webhooks/README.md).
+Map names such as `agent` with `linear_labels`; unmapped labels retain their UUID.
+Both issue creation and label additions populate `ext-added-labels`.
+
+| Environment variable | Purpose |
+| --- | --- |
+| `BUZZ_INGEST_PRIVATE_KEY` | Required publisher Nostr secret key. |
+| `BUZZ_INGEST_GITHUB_SECRET` | GitHub webhook HMAC secret; required with GitHub mappings. |
+| `BUZZ_INGEST_LINEAR_SECRET` | Linear webhook HMAC secret; required with Linear mappings. |
+| `BUZZ_INGEST_SLACK_SECRET` | Slack app signing secret; required with Slack mappings. |
+| `BUZZ_INGEST_BIND_ADDR` | Listener address; default `0.0.0.0:5010`. |
+
+Relay endpoints require HTTPS; loopback HTTP is allowed for isolated testing.
+The process logs no provider payload, secret, or upstream error body.
+It reads configuration and credentials at startup; restart after changes.
+
+## Provider setup and rollout
+
+1. Create the `buzz-ingest` ECR repository and grant the existing image publisher
+   push access. CI publishes both native architectures from main; PRs build only.
+2. Register the dedicated publisher as a relay member through `buzz-admin`.
+   Add its public key to each mapped channel through channel administration.
+3. Store the publisher key and selected provider secrets in a Kubernetes Secret.
+4. Deploy the companion CD chart with one replica, Recreate, and the image digest.
+5. Configure provider callbacks at `/ingest/github`, `/ingest/linear`, and
+   `/ingest/slack` under the existing HTTPS hostname.
+6. Create workflows with the publisher's public key in `trigger.authors`.
+
+`/health` reports process health. `/ready` performs a signed strong relay query
+and returns 503 when that query fails. Each relay request times out after two
+seconds; the complete delivery has a four-second deadline. Provider retries
+replay the stored plan. No background retry silently acknowledges failed work.
+
+Bodies are limited to 1 MiB. Each provider admits 20 requests per second, and
+one delivery runs at a time; excess requests receive 429. Message bodies retain
+at most 6000 UTF-8 bytes. Payloads exceeding 16 labels are refused, so workflow metadata is never silently
+truncated. Other metadata has separate bounds.
+
+Configure GitHub org webhooks for the supported events and a shared secret.
+[GitHub signs the raw body with HMAC-SHA256](https://docs.github.com/en/webhooks/using-webhooks/validating-webhook-deliveries).
+Configure Linear issue/comment webhooks for mapped teams or webhook IDs.
+[Linear signatures cover the raw body and its signed timestamp](https://linear.app/developers/webhooks);
+requests outside five minutes are refused. Slack needs `message.channels` event
+subscriptions, the `channels:history` bot scope, and membership in mapped public
+channels. [Slack signs the version, timestamp, and raw body](https://docs.slack.dev/authentication/verifying-requests-from-slack/).
+No DM/private-channel scopes are needed. Providers have finite retry policies;
+monitor delivery failures and manually redeliver after prolonged outages.
+
+## Validation
+
+```sh
+. ./bin/activate-hermit
+cargo test -p buzz-ingest
+cargo clippy -p buzz-ingest --all-targets -- -D warnings
+cargo run -p buzz-ingest
+```
+
+The tests drive the real HTTP router and NIP-98 client against a fake relay.
+They exercise signature refusal, freshness, private Slack exclusion, write-ahead
+ordering, lost acknowledgements, restart recovery, payload replay, stale pending
+plans, strong-read failures, and mapping. External provider delivery and production
+relay admission require deployment validation.
diff --git a/crates/buzz-ingest/src/config.rs b/crates/buzz-ingest/src/config.rs
new file mode 100644
index 00000000000..0551b9f0c62
--- /dev/null
+++ b/crates/buzz-ingest/src/config.rs
@@ -0,0 +1,122 @@
+//! Nonsecret provider-to-channel mappings. Credentials come only from env.
+
+use anyhow::{bail, Context, Result};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use uuid::Uuid;
+
+#[derive(Clone, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct Config {
+    pub relay_url: String,
+    #[serde(default)]
+    pub github_repos: BTreeMap<String, Uuid>,
+    #[serde(default)]
+    pub linear_teams: BTreeMap<String, Uuid>,
+    #[serde(default)]
+    pub linear_projects: BTreeMap<String, Uuid>,
+    #[serde(default)]
+    pub linear_webhooks: BTreeMap<String, Uuid>,
+    #[serde(default)]
+    pub linear_labels: BTreeMap<String, String>,
+    #[serde(default)]
+    pub slack_channels: BTreeMap<String, Uuid>,
+    #[serde(default)]
+    pub slack_workspace_url: String,
+}
+
+impl Config {
+    pub fn validate(&self) -> Result<()> {
+        secure_url(&self.relay_url)?;
+        let count = self.github_repos.len()
+            + self.linear_teams.len()
+            + self.linear_projects.len()
+            + self.linear_webhooks.len()
+            + self.slack_channels.len();
+        if count == 0 || count > 256 {
+            bail!("configure between 1 and 256 provider mappings");
+        }
+        if self.linear_labels.len() > 256
+            || self.linear_labels.iter().any(|(id, name)| {
+                id.is_empty() || id.len() > 256 || name.is_empty() || name.len() > 256
+            })
+        {
+            bail!("configure at most 256 nonempty Linear label names, each at most 256 bytes");
+        }
+        if !self.slack_channels.is_empty() {
+            let url = secure_url(&self.slack_workspace_url)?;
+            if url.scheme() != "https"
+                || !url
+                    .host_str()
+                    .is_some_and(|host| host.ends_with(".slack.com"))
+            {
+                bail!("Slack workspace URL must use https://<workspace>.slack.com");
+            }
+            if self.slack_channels.keys().any(|key| {
+                !key.split_once('/').is_some_and(|(team, channel)| {
+                    team.starts_with('T')
+                        && channel.starts_with('C')
+                        && team
+                            .bytes()
+                            .chain(channel.bytes())
+                            .all(|b| b.is_ascii_alphanumeric())
+                })
+            }) {
+                bail!("Slack mappings require team/public-channel IDs");
+            }
+        }
+        Ok(())
+    }
+}
+
+pub(crate) fn secure_url(raw: &str) -> Result<url::Url> {
+    let url = url::Url::parse(raw).context("invalid endpoint URL")?;
+    let loopback = url.host_str().is_some_and(|host| {
+        host == "localhost"
+            || host
+                .parse::<std::net::IpAddr>()
+                .is_ok_and(|ip| ip.is_loopback())
+    });
+    if !(url.scheme() == "https" || (url.scheme() == "http" && loopback))
+        || url.host_str().is_none()
+        || !url.username().is_empty()
+        || url.password().is_some()
+        || url.query().is_some()
+        || url.fragment().is_some()
+        || url.path() != "/"
+    {
+        bail!("endpoint must use HTTPS without path, credentials, query, or fragment (loopback HTTP is allowed)");
+    }
+    Ok(url)
+}
+
+pub(crate) struct Secrets {
+    pub github: Option<String>,
+    pub linear: Option<String>,
+    pub slack: Option<String>,
+}
+
+impl Secrets {
+    pub fn from_env(config: &Config) -> Result<Self> {
+        let get = |name: &str, required: bool| -> Result<Option<String>> {
+            match std::env::var(name) {
+                Ok(value) if !value.is_empty() && value.len() <= 8192 => Ok(Some(value)),
+                Err(std::env::VarError::NotPresent) if !required => Ok(None),
+                _ => bail!("missing or invalid {name}"),
+            }
+        };
+        Ok(Self {
+            github: get("BUZZ_INGEST_GITHUB_SECRET", !config.github_repos.is_empty())?,
+            linear: get(
+                "BUZZ_INGEST_LINEAR_SECRET",
+                !(config.linear_teams.is_empty()
+                    && config.linear_projects.is_empty()
+                    && config.linear_webhooks.is_empty()),
+            )?,
+            slack: get(
+                "BUZZ_INGEST_SLACK_SECRET",
+                !config.slack_channels.is_empty(),
+            )?,
+        })
+    }
+}
diff --git a/crates/buzz-ingest/src/http.rs b/crates/buzz-ingest/src/http.rs
new file mode 100644
index 00000000000..cadcbbdb0ea
--- /dev/null
+++ b/crates/buzz-ingest/src/http.rs
@@ -0,0 +1,286 @@
+//! Signature verification, bounded admission, and webhook acknowledgements.
+
+use crate::{
+    config::{Config, Secrets},
+    normalize, plan,
+    relay::Relay,
+};
+use axum::{
+    body::to_bytes,
+    extract::{Path, Request, State},
+    http::{HeaderMap, StatusCode},
+    response::{IntoResponse, Response},
+    routing::{get, post},
+    Json, Router,
+};
+use hmac::{Hmac, KeyInit, Mac};
+use serde_json::{json, Value};
+use sha2::{Digest, Sha256};
+use std::{
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use tokio::sync::{Mutex, Semaphore};
+
+const MAX_BODY: usize = 1024 * 1024;
+const DEADLINE: Duration = Duration::from_secs(4);
+const RATE: usize = 20;
+const AGE: u64 = 300;
+
+type Error = (StatusCode, &'static str);
+type Result<T> = std::result::Result<T, Error>;
+
+struct Bucket {
+    start: Instant,
+    count: usize,
+}
+struct App {
+    config: Config,
+    secrets: Secrets,
+    relay: Relay,
+    pipeline: Semaphore,
+    readiness: Semaphore,
+    rates: Mutex<[Bucket; 3]>,
+}
+
+pub(crate) fn router(config: Config, secrets: Secrets, relay: Relay) -> Router {
+    let now = Instant::now();
+    let app = Arc::new(App {
+        config,
+        secrets,
+        relay,
+        pipeline: Semaphore::new(1),
+        readiness: Semaphore::new(1),
+        rates: Mutex::new(std::array::from_fn(|_| Bucket {
+            start: now,
+            count: 0,
+        })),
+    });
+    Router::new()
+        .route("/health", get(|| async { "ok" }))
+        .route("/ready", get(ready))
+        .route("/ingest/{provider}", post(deliver))
+        .with_state(app)
+}
+
+async fn ready(State(app): State<Arc<App>>) -> StatusCode {
+    let Ok(_permit) = app.readiness.try_acquire() else {
+        return StatusCode::SERVICE_UNAVAILABLE;
+    };
+    if app.relay.ready().await.is_ok() {
+        StatusCode::OK
+    } else {
+        StatusCode::SERVICE_UNAVAILABLE
+    }
+}
+
+fn header<'a>(headers: &'a HeaderMap, name: &str) -> Result<&'a str> {
+    headers
+        .get(name)
+        .and_then(|value| value.to_str().ok())
+        .ok_or((StatusCode::UNAUTHORIZED, "missing signature metadata"))
+}
+
+fn signature(secret: &str, signed: &[u8], raw: &str) -> Result<()> {
+    let supplied = hex::decode(raw).map_err(|_| (StatusCode::UNAUTHORIZED, "invalid signature"))?;
+    let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).map_err(|_| {
+        (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "signature verifier unavailable",
+        )
+    })?;
+    mac.update(signed);
+    mac.verify_slice(&supplied)
+        .map_err(|_| (StatusCode::UNAUTHORIZED, "invalid signature"))
+}
+
+fn delivery_id(raw: &str) -> Result<&str> {
+    if raw.is_empty()
+        || raw.len() > 128
+        || !raw
+            .bytes()
+            .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
+    {
+        return Err((StatusCode::BAD_REQUEST, "invalid delivery id"));
+    }
+    Ok(raw)
+}
+
+async fn deliver(
+    State(app): State<Arc<App>>,
+    Path(provider): Path<String>,
+    request: Request,
+) -> Response {
+    let source = match provider.as_str() {
+        "github" => 0,
+        "linear" => 1,
+        "slack" => 2,
+        _ => return StatusCode::NOT_FOUND.into_response(),
+    };
+    let secret = match source {
+        0 => app.secrets.github.as_deref(),
+        1 => app.secrets.linear.as_deref(),
+        _ => app.secrets.slack.as_deref(),
+    };
+    let Some(secret) = secret else {
+        return StatusCode::NOT_FOUND.into_response();
+    };
+    let Ok(_permit) = app.pipeline.try_acquire() else {
+        return (StatusCode::TOO_MANY_REQUESTS, "ingestion busy; retry").into_response();
+    };
+    {
+        let mut rates = app.rates.lock().await;
+        let bucket = &mut rates[source];
+        if bucket.start.elapsed() >= Duration::from_secs(1) {
+            bucket.start = Instant::now();
+            bucket.count = 0;
+        }
+        if bucket.count >= RATE {
+            return (
+                StatusCode::TOO_MANY_REQUESTS,
+                "provider rate exceeded; retry",
+            )
+                .into_response();
+        }
+        bucket.count += 1;
+    }
+    let result = tokio::time::timeout(DEADLINE, process(&app, source, secret, request)).await;
+    match result {
+        Ok(Ok(response)) => Json(response).into_response(),
+        Ok(Err(error)) => error.into_response(),
+        Err(_) => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "ingestion deadline reached; retry",
+        )
+            .into_response(),
+    }
+}
+
+async fn process(app: &App, source: usize, secret: &str, request: Request) -> Result<Value> {
+    let (parts, body) = request.into_parts();
+    let body = to_bytes(body, MAX_BODY)
+        .await
+        .map_err(|_| (StatusCode::PAYLOAD_TOO_LARGE, "body exceeds 1 MiB"))?;
+    let now = nostr::Timestamp::now().as_secs();
+    let mut slack_delivery = None;
+    match source {
+        0 => signature(
+            secret,
+            &body,
+            header(&parts.headers, "x-hub-signature-256")?
+                .strip_prefix("sha256=")
+                .ok_or((StatusCode::UNAUTHORIZED, "invalid signature format"))?,
+        )?,
+        1 => signature(secret, &body, header(&parts.headers, "linear-signature")?)?,
+        _ => {
+            let raw_time = header(&parts.headers, "x-slack-request-timestamp")?;
+            let time = raw_time
+                .parse::<u64>()
+                .map_err(|_| (StatusCode::UNAUTHORIZED, "invalid timestamp"))?;
+            if now.abs_diff(time) > AGE {
+                return Err((StatusCode::UNAUTHORIZED, "stale webhook"));
+            }
+            let mut signed = format!("v0:{raw_time}:").into_bytes();
+            signed.extend_from_slice(&body);
+            signature(
+                secret,
+                &signed,
+                header(&parts.headers, "x-slack-signature")?
+                    .strip_prefix("v0=")
+                    .ok_or((StatusCode::UNAUTHORIZED, "invalid signature format"))?,
+            )?;
+        }
+    }
+    let payload: Value =
+        serde_json::from_slice(&body).map_err(|_| (StatusCode::BAD_REQUEST, "invalid JSON"))?;
+    if source == 1 {
+        let time = payload
+            .get("webhookTimestamp")
+            .and_then(Value::as_u64)
+            .ok_or((StatusCode::UNAUTHORIZED, "missing signed timestamp"))?
+            / 1000;
+        if now.abs_diff(time) > AGE {
+            return Err((StatusCode::UNAUTHORIZED, "stale webhook"));
+        }
+    }
+    if source == 2 {
+        if payload.get("type").and_then(Value::as_str) == Some("url_verification") {
+            let challenge = payload
+                .get("challenge")
+                .and_then(Value::as_str)
+                .filter(|v| v.len() <= 256)
+                .ok_or((StatusCode::BAD_REQUEST, "invalid challenge"))?;
+            return Ok(json!({"challenge":challenge}));
+        }
+        slack_delivery = payload.get("event_id").and_then(Value::as_str);
+    }
+    let delivery = delivery_id(match source {
+        0 => header(&parts.headers, "x-github-delivery")?,
+        1 => header(&parts.headers, "linear-delivery")?,
+        _ => slack_delivery.ok_or((StatusCode::BAD_REQUEST, "missing Slack event_id"))?,
+    })?;
+    let items = match source {
+        0 => normalize::github(
+            &app.config,
+            &payload,
+            delivery,
+            header(&parts.headers, "x-github-event")?,
+        ),
+        1 => {
+            let explicit = payload
+                .pointer("/data/teamId")
+                .and_then(Value::as_str)
+                .is_some_and(|id| app.config.linear_teams.contains_key(id))
+                || payload
+                    .pointer("/data/projectId")
+                    .and_then(Value::as_str)
+                    .is_some_and(|id| app.config.linear_projects.contains_key(id));
+            let route = if !explicit {
+                if let Some(issue) = normalize::linear_issue(&payload) {
+                    plan::linear_channel(&app.relay, issue, &app.config)
+                        .await
+                        .map_err(|_| {
+                            (
+                                StatusCode::SERVICE_UNAVAILABLE,
+                                "issue route lookup failed; retry",
+                            )
+                        })?
+                } else {
+                    None
+                }
+            } else {
+                None
+            };
+            normalize::linear(&app.config, &payload, delivery, route)
+        }
+        _ => normalize::slack(&app.config, &payload, delivery),
+    }
+    .map_err(|_| {
+        (
+            StatusCode::BAD_REQUEST,
+            "unsupported or malformed mapped provider payload",
+        )
+    })?;
+    // Linear can refresh the signed transport timestamp on redelivery. It
+    // must not create a second occurrence of the same authenticated payload.
+    let mut identity_payload = payload.clone();
+    if source == 1 {
+        if let Some(object) = identity_payload.as_object_mut() {
+            object.remove("webhookTimestamp");
+        }
+    }
+    let identity = serde_json::to_vec(&identity_payload)
+        .map_err(|_| (StatusCode::BAD_REQUEST, "invalid payload identity"))?;
+    let body_hash = hex::encode(Sha256::digest(identity));
+    for item in &items {
+        plan::publish(&app.relay, item, &body_hash)
+            .await
+            .map_err(|_| {
+                (
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    "relay write incomplete; retry delivery",
+                )
+            })?;
+    }
+    Ok(json!({"accepted":true,"occurrences":items.len(),"ignored":items.is_empty()}))
+}
diff --git a/crates/buzz-ingest/src/main.rs b/crates/buzz-ingest/src/main.rs
new file mode 100644
index 00000000000..64c37bb3f8e
--- /dev/null
+++ b/crates/buzz-ingest/src/main.rs
@@ -0,0 +1,66 @@
+//! Verified integration webhooks; the relay remains the only durable store.
+
+mod config;
+mod http;
+mod normalize;
+mod plan;
+mod relay;
+#[cfg(test)]
+mod tests;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let path = std::env::var("BUZZ_INGEST_CONFIG")
+        .map_err(|_| anyhow::anyhow!("BUZZ_INGEST_CONFIG is required"))?;
+    use std::io::Read;
+    let mut bytes = Vec::new();
+    std::fs::File::open(path)?
+        .take(256 * 1024 + 1)
+        .read_to_end(&mut bytes)?;
+    if bytes.len() > 256 * 1024 {
+        anyhow::bail!("configuration exceeds 256 KiB");
+    }
+    let config: config::Config = serde_json::from_slice(&bytes)?;
+    config.validate()?;
+    let secrets = config::Secrets::from_env(&config)?;
+    let raw = std::env::var("BUZZ_INGEST_PRIVATE_KEY")
+        .map_err(|_| anyhow::anyhow!("BUZZ_INGEST_PRIVATE_KEY is required"))?;
+    let keys =
+        nostr::Keys::parse(&raw).map_err(|_| anyhow::anyhow!("invalid BUZZ_INGEST_PRIVATE_KEY"))?;
+    let relay = relay::Relay::new(&config.relay_url, keys)?;
+    let addr: std::net::SocketAddr = std::env::var("BUZZ_INGEST_BIND_ADDR")
+        .unwrap_or_else(|_| "0.0.0.0:5010".into())
+        .parse()?;
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+    let stop = tokio_util::sync::CancellationToken::new();
+    let shutdown = stop.clone();
+    let mut server = tokio::spawn(async move {
+        axum::serve(listener, http::router(config, secrets, relay))
+            .with_graceful_shutdown(shutdown.cancelled_owned())
+            .await
+    });
+    tokio::select! {
+        result=&mut server=>result??,
+        result=shutdown_signal()=>{
+            result?;
+            stop.cancel();
+            tokio::time::timeout(std::time::Duration::from_secs(5),server).await???;
+        }
+    }
+    Ok(())
+}
+
+/// Kubernetes sends SIGTERM; local development also supports Ctrl-C.
+async fn shutdown_signal() -> std::io::Result<()> {
+    #[cfg(unix)]
+    {
+        use tokio::signal::unix::{signal, SignalKind};
+        let mut terminate = signal(SignalKind::terminate())?;
+        tokio::select! {
+            result=tokio::signal::ctrl_c()=>result,
+            received=terminate.recv()=>received.ok_or_else(||std::io::Error::other("SIGTERM stream closed")),
+        }
+    }
+    #[cfg(not(unix))]
+    tokio::signal::ctrl_c().await
+}
diff --git a/crates/buzz-ingest/src/normalize.rs b/crates/buzz-ingest/src/normalize.rs
new file mode 100644
index 00000000000..a11acc062ce
--- /dev/null
+++ b/crates/buzz-ingest/src/normalize.rs
@@ -0,0 +1,471 @@
+//! Provider payloads become ordinary channel replies with typed source tags.
+
+use crate::config::Config;
+use anyhow::{bail, Result};
+use serde_json::Value;
+use uuid::Uuid;
+
+pub(crate) struct Occurrence {
+    pub source: &'static str,
+    pub channel: Uuid,
+    pub delivery: String,
+    pub thread: String,
+    pub thread_url: String,
+    pub reference: String,
+    pub content: String,
+    pub fields: Vec<(String, String)>,
+}
+
+fn text<'a>(v: &'a Value, pointer: &str) -> &'a str {
+    v.pointer(pointer).and_then(Value::as_str).unwrap_or("")
+}
+fn short(raw: &str, n: usize) -> String {
+    let mut end = raw.len().min(n);
+    while !raw.is_char_boundary(end) {
+        end -= 1;
+    }
+    raw[..end].to_owned()
+}
+fn required(raw: &str) -> Result<&str> {
+    if raw.is_empty() || raw.len() > 4096 {
+        bail!("missing or oversized provider field");
+    }
+    Ok(raw)
+}
+fn check_labels(value: &Value) -> Result<()> {
+    if value.as_array().is_some_and(|labels| labels.len() > 16) {
+        bail!("provider payload exceeds the 16-label bound");
+    }
+    Ok(())
+}
+fn labels(v: &Value) -> Vec<(String, String)> {
+    v.as_array()
+        .into_iter()
+        .flatten()
+        .take(16)
+        .filter_map(|label| label.get("name").and_then(Value::as_str))
+        .map(|name| ("labels".into(), short(name, 256)))
+        .collect()
+}
+fn base(
+    source: &'static str,
+    channel: Uuid,
+    delivery: &str,
+    thread_info: (String, String, String),
+    action: String,
+    id: String,
+    actor: &str,
+) -> Occurrence {
+    let (thread, url, reference) = thread_info;
+    Occurrence {
+        source,
+        channel,
+        delivery: delivery.to_owned(),
+        thread: thread.clone(),
+        thread_url: url.clone(),
+        reference,
+        content: String::new(),
+        fields: vec![
+            ("source".into(), source.into()),
+            ("action".into(), action),
+            ("id".into(), id),
+            ("url".into(), url),
+            ("actor".into(), short(actor, 256)),
+            ("thread".into(), thread),
+        ],
+    }
+}
+
+/// Shape-check the unsigned GitHub event header against authenticated body fields.
+/// Display/actions derive from signed payload data, never the header alone.
+pub(crate) fn github(
+    config: &Config,
+    payload: &Value,
+    delivery: &str,
+    event_header: &str,
+) -> Result<Vec<Occurrence>> {
+    let inferred = if payload.get("review").is_some() {
+        "pull_request_review"
+    } else if payload.get("comment").is_some() && payload.get("pull_request").is_some() {
+        "pull_request_review_comment"
+    } else if payload.get("comment").is_some() && payload.get("issue").is_some() {
+        "issue_comment"
+    } else if payload.get("pull_request").is_some() {
+        "pull_request"
+    } else if payload.get("check_run").is_some() {
+        "check_run"
+    } else {
+        return Ok(vec![]);
+    };
+    if event_header != inferred {
+        bail!("GitHub event header does not match signed payload shape");
+    }
+    let repo = required(text(payload, "/repository/full_name"))?;
+    let Some(channel) = config.github_repos.get(repo).copied() else {
+        return Ok(vec![]);
+    };
+    let action = required(text(payload, "/action"))?;
+    let actor = text(payload, "/sender/login");
+    let mut targets = Vec::new();
+    let object = if inferred == "check_run" {
+        &payload["check_run"]
+    } else if inferred == "issue_comment" {
+        &payload["issue"]
+    } else {
+        &payload["pull_request"]
+    };
+    if inferred == "check_run" {
+        let prs = object["pull_requests"].as_array();
+        if prs.is_some_and(|prs| prs.len() > 4) {
+            bail!("check run has more than four pull requests");
+        }
+        for pr in prs.into_iter().flatten() {
+            if let Some(number) = pr["number"].as_u64() {
+                targets.push((
+                    format!("pr:{number}"),
+                    format!("{repo}#{number}"),
+                    format!("https://github.com/{repo}/pull/{number}"),
+                ));
+            }
+        }
+        if targets.is_empty() {
+            let sha = required(text(object, "/head_sha"))?;
+            targets.push((
+                format!("commit:{sha}"),
+                format!("{repo}@{}", short(sha, 12)),
+                format!("https://github.com/{repo}/commit/{sha}"),
+            ));
+        }
+    } else {
+        // Ordinary GitHub issues are outside the requested PR review surface.
+        if inferred == "issue_comment" && object.get("pull_request").is_none() {
+            return Ok(vec![]);
+        };
+        let number = object["number"]
+            .as_u64()
+            .ok_or_else(|| anyhow::anyhow!("missing PR number"))?;
+        targets.push((
+            format!("pr:{number}"),
+            format!("{repo}#{number}"),
+            format!("https://github.com/{repo}/pull/{number}"),
+        ));
+    }
+    check_labels(&object["labels"])?;
+    let detail = if payload.get("comment").is_some() {
+        &payload["comment"]
+    } else if payload.get("review").is_some() {
+        &payload["review"]
+    } else {
+        object
+    };
+    let id = detail.get("id").map(Value::to_string).unwrap_or_default();
+    let title = short(
+        if text(object, "/title").is_empty() {
+            text(object, "/name")
+        } else {
+            text(object, "/title")
+        },
+        256,
+    );
+    let body = short(
+        if text(detail, "/body").is_empty() {
+            text(detail, "/output/summary")
+        } else {
+            text(detail, "/body")
+        },
+        6000,
+    );
+    let outcome = if inferred == "check_run" {
+        text(object, "/conclusion")
+    } else if inferred == "pull_request_review" {
+        text(detail, "/state")
+    } else {
+        ""
+    };
+    let mut results = Vec::new();
+    for (thread, reference, url) in targets {
+        let mut item = base(
+            "github",
+            channel,
+            delivery,
+            (format!("{repo}:{thread}"), url, reference.clone()),
+            format!("{inferred}.{action}"),
+            id.clone(),
+            actor,
+        );
+        item.fields.extend([
+            ("repo".into(), repo.into()),
+            ("ref".into(), reference),
+            ("title".into(), title.clone()),
+        ]);
+        item.fields.extend(labels(&object["labels"]));
+        if inferred == "pull_request" && action == "labeled" {
+            let name = text(payload, "/label/name");
+            if !name.is_empty() {
+                item.fields.push(("added-labels".into(), short(name, 256)));
+            }
+        }
+        for (field, pointer) in [("state", "/state"), ("conclusion", "/conclusion")] {
+            let value = if field == "state"
+                && object.get("merged").and_then(Value::as_bool) == Some(true)
+            {
+                "merged"
+            } else if field == "state" && text(object, pointer).is_empty() {
+                text(object, "/status")
+            } else {
+                text(object, pointer)
+            };
+            if !value.is_empty() {
+                item.fields.push((field.into(), short(value, 128)));
+            }
+        }
+        if inferred == "pull_request_review" {
+            item.fields
+                .push(("review-state".into(), short(text(detail, "/state"), 128)));
+        }
+        if let Some(url) = detail.get("html_url").and_then(Value::as_str) {
+            item.fields
+                .iter_mut()
+                .filter(|(name, _)| name == "url")
+                .for_each(|(_, value)| *value = short(url, 4096));
+        }
+        item.content = format!(
+            "GitHub {} — {inferred}.{action} by {actor}\n{title}\n{outcome}\n{body}\n{}",
+            item.reference, item.thread_url
+        );
+        results.push(item);
+    }
+    Ok(results)
+}
+
+pub(crate) fn linear_issue(payload: &Value) -> Option<&str> {
+    (text(payload, "/type") == "Comment")
+        .then(|| text(payload, "/data/issueId"))
+        .filter(|id| !id.is_empty())
+}
+
+pub(crate) fn linear(
+    config: &Config,
+    payload: &Value,
+    delivery: &str,
+    routed: Option<Uuid>,
+) -> Result<Vec<Occurrence>> {
+    let kind = text(payload, "/type");
+    if kind != "Issue" && kind != "Comment" {
+        return Ok(vec![]);
+    };
+    let data = &payload["data"];
+    let channel = config
+        .linear_projects
+        .get(text(data, "/projectId"))
+        .or_else(|| config.linear_teams.get(text(data, "/teamId")))
+        .copied()
+        .or(routed)
+        .or_else(|| {
+            config
+                .linear_webhooks
+                .get(text(payload, "/webhookId"))
+                .copied()
+        });
+    let Some(channel) = channel else {
+        if kind == "Comment" {
+            bail!("Linear issue route is unavailable; deliver the issue before its comments");
+        }
+        return Ok(vec![]);
+    };
+    let id = required(text(data, "/id"))?;
+    let issue = if kind == "Comment" {
+        required(text(data, "/issueId"))?
+    } else {
+        id
+    };
+    let url = required(text(payload, "/url"))?
+        .split('#')
+        .next()
+        .unwrap_or_default()
+        .to_owned();
+    let reference = if text(data, "/identifier").is_empty() {
+        issue
+    } else {
+        text(data, "/identifier")
+    };
+    let actor = text(payload, "/actor/name");
+    let mut action = format!(
+        "{}.{}",
+        kind.to_ascii_lowercase(),
+        required(text(payload, "/action"))?
+    );
+    check_labels(&data["labelIds"])?;
+    check_labels(&data["labels"])?;
+    let label_entries: Vec<(String, String)> = if let Some(ids) = data["labelIds"].as_array() {
+        ids.iter()
+            .take(16)
+            .filter_map(Value::as_str)
+            .map(|id| {
+                let enriched = data["labels"]
+                    .as_array()
+                    .into_iter()
+                    .flatten()
+                    .find(|label| text(label, "/id") == id)
+                    .map(|label| text(label, "/name"));
+                let name = enriched
+                    .filter(|name| !name.is_empty())
+                    .or_else(|| config.linear_labels.get(id).map(String::as_str))
+                    .unwrap_or(id);
+                (id.to_owned(), short(name, 256))
+            })
+            .collect()
+    } else {
+        data["labels"]
+            .as_array()
+            .into_iter()
+            .flatten()
+            .take(16)
+            .map(|label| {
+                (
+                    text(label, "/id").to_owned(),
+                    short(text(label, "/name"), 256),
+                )
+            })
+            .collect()
+    };
+    let mut fields: Vec<_> = label_entries
+        .iter()
+        .map(|(_, name)| ("labels".into(), name.clone()))
+        .collect();
+    let previous = payload.pointer("/updatedFrom/labelIds");
+    for (id, name) in label_entries {
+        let added = text(payload, "/action") == "create"
+            || (text(payload, "/action") == "update"
+                && previous.is_some_and(|old| {
+                    old.is_null()
+                        || old
+                            .as_array()
+                            .is_some_and(|ids| !ids.contains(&Value::String(id.clone())))
+                }));
+        if added {
+            fields.push(("added-labels".into(), name));
+            if text(payload, "/action") == "update" {
+                action = "issue.labeled".into();
+            }
+        }
+    }
+    let mut item = base(
+        "linear",
+        channel,
+        delivery,
+        (format!("issue:{issue}"), url, reference.into()),
+        action.clone(),
+        id.into(),
+        actor,
+    );
+    item.fields
+        .iter_mut()
+        .filter(|(name, _)| name == "url")
+        .for_each(|(_, url)| *url = short(text(payload, "/url"), 4096));
+    item.fields.extend(fields);
+    for (field, pointer) in [
+        ("team", "/teamId"),
+        ("title", "/title"),
+        ("state", "/state/name"),
+    ] {
+        let value = text(data, pointer);
+        if !value.is_empty() {
+            item.fields.push((field.into(), short(value, 256)));
+        }
+    }
+    item.fields.push(("ref".into(), reference.into()));
+    item.content = format!(
+        "Linear {} — {action} by {actor}\n{}\n{}\n{}",
+        item.reference,
+        short(text(data, "/title"), 256),
+        short(
+            if kind == "Comment" {
+                text(data, "/body")
+            } else {
+                text(data, "/description")
+            },
+            6000
+        ),
+        item.thread_url
+    );
+    Ok(vec![item])
+}
+
+pub(crate) fn slack(config: &Config, payload: &Value, delivery: &str) -> Result<Vec<Occurrence>> {
+    let event = &payload["event"];
+    if text(event, "/type") != "message" || text(event, "/channel_type") != "channel" {
+        return Ok(vec![]);
+    };
+    let team = required(text(payload, "/team_id"))?;
+    let channel = required(text(event, "/channel"))?;
+    let Some(buzz_channel) = config
+        .slack_channels
+        .get(&format!("{team}/{channel}"))
+        .copied()
+    else {
+        return Ok(vec![]);
+    };
+    let subtype = text(event, "/subtype");
+    if !subtype.is_empty() && subtype != "message_changed" {
+        return Ok(vec![]);
+    };
+    let message = if subtype == "message_changed" {
+        &event["message"]
+    } else {
+        event
+    };
+    let ts = required(text(message, "/ts"))?;
+    if !ts.bytes().all(|b| b.is_ascii_digit() || b == b'.') {
+        bail!("invalid Slack timestamp");
+    }
+    let root = if text(message, "/thread_ts").is_empty() {
+        ts
+    } else {
+        text(message, "/thread_ts")
+    };
+    let url = format!(
+        "{}/archives/{channel}/p{}",
+        config.slack_workspace_url.trim_end_matches('/'),
+        root.replace('.', "")
+    );
+    let action = if subtype == "message_changed" {
+        "message.updated"
+    } else {
+        "message.created"
+    };
+    let mut item = base(
+        "slack",
+        buzz_channel,
+        delivery,
+        (
+            format!("{team}:{channel}:{root}"),
+            url,
+            format!("{team}/{channel}"),
+        ),
+        action.into(),
+        ts.into(),
+        text(message, "/user"),
+    );
+    let occurrence_url = format!(
+        "{}/archives/{channel}/p{}",
+        config.slack_workspace_url.trim_end_matches('/'),
+        ts.replace('.', "")
+    );
+    item.fields
+        .iter_mut()
+        .filter(|(name, _)| name == "url")
+        .for_each(|(_, url)| *url = occurrence_url.clone());
+    item.fields.extend([
+        ("team".into(), team.into()),
+        ("channel".into(), channel.into()),
+    ]);
+    item.content = format!(
+        "Slack {} — {action} by {}\n{}\n{}",
+        item.reference,
+        text(message, "/user"),
+        short(text(message, "/text"), 6000),
+        item.thread_url
+    );
+    Ok(vec![item])
+}
diff --git a/crates/buzz-ingest/src/plan.rs b/crates/buzz-ingest/src/plan.rs
new file mode 100644
index 00000000000..d1f80c80f6d
--- /dev/null
+++ b/crates/buzz-ingest/src/plan.rs
@@ -0,0 +1,273 @@
+//! Relay-backed write-ahead plans. Records are committed before readable messages.
+
+use crate::{normalize::Occurrence, relay::Relay};
+use anyhow::{bail, Context, Result};
+use base64::{engine::general_purpose::STANDARD, Engine};
+use nostr::{Event, EventBuilder, Kind, Tag, Timestamp};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+const MAX_PLAN: usize = 24 * 1024;
+const REFRESH_AFTER: u64 = 14 * 60;
+
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Plan {
+    version: u8,
+    payload_sha256: Option<String>,
+    event: Event,
+}
+
+fn tag(name: &str, value: &str) -> Result<Tag> {
+    Tag::parse([name, value]).context("invalid integration tag")
+}
+fn value<'a>(event: &'a Event, name: &str) -> Option<&'a str> {
+    event.tags.iter().find_map(|tag| {
+        let p = tag.as_slice();
+        (p.first().map(String::as_str) == Some(name))
+            .then(|| p.get(1).map(String::as_str))
+            .flatten()
+    })
+}
+
+fn validate(relay: &Relay, channel: Uuid, plan: &Plan) -> Result<()> {
+    if plan.version != 1
+        || plan.event.verify().is_err()
+        || plan.event.pubkey != relay.keys.public_key()
+        || plan.event.kind.as_u16() != 9
+        || value(&plan.event, "h") != Some(channel.to_string().as_str())
+        || serde_json::to_vec(plan)?.len() > MAX_PLAN
+    {
+        bail!("invalid saved integration plan");
+    }
+    Ok(())
+}
+
+async fn read(relay: &Relay, channel: Uuid, coordinate: &str) -> Result<Option<Plan>> {
+    let records=relay.query(json!({"kinds":[buzz_core::kind::KIND_INTEGRATION_RECORD],"authors":[relay.keys.public_key().to_hex()],"#d":[coordinate],"#h":[channel.to_string()],"limit":2})).await?;
+    if records.len() > 1 {
+        bail!("ambiguous integration record");
+    }
+    let Some(record) = records.first() else {
+        return Ok(None);
+    };
+    if record.verify().is_err()
+        || record.pubkey != relay.keys.public_key()
+        || record.kind.as_u16() != 30623
+        || value(record, "d") != Some(coordinate)
+        || value(record, "h") != Some(channel.to_string().as_str())
+        || record.content.len() > MAX_PLAN * 2
+    {
+        bail!("invalid integration record");
+    }
+    let bytes = STANDARD
+        .decode(&record.content)
+        .context("invalid plan encoding")?;
+    if bytes.len() > MAX_PLAN {
+        bail!("integration plan exceeded bound");
+    }
+    let plan: Plan = serde_json::from_slice(&bytes).context("invalid saved plan")?;
+    validate(relay, channel, &plan)?;
+    if value(record, "message") != Some(plan.event.id.to_hex().as_str()) {
+        bail!("integration record points to a different event");
+    }
+    Ok(Some(plan))
+}
+
+async fn save(relay: &Relay, channel: Uuid, coordinate: &str, plan: &Plan) -> Result<()> {
+    validate(relay, channel, plan)?;
+    let body = STANDARD.encode(serde_json::to_vec(plan)?);
+    let mut tags = vec![
+        tag("d", coordinate)?,
+        tag("h", &channel.to_string())?,
+        tag("message", &plan.event.id.to_hex())?,
+    ];
+    tags.extend(
+        plan.event
+            .tags
+            .iter()
+            .filter(|tag| {
+                tag.as_slice()
+                    .first()
+                    .is_some_and(|name| name.starts_with("ext-"))
+            })
+            .cloned(),
+    );
+    let event = EventBuilder::new(Kind::Custom(30623), body)
+        .tags(tags)
+        .sign_with_keys(&relay.keys)?;
+    relay.submit(&event).await
+}
+
+async fn exists(relay: &Relay, channel: Uuid, event: &Event) -> Result<bool> {
+    let events=relay.query(json!({"kinds":[9],"ids":[event.id.to_hex()],"authors":[relay.keys.public_key().to_hex()],"#h":[channel.to_string()],"limit":2})).await?;
+    if events.len() > 1 {
+        bail!("ambiguous exact event lookup");
+    }
+    match events.first() {
+        None => Ok(false),
+        Some(found)
+            if found.id == event.id
+                && found.verify().is_ok()
+                && found.pubkey == event.pubkey
+                && value(found, "h") == Some(channel.to_string().as_str()) =>
+        {
+            Ok(true)
+        }
+        Some(_) => bail!("invalid exact event lookup"),
+    }
+}
+
+fn freshen(relay: &Relay, plan: &mut Plan) -> Result<()> {
+    plan.event = EventBuilder::new(plan.event.kind, plan.event.content.clone())
+        .tags(plan.event.tags.clone().to_vec())
+        .custom_created_at(Timestamp::now())
+        .sign_with_keys(&relay.keys)?;
+    Ok(())
+}
+
+fn stale(plan: &Plan) -> bool {
+    Timestamp::now()
+        .as_secs()
+        .abs_diff(plan.event.created_at.as_secs())
+        > REFRESH_AFTER
+}
+
+/// Write one authenticated occurrence and its stable thread anchor. The global
+/// HTTP pipeline serializes this across requests; deployment must use Recreate.
+pub(crate) async fn publish(relay: &Relay, item: &Occurrence, body_hash: &str) -> Result<()> {
+    let scope = hex::encode(Sha256::digest(format!(
+        "{}:{}:{}",
+        item.source, item.channel, item.thread
+    )));
+    let root_coord = format!("thread:{scope}");
+    let mut root = match read(relay, item.channel, &root_coord).await? {
+        Some(plan) => plan,
+        None => Plan {
+            version: 1,
+            payload_sha256: None,
+            event: EventBuilder::new(
+                Kind::Custom(9),
+                format!("{}: {}\n{}", item.source, item.reference, item.thread_url),
+            )
+            .tags([
+                tag("h", &item.channel.to_string())?,
+                tag("ext-thread", &item.thread)?,
+                tag("ext-source", item.source)?,
+            ])
+            .sign_with_keys(&relay.keys)?,
+        },
+    };
+    let root_exists = exists(relay, item.channel, &root.event).await?;
+    if !root_exists {
+        if stale(&root) {
+            freshen(relay, &mut root)?;
+        }
+        save(relay, item.channel, &root_coord, &root).await?;
+        if item.source == "linear" {
+            save(
+                relay,
+                item.channel,
+                &format!("route:linear:{}", item.thread),
+                &root,
+            )
+            .await?;
+        }
+        relay.submit(&root.event).await?;
+    }
+    let delivery = format!("delivery:{scope}:{}", item.delivery);
+    let fingerprint = format!("fingerprint:{scope}:{body_hash}");
+    let a = read(relay, item.channel, &delivery).await?;
+    let b = read(relay, item.channel, &fingerprint).await?;
+    for plan in a.iter().chain(b.iter()) {
+        if plan.payload_sha256.as_deref() != Some(body_hash) {
+            bail!("delivery identity conflicts with saved payload");
+        }
+    }
+    let mut plan = match (a, b) {
+        (Some(a), Some(b)) if a.event.id != b.event.id => {
+            if exists(relay, item.channel, &a.event).await? {
+                a
+            } else if exists(relay, item.channel, &b.event).await? {
+                b
+            } else if a.event.created_at >= b.event.created_at {
+                a
+            } else {
+                b
+            }
+        }
+        (Some(plan), _) | (_, Some(plan)) => plan,
+        (None, None) => {
+            let mut tags = vec![
+                tag("h", &item.channel.to_string())?,
+                Tag::parse(["e", &root.event.id.to_hex(), "", "root"])?,
+            ];
+            for (key, val) in &item.fields {
+                tags.push(tag(&format!("ext-{key}"), val)?);
+            }
+            tags.push(tag("ext-key", &fingerprint)?);
+            Plan {
+                version: 1,
+                payload_sha256: Some(body_hash.to_owned()),
+                event: EventBuilder::new(Kind::Custom(9), &item.content)
+                    .tags(tags)
+                    .sign_with_keys(&relay.keys)?,
+            }
+        }
+    };
+    let accepted = exists(relay, item.channel, &plan.event).await?;
+    if !accepted && stale(&plan) {
+        // Only an authoritative successful empty lookup permits changing IDs.
+        freshen(relay, &mut plan)?;
+        let mut tags = plan.event.tags.clone().to_vec();
+        tags.retain(|tag| tag.as_slice().first().map(String::as_str) != Some("e"));
+        tags.push(Tag::parse(["e", &root.event.id.to_hex(), "", "root"])?);
+        plan.event = EventBuilder::new(Kind::Custom(9), plan.event.content.clone())
+            .tags(tags)
+            .sign_with_keys(&relay.keys)?;
+    }
+    save(relay, item.channel, &delivery, &plan).await?;
+    save(relay, item.channel, &fingerprint, &plan).await?;
+    if !accepted {
+        relay.submit(&plan.event).await?;
+    }
+    Ok(())
+}
+
+/// Resolve comments whose signed payload omits project/team information.
+/// Explicit webhook mappings take precedence over this previously stored route.
+pub(crate) async fn linear_channel(
+    relay: &Relay,
+    issue: &str,
+    config: &crate::config::Config,
+) -> Result<Option<Uuid>> {
+    let coordinate = format!("route:linear:issue:{issue}");
+    let records=relay.query(json!({"kinds":[30623],"authors":[relay.keys.public_key().to_hex()],"#d":[coordinate],"limit":2})).await?;
+    if records.len() > 1 {
+        bail!("ambiguous Linear issue route");
+    }
+    let Some(record) = records.first() else {
+        return Ok(None);
+    };
+    if record.verify().is_err()
+        || record.pubkey != relay.keys.public_key()
+        || value(record, "d") != Some(coordinate.as_str())
+    {
+        bail!("invalid Linear issue route");
+    }
+    let channel = value(record, "h")
+        .and_then(|id| Uuid::parse_str(id).ok())
+        .ok_or_else(|| anyhow::anyhow!("invalid issue route channel"))?;
+    let allowed = config
+        .linear_projects
+        .values()
+        .chain(config.linear_teams.values())
+        .chain(config.linear_webhooks.values())
+        .any(|known| *known == channel);
+    if !allowed {
+        bail!("Linear issue route is outside configured channels");
+    }
+    Ok(Some(channel))
+}
diff --git a/crates/buzz-ingest/src/relay.rs b/crates/buzz-ingest/src/relay.rs
new file mode 100644
index 00000000000..6e3e621ca4d
--- /dev/null
+++ b/crates/buzz-ingest/src/relay.rs
@@ -0,0 +1,111 @@
+//! NIP-98 signed HTTP bridge operations. Every query requests writer consistency.
+
+use anyhow::{bail, Context, Result};
+use base64::{engine::general_purpose::STANDARD, Engine};
+use nostr::{Event, EventBuilder, JsonUtil, Keys, Kind, Tag};
+use serde_json::{json, Value};
+use sha2::{Digest, Sha256};
+
+pub(crate) struct Relay {
+    pub keys: Keys,
+    base: String,
+    client: reqwest::Client,
+}
+
+impl Relay {
+    pub fn new(base: &str, keys: Keys) -> Result<Self> {
+        super::config::secure_url(base)?;
+        Ok(Self {
+            keys,
+            base: base.trim_end_matches('/').to_owned(),
+            client: reqwest::Client::builder()
+                .timeout(std::time::Duration::from_secs(2))
+                .redirect(reqwest::redirect::Policy::none())
+                .build()?,
+        })
+    }
+
+    async fn post(&self, path: &str, value: Value) -> Result<Value> {
+        let url = format!("{}{path}", self.base);
+        let body = serde_json::to_vec(&value)?;
+        let tags = [
+            ("u", url),
+            ("method", "POST".to_owned()),
+            ("payload", hex::encode(Sha256::digest(&body))),
+            ("nonce", uuid::Uuid::new_v4().to_string()),
+        ]
+        .into_iter()
+        .map(|(name, value)| Tag::parse([name, &value]))
+        .collect::<std::result::Result<Vec<_>, _>>()?;
+        let auth = EventBuilder::new(Kind::Custom(27235), "")
+            .tags(tags)
+            .sign_with_keys(&self.keys)?;
+        let response = self
+            .client
+            .post(format!("{}{path}", self.base))
+            .header(
+                "Authorization",
+                format!("Nostr {}", STANDARD.encode(auth.as_json())),
+            )
+            .header("Content-Type", "application/json")
+            .body(body)
+            .send()
+            .await?;
+        if !response.status().is_success() {
+            bail!("relay request refused");
+        }
+        // Bound even a chunked response before parsing it.
+        let mut response = response;
+        let mut bytes = Vec::new();
+        while let Some(chunk) = response.chunk().await? {
+            if bytes.len().saturating_add(chunk.len()) > 128 * 1024 {
+                bail!("relay response exceeded bound");
+            }
+            bytes.extend_from_slice(&chunk);
+        }
+        serde_json::from_slice(&bytes).context("invalid relay response")
+    }
+
+    pub async fn query(&self, mut filter: Value) -> Result<Vec<Event>> {
+        filter["consistency"] = json!("strong");
+        let value = self.post("/query", json!([filter])).await?;
+        let events: Vec<Event> =
+            serde_json::from_value(value).context("invalid relay event array")?;
+        if events.len() > 2 {
+            bail!("relay exceeded exact-query event bound");
+        }
+        Ok(events)
+    }
+
+    pub async fn submit(&self, event: &Event) -> Result<()> {
+        let response = self.post("/events", serde_json::to_value(event)?).await?;
+        // /events returns ["OK", id, accepted, message], as buzz-cli normalizes.
+        let accepted = response
+            .get("accepted")
+            .and_then(Value::as_bool)
+            .or_else(|| {
+                response
+                    .as_array()
+                    .and_then(|items| items.get(2))
+                    .and_then(Value::as_bool)
+            });
+        let acknowledged = response
+            .get("event_id")
+            .or_else(|| response.get("id"))
+            .and_then(Value::as_str)
+            .or_else(|| {
+                response
+                    .as_array()
+                    .and_then(|parts| parts.get(1))
+                    .and_then(Value::as_str)
+            });
+        if accepted != Some(true) || acknowledged != Some(event.id.to_hex().as_str()) {
+            bail!("relay did not accept event");
+        }
+        Ok(())
+    }
+
+    pub async fn ready(&self) -> Result<()> {
+        self.query(json!({"kinds":[buzz_core::kind::KIND_INTEGRATION_RECORD],"authors":[self.keys.public_key().to_hex()],"limit":1})).await.map(drop)
+    }
+}
diff --git a/crates/buzz-ingest/src/tests.rs b/crates/buzz-ingest/src/tests.rs
new file mode 100644
index 00000000000..649c04c07a5
--- /dev/null
+++ b/crates/buzz-ingest/src/tests.rs
@@ -0,0 +1,689 @@
+//! Real HTTP journeys. Only the external relay boundary is faked.
+
+use crate::{
+    config::{Config, Secrets},
+    relay::Relay,
+};
+use axum::{
+    body::Bytes,
+    extract::State,
+    http::{HeaderMap, StatusCode},
+    routing::post,
+    Json, Router,
+};
+use base64::{engine::general_purpose::STANDARD, Engine};
+use hmac::{Hmac, KeyInit, Mac};
+use nostr::{Event, Keys};
+use serde_json::{json, Value};
+use sha2::{Digest, Sha256};
+use std::{
+    collections::BTreeMap,
+    sync::{Arc, Mutex},
+};
+use tower::ServiceExt;
+
+const SECRET: &str = "synthetic-provider-signing-secret";
+const CHANNEL: &str = "0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a10";
+#[derive(Default)]
+struct Store {
+    events: BTreeMap<String, Event>,
+    heads: BTreeMap<String, String>,
+    fail_queries: usize,
+    fail_reply: usize,
+    fail_event_lookup: Option<String>,
+    lose_root_response: bool,
+    submissions: Vec<Event>,
+}
+type Shared = Arc<Mutex<Store>>;
+fn tag<'a>(event: &'a Event, name: &str) -> Option<&'a str> {
+    event.tags.iter().find_map(|tag| {
+        let p = tag.as_slice();
+        if p.first().map(String::as_str) == Some(name) {
+            p.get(1).map(String::as_str)
+        } else {
+            None
+        }
+    })
+}
+fn auth(headers: &HeaderMap, body: &[u8]) {
+    let raw = headers
+        .get("authorization")
+        .unwrap()
+        .to_str()
+        .unwrap()
+        .strip_prefix("Nostr ")
+        .unwrap();
+    let event: Event = serde_json::from_slice(&STANDARD.decode(raw).unwrap()).unwrap();
+    event.verify().unwrap();
+    assert_eq!(event.kind.as_u16(), 27235);
+    assert_eq!(
+        tag(&event, "payload"),
+        Some(hex::encode(Sha256::digest(body)).as_str())
+    );
+    assert_eq!(tag(&event, "method"), Some("POST"));
+}
+async fn query(
+    State(store): State<Shared>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> (StatusCode, Json<Value>) {
+    auth(&headers, &body);
+    let filters: Value = serde_json::from_slice(&body).unwrap();
+    let filter = &filters[0];
+    assert_eq!(filter["consistency"], "strong");
+    let mut store = store.lock().unwrap();
+    if store.fail_event_lookup.as_ref().is_some_and(|id| {
+        filter
+            .get("ids")
+            .and_then(Value::as_array)
+            .is_some_and(|ids| ids.contains(&json!(id)))
+    }) {
+        store.fail_event_lookup = None;
+        return (StatusCode::SERVICE_UNAVAILABLE, Json(json!({})));
+    }
+    if store.fail_queries > 0 {
+        store.fail_queries -= 1;
+        return (StatusCode::SERVICE_UNAVAILABLE, Json(json!({})));
+    }
+    let mut events = Vec::new();
+    for event in store.events.values() {
+        if event.kind.as_u16() == 30623
+            && tag(event, "d").is_some_and(|d| store.heads.get(d) != Some(&event.id.to_hex()))
+        {
+            continue;
+        }
+        let matches = filter["kinds"]
+            .as_array()
+            .unwrap()
+            .contains(&json!(event.kind.as_u16()))
+            && filter
+                .get("ids")
+                .and_then(Value::as_array)
+                .is_none_or(|ids| ids.contains(&json!(event.id.to_hex())))
+            && filter
+                .get("authors")
+                .and_then(Value::as_array)
+                .is_none_or(|ids| ids.contains(&json!(event.pubkey.to_hex())))
+            && ["h", "d"].iter().all(|name| {
+                filter
+                    .get(format!("#{name}"))
+                    .and_then(Value::as_array)
+                    .is_none_or(|values| {
+                        tag(event, name).is_some_and(|v| values.contains(&json!(v)))
+                    })
+            });
+        if matches {
+            events.push(event.clone());
+        }
+    }
+    events.truncate(filter["limit"].as_u64().unwrap_or(2) as usize);
+    (StatusCode::OK, Json(json!(events)))
+}
+async fn submit(
+    State(store): State<Shared>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> (StatusCode, Json<Value>) {
+    auth(&headers, &body);
+    let event: Event = serde_json::from_slice(&body).unwrap();
+    event.verify().unwrap();
+    let mut store = store.lock().unwrap();
+    store.submissions.push(event.clone());
+    if event.kind.as_u16() == 9 {
+        // Model relay timestamp admission. A stale unaccepted saved plan must
+        // be refreshed, while an accepted old event is reused by exact lookup.
+        if nostr::Timestamp::now()
+            .as_secs()
+            .abs_diff(event.created_at.as_secs())
+            > 900
+        {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!(["OK", event.id, false, "timestamp"])),
+            );
+        }
+        let record_count = store
+            .heads
+            .values()
+            .filter_map(|id| store.events.get(id))
+            .filter(|record| tag(record, "message") == Some(event.id.to_hex().as_str()))
+            .count();
+        assert!(
+            record_count >= 1,
+            "every message requires a durable plan before publication"
+        );
+        if tag(&event, "e").is_some() {
+            assert!(
+                record_count >= 2,
+                "delivery and body fingerprint plans precede replies"
+            );
+            if store.fail_reply > 0 {
+                store.fail_reply -= 1;
+                return (StatusCode::SERVICE_UNAVAILABLE, Json(json!({})));
+            }
+        }
+    }
+    if let Some(d) = tag(&event, "d") {
+        store.heads.insert(d.to_owned(), event.id.to_hex());
+    }
+    store.events.insert(event.id.to_hex(), event.clone());
+    if event.kind.as_u16() == 9 && tag(&event, "e").is_none() && store.lose_root_response {
+        store.lose_root_response = false;
+        return (StatusCode::SERVICE_UNAVAILABLE, Json(json!({})));
+    }
+    (StatusCode::OK, Json(json!(["OK", event.id, true, ""])))
+}
+async fn external() -> (String, Shared, tokio::task::JoinHandle<()>) {
+    let store = Shared::default();
+    let router = Router::new()
+        .route("/query", post(query))
+        .route("/events", post(submit))
+        .with_state(store.clone());
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let task = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
+    (format!("http://{addr}"), store, task)
+}
+fn config(url: &str) -> Config {
+    serde_json::from_value(json!({"relay_url":url,"github_repos":{"voltagecloud/buzz":CHANNEL},"linear_teams":{"TEAM":CHANNEL},"linear_webhooks":{"HOOK":"0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a11"},"linear_labels":{"LABEL-AGENT":"agent"},"slack_channels":{"T123/C123":CHANNEL},"slack_workspace_url":"https://voltage.slack.com"})).unwrap()
+}
+fn app(config: Config, keys: &Keys) -> Router {
+    crate::http::router(
+        config.clone(),
+        Secrets {
+            github: Some(SECRET.into()),
+            linear: Some(SECRET.into()),
+            slack: Some(SECRET.into()),
+        },
+        Relay::new(&config.relay_url, keys.clone()).unwrap(),
+    )
+}
+fn mac(body: &[u8]) -> String {
+    let mut h = Hmac::<Sha256>::new_from_slice(SECRET.as_bytes()).unwrap();
+    h.update(body);
+    hex::encode(h.finalize().into_bytes())
+}
+async fn request(
+    app: &Router,
+    provider: &str,
+    payload: &Value,
+    id: &str,
+    bad_signature: bool,
+    age: u64,
+    event: &str,
+) -> (StatusCode, Value) {
+    let body = payload.to_string();
+    let time = nostr::Timestamp::now()
+        .as_secs()
+        .saturating_sub(age)
+        .to_string();
+    let mut req = axum::http::Request::builder()
+        .method("POST")
+        .uri(format!("/ingest/{provider}"))
+        .header("content-type", "application/json");
+    let hash = if bad_signature {
+        "00".repeat(32)
+    } else if provider == "slack" {
+        mac(format!("v0:{time}:{body}").as_bytes())
+    } else {
+        mac(body.as_bytes())
+    };
+    req = match provider {
+        "github" => req
+            .header("x-hub-signature-256", format!("sha256={hash}"))
+            .header("x-github-delivery", id)
+            .header("x-github-event", event),
+        "linear" => req
+            .header("linear-signature", hash)
+            .header("linear-delivery", id),
+        _ => req
+            .header("x-slack-signature", format!("v0={hash}"))
+            .header("x-slack-request-timestamp", time),
+    };
+    let response = app
+        .clone()
+        .oneshot(req.body(axum::body::Body::from(body)).unwrap())
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = axum::body::to_bytes(response.into_body(), 1024 * 1024)
+        .await
+        .unwrap();
+    (
+        status,
+        serde_json::from_slice(&body).unwrap_or_else(|_| json!(String::from_utf8_lossy(&body))),
+    )
+}
+fn messages(store: &Shared) -> Vec<Event> {
+    store
+        .lock()
+        .unwrap()
+        .events
+        .values()
+        .filter(|e| e.kind.as_u16() == 9)
+        .cloned()
+        .collect()
+}
+fn github() -> Value {
+    json!({"action":"opened","repository":{"full_name":"voltagecloud/buzz"},"pull_request":{"id":42,"number":7,"title":"A useful change","body":"Please review","state":"open","labels":[{"name":"agent"}]},"sender":{"login":"ada"}})
+}
+
+#[tokio::test]
+async fn github_retries_preserve_plans_and_do_not_duplicate_threads_or_effects() {
+    let (url, store, server) = external().await;
+    let keys = Keys::generate();
+    let cfg = config(&url);
+    let first = app(cfg.clone(), &keys);
+    let payload = github();
+    assert_eq!(
+        request(
+            &first,
+            "github",
+            &payload,
+            "delivery-1",
+            true,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::UNAUTHORIZED
+    );
+    assert!(messages(&store).is_empty());
+    assert_eq!(
+        request(
+            &first,
+            "github",
+            &json!({"zen":"A signed ping","hook":{},"hook_id":1}),
+            "ping-1",
+            false,
+            0,
+            "ping"
+        )
+        .await
+        .1["ignored"],
+        true
+    );
+    store.lock().unwrap().lose_root_response = true;
+    assert_eq!(
+        request(
+            &first,
+            "github",
+            &payload,
+            "delivery-1",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::SERVICE_UNAVAILABLE
+    );
+    assert_eq!(messages(&store).len(), 1);
+    let restarted = app(cfg, &keys);
+    store.lock().unwrap().fail_reply = 1;
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &payload,
+            "delivery-1",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::SERVICE_UNAVAILABLE
+    );
+    assert!(store
+        .lock()
+        .unwrap()
+        .heads
+        .keys()
+        .any(|k| k.starts_with("fingerprint:")));
+    // A pending plan can outlive relay timestamp admission during an outage.
+    // Replace its event with an old signed equivalent in both journal aliases.
+    let mut old_id = None;
+    {
+        let mut locked = store.lock().unwrap();
+        let records: Vec<_> = locked
+            .heads
+            .values()
+            .filter_map(|id| locked.events.get(id))
+            .cloned()
+            .collect();
+        for record in records {
+            let mut plan: Value =
+                serde_json::from_slice(&STANDARD.decode(&record.content).unwrap()).unwrap();
+            let saved: Event = serde_json::from_value(plan["event"].clone()).unwrap();
+            if tag(&saved, "e").is_none() {
+                continue;
+            }
+            let old = nostr::EventBuilder::new(saved.kind, saved.content)
+                .tags(saved.tags.to_vec())
+                .custom_created_at(nostr::Timestamp::from(
+                    nostr::Timestamp::now().as_secs() - 1000,
+                ))
+                .sign_with_keys(&keys)
+                .unwrap();
+            old_id = Some(old.id.to_hex());
+            plan["event"] = serde_json::to_value(&old).unwrap();
+            let mut tags = record.tags.to_vec();
+            tags.retain(|tag| tag.as_slice()[0] != "message");
+            tags.push(nostr::Tag::parse(["message", &old.id.to_hex()]).unwrap());
+            let replaced = nostr::EventBuilder::new(
+                record.kind,
+                STANDARD.encode(serde_json::to_vec(&plan).unwrap()),
+            )
+            .tags(tags)
+            .sign_with_keys(&keys)
+            .unwrap();
+            locked.heads.insert(
+                tag(&replaced, "d").unwrap().to_owned(),
+                replaced.id.to_hex(),
+            );
+            locked.events.insert(replaced.id.to_hex(), replaced);
+        }
+        locked.fail_event_lookup = old_id.clone();
+    }
+    assert!(old_id.is_some());
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &payload,
+            "delivery-1",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::SERVICE_UNAVAILABLE
+    );
+    assert_eq!(
+        messages(&store).len(),
+        1,
+        "lookup failure must not refresh or publish an ambiguous plan"
+    );
+
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &payload,
+            "delivery-1",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::OK
+    );
+    let before = messages(&store);
+    assert_eq!(before.len(), 2);
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &payload,
+            "delivery-1",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::OK
+    );
+    // Delivery headers are unsigned. Replaying a valid signed body under a
+    // changed header must retain the same display event.
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &payload,
+            "different-header",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::OK
+    );
+    assert_eq!(messages(&store).len(), 2);
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &payload,
+            "different-header",
+            false,
+            0,
+            "issue_comment"
+        )
+        .await
+        .0,
+        StatusCode::BAD_REQUEST
+    );
+    store.lock().unwrap().fail_queries = 1;
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &payload,
+            "delivery-1",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::SERVICE_UNAVAILABLE
+    );
+    assert_eq!(messages(&store).len(), 2);
+    let mut review = payload.clone();
+    review["action"] = json!("submitted");
+    review["review"] = json!({"id":77,"state":"approved","body":"Looks good"});
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &review,
+            "review-1",
+            false,
+            0,
+            "pull_request_review"
+        )
+        .await
+        .0,
+        StatusCode::OK
+    );
+    let checks = json!({"action":"completed","repository":{"full_name":"voltagecloud/buzz"},"sender":{"login":"ci"},"check_run":{"id":88,"name":"Rust tests","head_sha":"abc123","conclusion":"failure","pull_requests":[{"number":7}],"output":{"summary":"A test failed"}}});
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &checks,
+            "check-1",
+            false,
+            0,
+            "check_run"
+        )
+        .await
+        .0,
+        StatusCode::OK
+    );
+    let current = messages(&store);
+    assert!(current
+        .iter()
+        .any(|event| event.content.contains("approved")));
+    assert!(current
+        .iter()
+        .any(|event| event.content.contains("Rust tests") && event.content.contains("failure")));
+    assert_eq!(
+        current
+            .iter()
+            .filter(|event| tag(event, "e").is_none())
+            .count(),
+        1
+    );
+    let mut too_many = payload.clone();
+    too_many["pull_request"]["labels"] = json!((0..17)
+        .map(|i| json!({"name":format!("label-{i}")}))
+        .collect::<Vec<_>>());
+    let prior = messages(&store).len();
+    assert_eq!(
+        request(
+            &restarted,
+            "github",
+            &too_many,
+            "too-many-labels",
+            false,
+            0,
+            "pull_request"
+        )
+        .await
+        .0,
+        StatusCode::BAD_REQUEST
+    );
+    assert_eq!(messages(&store).len(), prior);
+    server.abort();
+}
+
+#[tokio::test]
+async fn linear_and_slack_only_publish_authenticated_mapped_public_work() {
+    let (url, store, server) = external().await;
+    let keys = Keys::generate();
+    let router = app(config(&url), &keys);
+    let now = nostr::Timestamp::now().as_secs() * 1000;
+    let issue = json!({"type":"Issue","action":"create","webhookTimestamp":now,"webhookId":"HOOK","url":"https://linear.app/issue/ENG-7","actor":{"name":"Ada"},"data":{"id":"issue-7","identifier":"ENG-7","teamId":"TEAM","title":"Fix it","description":"Details"}});
+    assert_eq!(
+        request(&router, "linear", &issue, "linear-1", true, 0, "")
+            .await
+            .0,
+        StatusCode::UNAUTHORIZED
+    );
+    assert_eq!(
+        request(&router, "linear", &issue, "linear-1", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    let mut labeled = issue.clone();
+    labeled["action"] = json!("update");
+    labeled["data"]["labelIds"] = json!(["LABEL-AGENT"]);
+    labeled["updatedFrom"] = json!({"labelIds":null});
+    assert_eq!(
+        request(&router, "linear", &labeled, "linear-label", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    let label_event = messages(&store)
+        .into_iter()
+        .find(|e| tag(e, "ext-action") == Some("issue.labeled"))
+        .unwrap();
+    assert_eq!(tag(&label_event, "ext-added-labels"), Some("agent"));
+    let mut redelivery = issue.clone();
+    redelivery["webhookTimestamp"] = json!(now + 1000);
+    assert_eq!(
+        request(&router, "linear", &redelivery, "linear-1", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    assert_eq!(
+        messages(&store).len(),
+        3,
+        "transport timestamps must not duplicate a Linear occurrence"
+    );
+    let comment = json!({"type":"Comment","action":"create","webhookTimestamp":now,"webhookId":"HOOK","url":"https://linear.app/issue/ENG-7#comment-1","actor":{"name":"Ada"},"data":{"id":"comment-1","issueId":"issue-7","body":"An update"}});
+    assert_eq!(
+        request(&router, "linear", &comment, "linear-2", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    assert_eq!(messages(&store).len(), 4); // one anchor, three ordinary replies
+    let mut stale = issue.clone();
+    stale["webhookTimestamp"] = json!(now - 600_000);
+    assert_eq!(
+        request(&router, "linear", &stale, "linear-stale", false, 0, "")
+            .await
+            .0,
+        StatusCode::UNAUTHORIZED
+    );
+    let challenge = json!({"type":"url_verification","challenge":"verify-this"});
+    assert_eq!(
+        request(&router, "slack", &challenge, "", true, 0, "")
+            .await
+            .0,
+        StatusCode::UNAUTHORIZED
+    );
+    assert_eq!(
+        request(&router, "slack", &challenge, "", false, 0, "")
+            .await
+            .1["challenge"],
+        "verify-this"
+    );
+    let mut message = json!({"type":"event_callback","team_id":"T123","event_id":"Ev1","event":{"type":"message","channel_type":"im","channel":"C123","ts":"1700000000.000001","user":"U123","text":"private content"}});
+    assert_eq!(
+        request(&router, "slack", &message, "", false, 0, "")
+            .await
+            .1["occurrences"],
+        0
+    );
+    assert_eq!(messages(&store).len(), 4);
+    message["event"]["channel_type"] = json!("channel");
+    message["event"]["text"] = json!("public discussion");
+    assert_eq!(
+        request(&router, "slack", &message, "", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    assert_eq!(messages(&store).len(), 6);
+    assert_eq!(
+        request(&router, "slack", &message, "", false, 600, "")
+            .await
+            .0,
+        StatusCode::UNAUTHORIZED
+    );
+    let mut huge = message;
+    huge["event"]["text"] = json!("x".repeat(1024 * 1024));
+    assert_eq!(
+        request(&router, "slack", &huge, "", false, 0, "").await.0,
+        StatusCode::PAYLOAD_TOO_LARGE
+    );
+    // An existing issue without a saved route can use an explicitly mapped
+    // signed webhook ID. A known route above must retain the team channel.
+    let mut orphan = comment.clone();
+    orphan["data"]["id"] = json!("orphan-comment");
+    orphan["data"]["issueId"] = json!("never-seen-issue");
+    assert_eq!(
+        request(&router, "linear", &orphan, "linear-orphan", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    assert!(messages(&store)
+        .iter()
+        .any(|event| tag(event, "h") == Some("0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a11")));
+    server.abort();
+}
+
+#[test]
+fn endpoints_and_provider_mappings_fail_closed() {
+    let mut cfg = config("https://buzz.voltage.cloud");
+    cfg.validate().unwrap();
+    cfg.relay_url = "http://public.example".into();
+    assert!(cfg.validate().is_err());
+    cfg.relay_url = "https://buzz.voltage.cloud".into();
+    cfg.slack_channels
+        .insert("T123/GPRIVATE".into(), CHANNEL.parse().unwrap());
+    assert!(cfg.validate().is_err());
+}
diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs
index acff4b7c3eb..8b3a1b57f06 100644
--- a/crates/buzz-relay/src/handlers/ingest.rs
+++ b/crates/buzz-relay/src/handlers/ingest.rs
@@ -551,7 +551,7 @@ fn map_push_accept_error(error: super::push_lease::AcceptError) -> IngestError {
 /// Returns `Err` for unknown kinds — the relay rejects them.
 fn required_scope_for_kind(kind: u32, event: &Event) -> Result<Scope, &'static str> {
     match kind {
-        buzz_core::kind::KIND_AGENT_CHANNEL_GRANT => Ok(Scope::MessagesWrite),
+        buzz_core::kind::KIND_AGENT_CHANNEL_GRANT | buzz_core::kind::KIND_INTEGRATION_RECORD => Ok(Scope::MessagesWrite),
         KIND_PROFILE => Ok(Scope::UsersWrite),
         KIND_TEXT_NOTE | KIND_LONG_FORM | buzz_core::kind::KIND_ARTIFACT => Ok(Scope::MessagesWrite),
         KIND_CONTACT_LIST | KIND_READ_STATE | KIND_USER_STATUS | KIND_AGENT_ENGRAM
@@ -836,6 +836,7 @@ pub(crate) fn requires_h_channel_scope(kind: u32) -> bool {
     matches!(
         kind,
         buzz_core::kind::KIND_AGENT_CHANNEL_GRANT
+            | buzz_core::kind::KIND_INTEGRATION_RECORD
             | KIND_STREAM_MESSAGE
             | KIND_STREAM_MESSAGE_V2
             | KIND_STREAM_MESSAGE_EDIT
@@ -4263,6 +4264,7 @@ mod postgres_tests {
     #[test]
     fn channel_scoped_content_kinds_require_h_tags() {
         for kind in [
+            buzz_core::kind::KIND_INTEGRATION_RECORD,
             KIND_STREAM_MESSAGE,
             KIND_STREAM_MESSAGE_DIFF,
             KIND_CANVAS,
diff --git a/crates/buzz-workflow/src/integration_tests.rs b/crates/buzz-workflow/src/integration_tests.rs
new file mode 100644
index 00000000000..4fe2e2a2878
--- /dev/null
+++ b/crates/buzz-workflow/src/integration_tests.rs
@@ -0,0 +1,101 @@
+//! Integration workflows use the production trigger extraction and filter path.
+
+use super::*;
+use nostr::{EventBuilder, Keys, Kind, Tag};
+
+#[tokio::test]
+async fn trusted_integration_messages_drive_filtered_thread_workflows() {
+    let publisher = Keys::generate();
+    let other = Keys::generate();
+    let yaml = format!(
+        r#"
+name: Route integration work
+trigger:
+  on: integration_event
+  source: linear
+  action: issue.update
+  authors: ["{}"]
+  filter: 'str_contains(trigger_labels, "agent")'
+steps:
+  - id: delegate
+    action: send_message
+    reply_in_thread: true
+    text: "Review {{{{trigger.url}}}}"
+"#,
+        publisher.public_key().to_hex()
+    );
+    let (def, _) = schema::parse_yaml(&yaml).unwrap();
+    let make = |keys: &Keys, source: &str, action: &str, labels: &str| {
+        let event = EventBuilder::new(Kind::Custom(9), "Issue changed")
+            .tags([
+                Tag::parse(["ext-source", source]).unwrap(),
+                Tag::parse(["ext-action", action]).unwrap(),
+                Tag::parse(["ext-labels", labels]).unwrap(),
+                Tag::parse(["ext-url", "https://linear.app/team/issue/ABC-1"]).unwrap(),
+                Tag::parse(["ext-author", "spoof"]).unwrap(),
+            ])
+            .sign_with_keys(keys)
+            .unwrap();
+        buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()))
+    };
+    for (keys, source, action, labels, expected) in [
+        (&publisher, "linear", "issue.update", "agent,bug", true),
+        (&other, "linear", "issue.update", "agent", false),
+        (&publisher, "github", "issue.update", "agent", false),
+        (&publisher, "linear", "issue.create", "agent", false),
+        (&publisher, "linear", "", "agent", false),
+        (&publisher, "", "issue.update", "agent", false),
+        (&publisher, "linear", "issue.update", "bug", false),
+    ] {
+        let stored = make(keys, source, action, labels);
+        let ctx = build_trigger_context(&stored);
+        assert_eq!(ctx.author, keys.public_key().to_hex());
+        assert_eq!(ctx.message_id, stored.event.id.to_hex());
+        assert!(trigger_matches_event(&def.trigger, 9));
+        assert!(!trigger_matches_event(
+            &def.trigger,
+            buzz_core::kind::KIND_INTEGRATION_RECORD
+        ));
+        assert_eq!(
+            should_fire_workflow(&def, &ctx, Uuid::new_v4()).await,
+            expected,
+            "{source}/{action} by {} with {labels}",
+            ctx.author
+        );
+        assert_eq!(
+            executor::resolve_template("{{trigger.author}} {{trigger.url}}", &ctx, &HashMap::new())
+                .unwrap(),
+            format!(
+                "{} https://linear.app/team/issue/ABC-1",
+                keys.public_key().to_hex()
+            )
+        );
+    }
+    // A source-wide workflow still requires a nonempty action tag. Uppercase
+    // configured hex keys identify the same trusted publisher.
+    let mut any_action = def.clone();
+    if let TriggerDef::IntegrationEvent {
+        action, authors, ..
+    } = &mut any_action.trigger
+    {
+        *action = None;
+        authors[0] = authors[0].to_uppercase();
+    }
+    for (action, expected) in [("", false), ("issue.create", true)] {
+        let stored = make(&publisher, "linear", action, "agent");
+        assert_eq!(
+            should_fire_workflow(&any_action, &build_trigger_context(&stored), Uuid::new_v4())
+                .await,
+            expected
+        );
+    }
+
+    // Missing/untrusted publishers cannot be admitted through YAML validation.
+    for replacement in ["[]", "[bad-key]"] {
+        let invalid = yaml.replace(
+            &format!("[\"{}\"]", publisher.public_key().to_hex()),
+            replacement,
+        );
+        assert!(schema::parse_yaml(&invalid).is_err());
+    }
+}
diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs
index 9dabc5732a7..01df518b90c 100644
--- a/crates/buzz-workflow/src/lib.rs
+++ b/crates/buzz-workflow/src/lib.rs
@@ -977,10 +977,36 @@ async fn should_fire_workflow(
         }
     }
 
+    if let TriggerDef::IntegrationEvent {
+        source,
+        action,
+        authors,
+        ..
+    } = &def.trigger
+    {
+        // Source tags are signer-controlled. Only a named publisher may drive
+        // this workflow; a channel member cannot spoof an integration source.
+        if !authors
+            .iter()
+            .any(|author| author.eq_ignore_ascii_case(&trigger_ctx.author))
+            || trigger_ctx.webhook_fields.get("source") != Some(source)
+            || trigger_ctx
+                .webhook_fields
+                .get("action")
+                .is_none_or(String::is_empty)
+            || action
+                .as_ref()
+                .is_some_and(|expected| trigger_ctx.webhook_fields.get("action") != Some(expected))
+        {
+            return false;
+        }
+    }
+
     let filter = match &def.trigger {
         TriggerDef::MessagePosted { filter }
         | TriggerDef::ReactionAdded { filter, .. }
-        | TriggerDef::DiffPosted { filter } => filter.as_ref(),
+        | TriggerDef::DiffPosted { filter }
+        | TriggerDef::IntegrationEvent { filter, .. } => filter.as_ref(),
         TriggerDef::Schedule { .. } | TriggerDef::Webhook => None,
     };
     if let Some(expr) = filter {
@@ -1075,7 +1101,9 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge
         emoji,
         message_id,
         is_reply: event_is_reply(&event.event),
-        webhook_fields: HashMap::new(),
+        webhook_fields: buzz_core::integration::ext_fields(&event.event.tags)
+            .into_iter()
+            .collect(),
     }
 }
 
@@ -1120,7 +1148,9 @@ fn owner_authority_allows(role: Option<&str>, needs_elevated: bool) -> bool {
 fn trigger_matches_event(trigger: &TriggerDef, kind_u32: u32) -> bool {
     use buzz_core::kind::{KIND_REACTION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_DIFF};
     match trigger {
-        TriggerDef::MessagePosted { .. } => kind_u32 == KIND_STREAM_MESSAGE,
+        TriggerDef::MessagePosted { .. } | TriggerDef::IntegrationEvent { .. } => {
+            kind_u32 == KIND_STREAM_MESSAGE
+        }
         TriggerDef::ReactionAdded { .. } => kind_u32 == KIND_REACTION,
         TriggerDef::DiffPosted { .. } => kind_u32 == KIND_STREAM_MESSAGE_DIFF,
         // Schedule and Webhook triggers are not fired by channel events.
@@ -1128,6 +1158,9 @@ fn trigger_matches_event(trigger: &TriggerDef, kind_u32: u32) -> bool {
     }
 }
 
+#[cfg(test)]
+mod integration_tests;
+
 #[cfg(test)]
 mod postgres_tests {
     use super::*;
diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs
index 0e8dfdb52ef..ddc395f8eea 100644
--- a/crates/buzz-workflow/src/schema.rs
+++ b/crates/buzz-workflow/src/schema.rs
@@ -57,6 +57,19 @@ pub enum TriggerDef {
         #[serde(default)]
         filter: Option<String>,
     },
+    /// Fires on signed integration messages from explicitly trusted publishers.
+    IntegrationEvent {
+        /// Provider identifier carried in the event's `ext-source` tag.
+        source: String,
+        /// Optional exact normalized action, such as `issue.update`.
+        #[serde(default)]
+        action: Option<String>,
+        /// Trusted integration publisher public keys (hex), never an empty list.
+        authors: Vec<String>,
+        /// Optional evalexpr expression over `trigger_*` metadata fields.
+        #[serde(default)]
+        filter: Option<String>,
+    },
     /// Fires on a cron schedule.
     Schedule {
         /// Cron expression (UTC). Mutually exclusive with `interval`.
@@ -192,6 +205,33 @@ impl WorkflowDef {
                 && id.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
         };
 
+        if let TriggerDef::IntegrationEvent {
+            source,
+            action,
+            authors,
+            ..
+        } = &self.trigger
+        {
+            let valid_name = |name: &str| {
+                !name.is_empty()
+                    && name.len() <= 128
+                    && name
+                        .bytes()
+                        .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'.'))
+            };
+            if !valid_name(source)
+                || action.as_ref().is_some_and(|a| !valid_name(a))
+                || authors.is_empty()
+                || authors.len() > 32
+                || authors
+                    .iter()
+                    .any(|a| a.len() != 64 || nostr::PublicKey::from_hex(a).is_err())
+            {
+                return Err(WorkflowError::InvalidDefinition(
+                    "integration_event needs a source, an optional action, and 1–32 valid hex publisher keys".into()
+                ));
+            }
+        }
         let mut seen_ids: HashSet<&str> = HashSet::new();
         for step in &self.steps {
             if step.id.trim().is_empty() {
@@ -221,6 +261,7 @@ impl WorkflowDef {
             TriggerDef::MessagePosted { .. }
                 | TriggerDef::ReactionAdded { .. }
                 | TriggerDef::DiffPosted { .. }
+                | TriggerDef::IntegrationEvent { .. }
         );
         if !trigger_has_message {
             for step in &self.steps {
diff --git a/docs/nips/NIP-IG.md b/docs/nips/NIP-IG.md
new file mode 100644
index 00000000000..62da432fa70
--- /dev/null
+++ b/docs/nips/NIP-IG.md
@@ -0,0 +1,75 @@
+# NIP-IG: integration occurrences and delivery plans
+
+An integration publisher mirrors external activity as signed, channel-scoped
+kind 9 messages. The relay applies ordinary membership, threading, search, and
+thread-counter rules. The relay contains no provider-specific webhook handlers.
+
+| Event | Purpose |
+| --- | --- |
+| Kind 9 root | A stable external thread anchor with `h`, `ext-source`, and `ext-thread`. It has no `ext-action`. |
+| Kind 9 reply | One occurrence, with NIP-10 root reference and typed `ext-*` metadata. |
+| Kind 30623 | Parameterized replaceable write-ahead plan. `d` identifies the journal coordinate; `h` scopes access; `message` names the planned kind 9 event. |
+
+A record's content is standard-base64 JSON:
+
+```json
+{"version":1,"payload_sha256":"<canonical payload hash or null>","event":{"id":"...","pubkey":"...","created_at":0,"kind":9,"tags":[],"content":"...","sig":"..."}}
+```
+
+The decoded plan is bounded to 24 KiB. Base64 keeps duplicated message words
+out of the record's lexical representation; it provides no encryption.
+Normal conversation search selects kind 9. Records remain readable to channel
+members and must receive the same retention controls as their messages.
+
+Coordinates distinguish thread anchors, provider delivery IDs, payload
+fingerprints, and Linear issue routing. Delivery and fingerprint records are
+committed before the display message. The fingerprint prevents replay under
+an altered delivery header, which raw-body provider signatures do not bind.
+Linear transport `webhookTimestamp` is excluded from canonical occurrence
+identity so a refreshed delivery timestamp does not duplicate its message.
+
+Before replay, the publisher verifies the record and planned event signatures,
+its publisher identity, channel, coordinate, payload hash, and `message` ID.
+It reads records and exact message IDs with `consistency: strong`. Lookup
+failure propagates; it never authorizes a new plan. An accepted event is reused.
+An unaccepted plan outside the relay timestamp window receives a fresh timestamp
+only after an authoritative empty ID lookup; its new plan precedes publication.
+
+Run one publisher replica with a Recreate deployment. The current implementation
+serializes requests in that process. It does not coordinate competing writers
+across replicas. Preserve journal records while webhook delivery can be retried.
+Deleting journals or rotating the publisher key requires an operator-managed
+cutover; neither operation preserves automatic deduplication.
+
+## Source tags and workflow trust
+
+Occurrences include `ext-source`, `ext-action`, `ext-id`, `ext-url`, `ext-actor`,
+`ext-thread`, and `ext-key`. Optional fields include `ext-repo`, `ext-team`,
+`ext-ref`, `ext-title`, `ext-state`, `ext-labels`, `ext-added-labels`,
+`ext-conclusion`, `ext-review-state`, and `ext-channel`.
+
+Repeated `ext-labels` / `ext-added-labels` tags describe individual labels.
+Consumers fold dashes to underscores and join repeated fields with commas.
+These fields are publisher-controlled data, not relay attestations.
+
+`integration_event` workflows require an explicit, nonempty `authors` allowlist
+of trusted publisher keys. They match only kind 9 occurrence replies. Roots
+lack `ext-action`, and journal records never trigger the same workflow again.
+
+```yaml
+name: linear-agent-work
+trigger:
+  on: integration_event
+  source: linear
+  authors:
+    - "<64-character integration publisher public key>"
+  filter: 'str_contains("," + trigger_added_labels + ",", ",agent,")'
+steps:
+  - id: delegate
+    action: send_message
+    reply_in_thread: true
+    text: "nostr:<agent-npub> Review {{trigger.url}} and propose a plan."
+```
+
+Use the repository workflow schema for action fields and mention targeting.
+Manage these workflows through `buzz workflows` from Paseo or a Buzz client.
diff --git a/mobile/lib/shared/relay/nostr_models.dart b/mobile/lib/shared/relay/nostr_models.dart
index 60bed113112..fd799b994e6 100644
--- a/mobile/lib/shared/relay/nostr_models.dart
+++ b/mobile/lib/shared/relay/nostr_models.dart
@@ -38,6 +38,7 @@ abstract final class EventKind {
   static const eventReminder = 30300;
   static const userStatus = 30315;
   static const dmVisibility = 30622;
+  static const integrationRecord = 30623;
   static const streamMessageV2 = 40002;
   static const channelThreadSummary = 39005;
   static const channelWindowBounds = 39006;
diff --git a/voltage/images/README.md b/voltage/images/README.md
index a996b3d29fa..7898fef6d9b 100644
--- a/voltage/images/README.md
+++ b/voltage/images/README.md
@@ -11,10 +11,11 @@ The same workflow also builds the Buzz relay image from the upstream root `Docke
 | [buzz-agent](buzz-agent/README.md) | Remote developer agent Sandbox: the `sprig` harness, Codex and Claude ACP adapters, and developer tools. It also runs as the `buzz-agent-gate` DaemonSet. | `voltage-buzz-agent-staging` |
 | [buzz-sandbox-reaper](buzz-sandbox-reaper/Dockerfile) | In-cluster reaper for ended and orphaned agent Sandboxes (`buzz-sandbox-reaper reconcile`) | `voltage-buzz-sandbox-reaper-staging` |
 | [buzz-relay](#the-relay-image) | Buzz relay, admin, and pair relay binaries with the web bundles, from the root `Dockerfile` | `buzz-relay` |
+| [buzz-ingest](../../crates/buzz-ingest/README.md) | Verified GitHub, Linear, and public Slack webhooks, stored as Buzz threads | `buzz-ingest` (provision before publication) |
 
 Each image compiles the Buzz crates of the commit it is built from.
 The build context is the repository root.
-For the two agent images, BuildKit reads the `Dockerfile.dockerignore` next to each Dockerfile instead of the root `.dockerignore`.
+For the Voltage images, BuildKit reads the `Dockerfile.dockerignore` next to each Dockerfile instead of the root `.dockerignore`.
 These files keep only the Cargo workspace and, for `buzz-agent`, the files that its Dockerfile copies from `voltage/images/buzz-agent`.
 
 ## Publishing
@@ -33,7 +34,9 @@ Its first job selects only the images whose inputs changed; manual runs build al
 
 The voltage-terraform `v2/environments/staging` root creates the two agent ECR repositories; `v2/environments/prod` creates `buzz-relay`.
 Publishing assumes the `github-buzz-ecr-push` IAM role (voltage-terraform `v2/environments/github`).
-The role trusts only `repo:voltagecloud/buzz:ref:refs/heads/main` and can push to and read only these three repositories.
+The role trusts only `repo:voltagecloud/buzz:ref:refs/heads/main`.
+Provision `buzz-ingest` in prod and add it to this role before publishing the new sidecar.
+The existing three repository grants remain unchanged by this source PR.
 The repository secret `AWS_ACCOUNT_ID` supplies the account ID, so the account ID is not in the repository and is masked in the public logs.
 
 ## Deploying a digest
diff --git a/voltage/images/buzz-ingest/Dockerfile b/voltage/images/buzz-ingest/Dockerfile
new file mode 100644
index 00000000000..2e68cca4714
--- /dev/null
+++ b/voltage/images/buzz-ingest/Dockerfile
@@ -0,0 +1,16 @@
+# syntax=docker/dockerfile:1.7
+FROM rust:1.95-alpine3.22@sha256:064dfc925d68d1a63f4fd2871bd7dc6e6ea56692989a487185855d62885d90aa AS builder
+RUN apk add --no-cache build-base cmake git musl-dev perl pkgconf
+WORKDIR /build
+ENV CARGO_INCREMENTAL=0 CARGO_BUILD_JOBS=2
+COPY . .
+RUN --mount=type=cache,target=/usr/local/cargo/registry \
+    --mount=type=cache,target=/usr/local/cargo/git \
+    cargo build --locked --release -p buzz-ingest \
+    && strip target/release/buzz-ingest
+
+FROM gcr.io/distroless/static-debian12:nonroot@sha256:afa5c872c891853ca7fcf1f12c3edb23f7eeef36189728842dd51042ff57f7ab
+COPY --from=builder /build/target/release/buzz-ingest /usr/local/bin/buzz-ingest
+USER 65532:65532
+EXPOSE 5010
+ENTRYPOINT ["/usr/local/bin/buzz-ingest"]
diff --git a/voltage/images/buzz-ingest/Dockerfile.dockerignore b/voltage/images/buzz-ingest/Dockerfile.dockerignore
new file mode 100644
index 00000000000..e36621e8ee2
--- /dev/null
+++ b/voltage/images/buzz-ingest/Dockerfile.dockerignore
@@ -0,0 +1,82 @@
+# Build outputs
+target/
+**/target/
+desktop/src-tauri/target/
+
+# Native non-relay UI builds (not part of the public image)
+desktop/
+mobile/
+
+# Node artifacts (built inside the image; host outputs would invalidate cache)
+node_modules/
+**/node_modules/
+web/dist/
+
+# VCS, IDE, scratch
+.git/
+.github/
+.gitignore
+.gitattributes
+.vscode/
+.idea/
+.scratch/
+
+# Local environment / secrets — never want these in the build context
+.env
+.env.*
+!.env.example
+*.pem
+*.key
+secrets/
+
+# OS junk
+.DS_Store
+Thumbs.db
+
+# CI artifacts and tooling that the image doesn't need
+.cache/
+coverage/
+dist/
+playwright-report/
+test-results/
+bin/
+sidecars/
+
+# Docs the image doesn't ship. examples/ stays — examples/countdown-bot is
+# a Cargo workspace member, so cargo-chef needs its manifest. Markdown files
+# inside crates (Cargo.toml `readme = ...`) likewise need to be present to
+# avoid manifest warnings; the savings from excluding them are negligible.
+docs/
+
+# Old Dockerfile noise (this Dockerfile is THE source of truth now)
+docker-compose.yml
+docker-compose.*.yml
+prometheus.yml
+
+# --- Voltage agent images ---------------------------------------------------
+# The lines above copy the root .dockerignore. The lines below also exclude
+# what `cargo build --locked` does not read. Keep the workspace manifest and
+# lockfile, .cargo/, rust-toolchain.toml, every root workspace member (crates/,
+# examples/), and scripts/ and test-fixtures/, which crates include at
+# compile time.
+web/
+apps/
+admin-web/
+benchmarks/
+deploy/
+migrations/
+perf/
+PLANS/
+patches/
+schema/
+script/
+.agents/
+.claude/
+.codex/
+.config/
+.goose/
+.hermit/
+.intersect/
+.release/
+**/__pycache__/
+voltage/
diff --git a/voltage/images/buzz-ingest/test-image.sh b/voltage/images/buzz-ingest/test-image.sh
new file mode 100755
index 00000000000..36d9158c8ce
--- /dev/null
+++ b/voltage/images/buzz-ingest/test-image.sh
@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+set -euo pipefail
+image=${1:?usage: test-image.sh IMAGE}
+scratch=$(mktemp -d)
+container=
+cleanup() {
+  if [[ -n "$container" ]]; then docker rm -f "$container" >/dev/null; fi
+  rm -rf "$scratch"
+}
+trap cleanup EXIT
+chmod 755 "$scratch"
+cat > "$scratch/config.json" <<'JSON'
+{"relay_url":"https://127.0.0.1:9","github_repos":{"example/repo":"0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a10"}}
+JSON
+chmod 444 "$scratch/config.json"
+container=$(docker run -d --read-only --cap-drop ALL --security-opt no-new-privileges \
+  --user 65532:65532 --memory 128m --cpus 1 \
+  -p 127.0.0.1::5010 -v "$scratch/config.json:/etc/buzz-ingest/config.json:ro" \
+  -e BUZZ_INGEST_CONFIG=/etc/buzz-ingest/config.json \
+  -e BUZZ_INGEST_PRIVATE_KEY=1111111111111111111111111111111111111111111111111111111111111111 \
+  -e BUZZ_INGEST_GITHUB_SECRET=synthetic-image-test "$image")
+address=$(docker port "$container" 5010/tcp)
+for attempt in {1..40}; do
+  if curl -fsS "http://$address/health" >/dev/null; then break; fi
+  sleep 0.2
+done
+[[ $(curl -s -o /dev/null -w '%{http_code}' "http://$address/health") = 200 ]]
+[[ $(curl -s -o /dev/null -w '%{http_code}' "http://$address/ready") = 503 ]]
+[[ $(curl -s -o /dev/null -w '%{http_code}' -H 'Content-Type: application/json' \
+  -d '{}' "http://$address/ingest/github") = 401 ]]
+echo 'buzz-ingest image: hardened startup, health, readiness and signature refusal passed'

From c20572585ddbfabf1b2e28ee176ac1f166ceaef9 Mon Sep 17 00:00:00 2001
From: Tee8z <tee8z@protonmail.com>
Date: Tue, 10 Nov 2026 14:26:36 -0500
Subject: [PATCH 2/2] fix(ingest): preserve reply semantics and current channel
 routing

Signed-off-by: Tee8z <tee8z@protonmail.com>
---
 crates/buzz-ingest/README.md        | 14 +++--
 crates/buzz-ingest/src/http.rs      | 17 +++++-
 crates/buzz-ingest/src/normalize.rs | 10 ++--
 crates/buzz-ingest/src/plan.rs      |  2 +
 crates/buzz-ingest/src/tests.rs     | 82 ++++++++++++++++++++++++++++-
 docs/nips/NIP-IG.md                 |  2 +-
 6 files changed, 114 insertions(+), 13 deletions(-)

diff --git a/crates/buzz-ingest/README.md b/crates/buzz-ingest/README.md
index 789c12a383d..d6f6cbd1862 100644
--- a/crates/buzz-ingest/README.md
+++ b/crates/buzz-ingest/README.md
@@ -36,8 +36,9 @@ Set `BUZZ_INGEST_CONFIG` to this JSON file:
 ```
 
 Every mapping is optional; configure between one and 256 total entries.
-Linear project mappings precede team mappings. Comments retain a previously saved
-issue route; a signed `webhookId` mapping supplies the fallback for unseen issues.
+Linear project mappings precede team mappings, then signed `webhookId` mappings.
+A previously saved issue route is used only when none of those mappings applies.
+Changing an explicit mapping redirects subsequent activity to the configured channel.
 Use `linear_webhooks` when comments on existing issues omit project/team fields.
 Previously mirrored issues also retain a relay-backed routing record.
 No Linear API credential or data mutation is required.
@@ -75,7 +76,8 @@ seconds; the complete delivery has a four-second deadline. Provider retries
 replay the stored plan. No background retry silently acknowledges failed work.
 
 Bodies are limited to 1 MiB. Each provider admits 20 requests per second, and
-one delivery runs at a time; excess requests receive 429. Message bodies retain
+one delivery publishes at a time. Up to 32 admitted requests can authenticate and
+wait within the four-second deadline; excess requests receive 429. Message bodies retain
 at most 6000 UTF-8 bytes. Payloads exceeding 16 labels are refused, so workflow metadata is never silently
 truncated. Other metadata has separate bounds.
 
@@ -86,8 +88,10 @@ Configure Linear issue/comment webhooks for mapped teams or webhook IDs.
 requests outside five minutes are refused. Slack needs `message.channels` event
 subscriptions, the `channels:history` bot scope, and membership in mapped public
 channels. [Slack signs the version, timestamp, and raw body](https://docs.slack.dev/authentication/verifying-requests-from-slack/).
-No DM/private-channel scopes are needed. Providers have finite retry policies;
-monitor delivery failures and manually redeliver after prolonged outages.
+No DM/private-channel scopes are needed. GitHub does not automatically redeliver
+failed webhooks; use its delivery history or an operator retry job to redeliver
+failures. Linear and Slack retries are finite. Monitor failures and redeliver
+after prolonged outages.
 
 ## Validation
 
diff --git a/crates/buzz-ingest/src/http.rs b/crates/buzz-ingest/src/http.rs
index cadcbbdb0ea..c9133b23b51 100644
--- a/crates/buzz-ingest/src/http.rs
+++ b/crates/buzz-ingest/src/http.rs
@@ -39,6 +39,7 @@ struct App {
     secrets: Secrets,
     relay: Relay,
     pipeline: Semaphore,
+    admission: Semaphore,
     readiness: Semaphore,
     rates: Mutex<[Bucket; 3]>,
 }
@@ -50,6 +51,7 @@ pub(crate) fn router(config: Config, secrets: Secrets, relay: Relay) -> Router {
         secrets,
         relay,
         pipeline: Semaphore::new(1),
+        admission: Semaphore::new(32),
         readiness: Semaphore::new(1),
         rates: Mutex::new(std::array::from_fn(|_| Bucket {
             start: now,
@@ -125,7 +127,7 @@ async fn deliver(
     let Some(secret) = secret else {
         return StatusCode::NOT_FOUND.into_response();
     };
-    let Ok(_permit) = app.pipeline.try_acquire() else {
+    let Ok(_permit) = app.admission.try_acquire() else {
         return (StatusCode::TOO_MANY_REQUESTS, "ingestion busy; retry").into_response();
     };
     {
@@ -234,7 +236,11 @@ async fn process(app: &App, source: usize, secret: &str, request: Request) -> Re
                 || payload
                     .pointer("/data/projectId")
                     .and_then(Value::as_str)
-                    .is_some_and(|id| app.config.linear_projects.contains_key(id));
+                    .is_some_and(|id| app.config.linear_projects.contains_key(id))
+                || payload
+                    .get("webhookId")
+                    .and_then(Value::as_str)
+                    .is_some_and(|id| app.config.linear_webhooks.contains_key(id));
             let route = if !explicit {
                 if let Some(issue) = normalize::linear_issue(&payload) {
                     plan::linear_channel(&app.relay, issue, &app.config)
@@ -272,6 +278,13 @@ async fn process(app: &App, source: usize, secret: &str, request: Request) -> Re
     let identity = serde_json::to_vec(&identity_payload)
         .map_err(|_| (StatusCode::BAD_REQUEST, "invalid payload identity"))?;
     let body_hash = hex::encode(Sha256::digest(identity));
+    // Authenticate and normalize before waiting for the single plan writer.
+    // The outer deadline bounds both this wait and the complete publication.
+    let _pipeline = app
+        .pipeline
+        .acquire()
+        .await
+        .map_err(|_| (StatusCode::SERVICE_UNAVAILABLE, "ingestion closing; retry"))?;
     for item in &items {
         plan::publish(&app.relay, item, &body_hash)
             .await
diff --git a/crates/buzz-ingest/src/normalize.rs b/crates/buzz-ingest/src/normalize.rs
index a11acc062ce..27e41548d69 100644
--- a/crates/buzz-ingest/src/normalize.rs
+++ b/crates/buzz-ingest/src/normalize.rs
@@ -260,13 +260,13 @@ pub(crate) fn linear(
         .get(text(data, "/projectId"))
         .or_else(|| config.linear_teams.get(text(data, "/teamId")))
         .copied()
-        .or(routed)
         .or_else(|| {
             config
                 .linear_webhooks
                 .get(text(payload, "/webhookId"))
                 .copied()
-        });
+        })
+        .or(routed);
     let Some(channel) = channel else {
         if kind == "Comment" {
             bail!("Linear issue route is unavailable; deliver the issue before its comments");
@@ -369,7 +369,11 @@ pub(crate) fn linear(
         ("title", "/title"),
         ("state", "/state/name"),
     ] {
-        let value = text(data, pointer);
+        let value = if field == "state" && text(data, pointer).is_empty() {
+            text(data, "/stateId")
+        } else {
+            text(data, pointer)
+        };
         if !value.is_empty() {
             item.fields.push((field.into(), short(value, 256)));
         }
diff --git a/crates/buzz-ingest/src/plan.rs b/crates/buzz-ingest/src/plan.rs
index d1f80c80f6d..4573a6406c8 100644
--- a/crates/buzz-ingest/src/plan.rs
+++ b/crates/buzz-ingest/src/plan.rs
@@ -203,6 +203,7 @@ pub(crate) async fn publish(relay: &Relay, item: &Occurrence, body_hash: &str) -
             let mut tags = vec![
                 tag("h", &item.channel.to_string())?,
                 Tag::parse(["e", &root.event.id.to_hex(), "", "root"])?,
+                Tag::parse(["e", &root.event.id.to_hex(), "", "reply"])?,
             ];
             for (key, val) in &item.fields {
                 tags.push(tag(&format!("ext-{key}"), val)?);
@@ -224,6 +225,7 @@ pub(crate) async fn publish(relay: &Relay, item: &Occurrence, body_hash: &str) -
         let mut tags = plan.event.tags.clone().to_vec();
         tags.retain(|tag| tag.as_slice().first().map(String::as_str) != Some("e"));
         tags.push(Tag::parse(["e", &root.event.id.to_hex(), "", "root"])?);
+        tags.push(Tag::parse(["e", &root.event.id.to_hex(), "", "reply"])?);
         plan.event = EventBuilder::new(Kind::Custom(9), plan.event.content.clone())
             .tags(tags)
             .sign_with_keys(&relay.keys)?;
diff --git a/crates/buzz-ingest/src/tests.rs b/crates/buzz-ingest/src/tests.rs
index 649c04c07a5..1025c805c66 100644
--- a/crates/buzz-ingest/src/tests.rs
+++ b/crates/buzz-ingest/src/tests.rs
@@ -153,6 +153,14 @@ async fn submit(
             "every message requires a durable plan before publication"
         );
         if tag(&event, "e").is_some() {
+            let (root, parent) = buzz_core::nip10::parse_thread_markers(&event.tags)
+                .resolve()
+                .expect("display plans must be real replies, including after refresh");
+            assert_eq!(root, parent);
+            assert!(
+                store.events.contains_key(&root),
+                "reply anchor must already exist"
+            );
             assert!(
                 record_count >= 2,
                 "delivery and body fingerprint plans precede replies"
@@ -560,7 +568,7 @@ async fn linear_and_slack_only_publish_authenticated_mapped_public_work() {
     let keys = Keys::generate();
     let router = app(config(&url), &keys);
     let now = nostr::Timestamp::now().as_secs() * 1000;
-    let issue = json!({"type":"Issue","action":"create","webhookTimestamp":now,"webhookId":"HOOK","url":"https://linear.app/issue/ENG-7","actor":{"name":"Ada"},"data":{"id":"issue-7","identifier":"ENG-7","teamId":"TEAM","title":"Fix it","description":"Details"}});
+    let issue = json!({"type":"Issue","action":"create","webhookTimestamp":now,"webhookId":"HOOK","url":"https://linear.app/issue/ENG-7","actor":{"name":"Ada"},"data":{"id":"issue-7","identifier":"ENG-7","teamId":"TEAM","title":"Fix it","description":"Details","stateId":"STATE-TODO"}});
     assert_eq!(
         request(&router, "linear", &issue, "linear-1", true, 0, "")
             .await
@@ -573,6 +581,9 @@ async fn linear_and_slack_only_publish_authenticated_mapped_public_work() {
             .0,
         StatusCode::OK
     );
+    assert!(messages(&store)
+        .iter()
+        .any(|event| tag(event, "ext-state") == Some("STATE-TODO")));
     let mut labeled = issue.clone();
     labeled["action"] = json!("update");
     labeled["data"]["labelIds"] = json!(["LABEL-AGENT"]);
@@ -601,7 +612,7 @@ async fn linear_and_slack_only_publish_authenticated_mapped_public_work() {
         3,
         "transport timestamps must not duplicate a Linear occurrence"
     );
-    let comment = json!({"type":"Comment","action":"create","webhookTimestamp":now,"webhookId":"HOOK","url":"https://linear.app/issue/ENG-7#comment-1","actor":{"name":"Ada"},"data":{"id":"comment-1","issueId":"issue-7","body":"An update"}});
+    let comment = json!({"type":"Comment","action":"create","webhookTimestamp":now,"url":"https://linear.app/issue/ENG-7#comment-1","actor":{"name":"Ada"},"data":{"id":"comment-1","issueId":"issue-7","body":"An update"}});
     assert_eq!(
         request(&router, "linear", &comment, "linear-2", false, 0, "")
             .await
@@ -662,6 +673,7 @@ async fn linear_and_slack_only_publish_authenticated_mapped_public_work() {
     // An existing issue without a saved route can use an explicitly mapped
     // signed webhook ID. A known route above must retain the team channel.
     let mut orphan = comment.clone();
+    orphan["webhookId"] = json!("HOOK");
     orphan["data"]["id"] = json!("orphan-comment");
     orphan["data"]["issueId"] = json!("never-seen-issue");
     assert_eq!(
@@ -687,3 +699,69 @@ fn endpoints_and_provider_mappings_fail_closed() {
         .insert("T123/GPRIVATE".into(), CHANNEL.parse().unwrap());
     assert!(cfg.validate().is_err());
 }
+
+/// A bounded burst of authenticated CI callbacks waits for the serialized
+/// journal writer instead of rejecting every overlapping GitHub delivery.
+#[tokio::test]
+async fn authenticated_ci_bursts_preserve_every_delivery() {
+    let (url, store, server) = external().await;
+    let keys = Keys::generate();
+    let router = app(config(&url), &keys);
+    let mut tasks = tokio::task::JoinSet::new();
+    for index in 0..5 {
+        let router = router.clone();
+        tasks.spawn(async move {
+            let mut payload = github();
+            payload["pull_request"]["number"] = json!(100 + index);
+            request(
+                &router,
+                "github",
+                &payload,
+                &format!("burst-{index}"),
+                false,
+                0,
+                "pull_request",
+            )
+            .await
+            .0
+        });
+    }
+    while let Some(outcome) = tasks.join_next().await {
+        assert_eq!(outcome.unwrap(), StatusCode::OK);
+    }
+    assert_eq!(messages(&store).len(), 10);
+    server.abort();
+}
+
+/// A saved route is a fallback; current explicit signed webhook configuration
+/// can redirect subsequent activity without leaking it to the previous channel.
+#[tokio::test]
+async fn linear_current_mapping_overrides_saved_issue_route() {
+    let (url, store, server) = external().await;
+    let keys = Keys::generate();
+    let router = app(config(&url), &keys);
+    let now = nostr::Timestamp::now().as_secs() * 1000;
+    let issue = json!({"type":"Issue","action":"create","webhookTimestamp":now,"data":{"id":"redirect-issue","identifier":"ENG-9","teamId":"TEAM","title":"Original route"},"url":"https://linear.app/issue/ENG-9"});
+    assert_eq!(
+        request(&router, "linear", &issue, "route-before", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    let comment = json!({"type":"Comment","action":"create","webhookTimestamp":now,"webhookId":"HOOK","data":{"id":"redirect-comment","issueId":"redirect-issue","body":"Current route"},"url":"https://linear.app/issue/ENG-9#comment"});
+    assert_eq!(
+        request(&router, "linear", &comment, "route-after", false, 0, "")
+            .await
+            .0,
+        StatusCode::OK
+    );
+    let imported = messages(&store)
+        .into_iter()
+        .find(|event| event.content.contains("Current route"))
+        .unwrap();
+    assert_eq!(
+        tag(&imported, "h"),
+        Some("0b6c1f0e-2a7d-4c1e-9f3a-5d2b8e4c7a11")
+    );
+    server.abort();
+}
diff --git a/docs/nips/NIP-IG.md b/docs/nips/NIP-IG.md
index 62da432fa70..c30a7d1b0c6 100644
--- a/docs/nips/NIP-IG.md
+++ b/docs/nips/NIP-IG.md
@@ -7,7 +7,7 @@ thread-counter rules. The relay contains no provider-specific webhook handlers.
 | Event | Purpose |
 | --- | --- |
 | Kind 9 root | A stable external thread anchor with `h`, `ext-source`, and `ext-thread`. It has no `ext-action`. |
-| Kind 9 reply | One occurrence, with NIP-10 root reference and typed `ext-*` metadata. |
+| Kind 9 reply | One occurrence, with both NIP-10 root and reply references to the anchor, plus typed `ext-*` metadata. |
 | Kind 30623 | Parameterized replaceable write-ahead plan. `d` identifies the journal coordinate; `h` scopes access; `message` names the planned kind 9 event. |
 
 A record's content is standard-base64 JSON: