diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..ce628aa --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,90 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install -e '.[dev]' + - run: ruff check --output-format=github . + + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + # The floor is `requires-python`; the ceiling is whatever is current. + # Hermes decides which of these a real install runs on, so the plugin + # should not be the thing that narrows it. + python-version: ["3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: pip install -e '.[dev]' + - run: python -m pytest -q + + package: + # A plugin that installs without its manifest, dashboard bundle or skill + # registers nothing, and the failure is silent — the platform simply never + # appears. Building the artifact and looking inside it is the only way that + # gets caught before a release. + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install build twine + - run: python -m build + - run: twine check dist/* + - name: The wheel must carry the non-Python half of the plugin + run: | + set -euo pipefail + wheel=$(ls dist/*.whl) + echo "inspecting $wheel" + contents=$(python -m zipfile --list "$wheel") + for required in \ + hookdeck/plugin.yaml \ + hookdeck/dashboard/manifest.json \ + hookdeck/dashboard/dist/index.js \ + hookdeck/skills/triage-webhook-failures/SKILL.md + do + if ! grep -qF "$required" <<<"$contents"; then + echo "::error::$required is missing from the wheel" + exit 1 + fi + echo " ✓ $required" + done + - name: The entry point Hermes discovers the plugin by must be declared + run: | + set -euo pipefail + pip install dist/*.whl + python - <<'PY' + from importlib.metadata import entry_points + found = entry_points(group="hermes_agent.plugins") + names = {e.name: e.value for e in found} + assert names.get("hookdeck") == "hookdeck", names + print("✓ hermes_agent.plugins entry point:", names) + PY + - uses: actions/upload-artifact@v4 + with: + name: dist + path: dist/ diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..4d7dffe --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,90 @@ +# Tag-driven release to PyPI. +# +# 1. bump `__version__` in hookdeck/__init__.py (pyproject reads it from there) +# 2. update CHANGELOG.md +# 3. git tag v0.2.0 && git push --tags +# +# Publishing uses PyPI Trusted Publishing (OIDC), so there is no API token in +# the repository to leak or rotate. It needs a one-time setup on PyPI: +# Project → Publishing → add a GitHub publisher for hookdeck/hermes-hookdeck, +# workflow `release.yml`, environment `pypi`. +name: Release + +on: + push: + tags: ["v*"] + workflow_dispatch: + +permissions: + contents: read + +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install -e '.[dev]' build twine + + - name: The tag and the package version must agree + if: startsWith(github.ref, 'refs/tags/v') + run: | + set -euo pipefail + tagged="${GITHUB_REF_NAME#v}" + declared=$(python -c 'import hookdeck; print(hookdeck.__version__)') + if [ "$tagged" != "$declared" ]; then + echo "::error::tag $GITHUB_REF_NAME does not match hookdeck.__version__ ($declared)" + exit 1 + fi + echo "✓ releasing $declared" + + # A release that cannot pass its own test suite is not a release. + - run: ruff check . + - run: python -m pytest -q + + - run: python -m build + - run: twine check dist/* + - uses: actions/upload-artifact@v4 + with: + name: dist + path: dist/ + + publish: + needs: build + runs-on: ubuntu-latest + environment: + name: pypi + url: https://pypi.org/p/hermes-hookdeck + permissions: + # The OIDC token pypa/gh-action-pypi-publish exchanges for an upload + # token. Nothing else in this workflow needs it, which is why it is + # scoped to this job rather than the file. + id-token: write + steps: + - uses: actions/download-artifact@v4 + with: + name: dist + path: dist/ + - uses: pypa/gh-action-pypi-publish@release/v1 + + github-release: + needs: publish + runs-on: ubuntu-latest + if: startsWith(github.ref, 'refs/tags/v') + permissions: + contents: write + steps: + - uses: actions/checkout@v4 + - uses: actions/download-artifact@v4 + with: + name: dist + path: dist/ + - name: Publish the GitHub release + env: + GH_TOKEN: ${{ github.token }} + run: | + gh release create "$GITHUB_REF_NAME" dist/* \ + --title "$GITHUB_REF_NAME" \ + --notes "See [CHANGELOG.md](https://github.com/${GITHUB_REPOSITORY}/blob/${GITHUB_REF_NAME}/CHANGELOG.md)." diff --git a/.github/workflows/upstream-contract.yml b/.github/workflows/upstream-contract.yml new file mode 100644 index 0000000..2d75546 --- /dev/null +++ b/.github/workflows/upstream-contract.yml @@ -0,0 +1,37 @@ +# The test suite runs against tests/hermes_stub.py, because the plugin lives +# outside the Hermes tree and there is no other way to exercise the ingest path +# without a Hermes checkout. The blind spot that buys is real: the stub cannot +# notice when the thing it stands in for changes. +# +# So this asks upstream directly, on a schedule rather than on every PR — the +# answer changes when Hermes changes, not when this repo does. A failure here +# is a heads-up, not a broken build. +name: Upstream contract + +on: + schedule: + - cron: "0 6 * * 1" + workflow_dispatch: + push: + paths: + - "scripts/check_upstream_contract.py" + - "tests/hermes_stub.py" + - ".github/workflows/upstream-contract.yml" + +permissions: + contents: read + +jobs: + check: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Fetch the parts of Hermes this plugin borrows from + run: | + git clone --depth 1 --filter=blob:none --sparse \ + https://github.com/NousResearch/hermes-agent.git upstream + git -C upstream sparse-checkout set gateway agent + - run: python scripts/check_upstream_contract.py upstream diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..73548ec --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,65 @@ +# Changelog + +Notable changes per release. Versions follow [semantic versioning](https://semver.org), +with the caveat that until 1.0 the config surface — `platforms.hookdeck.extra` +— may change in a minor release, and any such change is listed here. + +## Unreleased + +### Fixed + +- A signature header containing a non-ASCII byte answered 500 instead of 401. + The 500 skipped the `EMITTED_STATUS_RETRYABLE` guard and fell inside the + provisioned retry rule's `500-599` range, so Hookdeck retried a forged + request rather than dropping it. +- A delivery's event id no longer falls back to a bare `X-Request-ID` header. + That header is not a Hookdeck identifier, is not subject to `header_prefix`, + and is not unique per delivery — one request fans out to one event per + matching connection. Deliveries genuinely missing `x-hookdeck-eventid` are + processed with the existing warning that dedup and retry are unavailable. +- The marker tracking a `sync` run that outlasted its timeout is now cleared on + every terminal path, not only on success. It previously survived an exhausted + retry budget or an abandoned hand-back, and a later run of the same event id + would be treated as having been acked early when it had not. + +### Added + +- Continuous integration: lint, a test matrix across Python 3.10–3.13, and a + packaging job that asserts the built wheel still carries `plugin.yaml`, the + dashboard bundle and the bundled skill — without which the plugin installs + but registers nothing. +- A weekly check that the Hermes internals this plugin subclasses and calls + still exist upstream, since the test suite otherwise runs entirely against + `tests/hermes_stub.py` and cannot notice the real thing moving. +- Ruff, with `BLE001` selected so each deliberate blind `except` carries its + reason at the point it is written. + +### Changed + +- The package version is read from `hookdeck.__version__` rather than declared + a second time in `pyproject.toml`. + +### Documentation + +- The README opens by saying what Hermes Agent and Hookdeck Event Gateway are, + rather than assuming both. +- It also says *which* Hookdeck. This is the Event Gateway — inbound events + arriving at your agent — and not Outpost, which is the other direction. And + "platform" in `kind: platform` is Hermes' word for a source of inbound work, + not a reference to the Hookdeck platform. +- A new architecture section with two diagrams: topology, and a sequence diagram + for the ack-then-hand-back contract that the reliability rests on. The three + different things called "CLI" are separated there. +- A new section listing Hookdeck capabilities the plugin does not currently use + — the Publish API, bulk operation plans and cancellation, request replay, + ignored-event retry, issue triggers, transformations, the wider metrics — so + the edge of `hookdeck/api.py` is not mistaken for the edge of the product. +- The "no pull API" limitation now explains what the durability claim actually + rests on: an event is recoverable because a delivered-but-failed event stays + retryable, not because anything holds a lease on it. + +## 0.1.0 + +First release. Hookdeck platform adapter, `hermes hookdeck` operator commands, +the `hookdeck` agent toolset, the `triage-webhook-failures` skill and the +dashboard tab. diff --git a/README.md b/README.md index dcfecf9..0dccbda 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,32 @@ # hermes-hookdeck -A Hookdeck platform plugin for [Hermes Agent](https://github.com/NousResearch/hermes-agent). -It puts a durable, verified queue in front of the agent, so a webhook can -trigger an agent run without the usual ways that goes wrong. +A [Hookdeck Event Gateway](https://hookdeck.com/docs) plugin for +[Hermes Agent](https://github.com/NousResearch/hermes-agent). It puts a durable, +verified queue in front of the agent, so a webhook can trigger an agent run +without the usual ways that goes wrong. + +If you have not met both halves: **Hermes Agent** is a self-hosted AI agent from +Nous Research that runs as a long-lived gateway process — on a laptop, a $5 VPS, +wherever — and takes work from Telegram, Discord, Slack, a terminal, a cron +schedule, or a webhook. **Hookdeck Event Gateway** is a hosted service that sits +between a webhook provider and you: it verifies the provider's signature, queues +each event, applies filters and retries, and holds everything it has not yet +delivered so you can inspect or replay it. This plugin makes Hookdeck the front +door for Hermes' webhook trigger — so a GitHub pull request, a Stripe payment or +a Shopify order becomes an agent run that is verified once, runs once, and is +not silently lost when the run fails or the machine restarts. + +Two names worth pinning down, because both are overloaded: + +- **Event Gateway, not the rest of Hookdeck.** Hookdeck's platform also includes + [Outpost](https://hookdeck.com/docs/outpost), which is the other direction — + self-hosted infrastructure for sending *your* webhooks to *your* users. This + plugin is inbound only: third-party events arriving at your agent. Nothing + here helps Hermes publish webhooks, and it does not talk to Outpost. +- **"Platform" is Hermes' word, not Hookdeck's.** In Hermes a *platform* is a + source of inbound work — Telegram is a platform, Slack is a platform — and + this plugin registers a new one called `hookdeck`, alongside the built-in + `webhook`. It is unrelated to the Hookdeck platform in the marketing sense. Hermes already has a good webhook trigger: a POST arrives, a route matches, a prompt template renders, the agent runs, the response gets delivered. This @@ -27,6 +51,111 @@ The last two rows are the ones that matter most. The built-in adapter answers 202 as soon as it dispatches, which is the right thing to do — but it means a failed run has already been acknowledged, and nothing remembers it happened. +## How it fits together + +```mermaid +flowchart LR + P["Provider
GitHub · Stripe · Shopify · …"] + + subgraph HD["Hookdeck Event Gateway — hosted"] + direction TB + SRC["Source
verifies the provider's
own signature"] + RULES["Connection rules
filter · deduplicate · retry"] + Q[("Event queue
holds what is not yet
delivered, within retention")] + SRC --> RULES --> Q + end + + subgraph GW["Your machine — one hermes gateway process"] + direction TB + AD["hookdeck adapter
verifies x-hookdeck-signature
deduplicates · admission control"] + LED[("Delivery ledger
SQLite, survives restarts")] + RUN["Agent run
prompt → tools → response"] + AD <--> LED + AD --> RUN + end + + P -->|"POST, signed by the provider"| SRC + Q -->|"cli mode
hookdeck listen holds an outbound
connection — no public URL"| AD + Q -->|"push mode
HTTPS to your reachable URL"| AD + + style HD fill:#f4f7ff,stroke:#4571d1,color:#26324d + style GW fill:#f3faf1,stroke:#3f8f3c,color:#1f3d1e +``` + +Two signatures, two different jobs. Hookdeck checks the *provider's* signature +at the edge — Stripe's, Shopify's, Twilio's, ~140 schemes — and then signs its +own delivery. The adapter checks only that one, which is the whole point of the +integration: Hermes implements one verifier instead of one per provider. + +### The delivery that has to come back + +The diagram above is just the path in. What makes this more than a webhook +listener is the arrow it does not show — the adapter telling Hookdeck that a run +failed, so the event returns instead of being forgotten: + +```mermaid +sequenceDiagram + autonumber + participant H as Hookdeck + participant A as Adapter + participant L as Ledger + participant R as Agent run + + H->>A: deliver — event id, attempt 1, x-hookdeck-signature + A->>A: verify · route · parse · filter + A->>L: is this new work? + L-->>A: yes — attempt 1 beats nothing seen + A-->>H: 202 accepted + Note over A,H: The ack goes out before the run finishes.
Recoverable in both directions, which is what lets
Hookdeck be the queue instead of the plugin owning one. + A->>R: dispatch + R-->>A: failed + A->>L: mark failed + A->>H: POST /events/{id}/retry + H->>A: deliver — same event id, attempt 2 + Note over L: attempt 2 > attempt 1, so this is a retry, not a duplicate.
A repeat of attempt 1 would be refused. +``` + +The attempt counter is what lets deduplication and retry coexist rather than +cancelling each other out. It is also how a gateway that dies at step 7 recovers: +the ledger row is still `running` at the next boot, which by then can only mean +the process that owned it is gone, so the adapter asks for the same redelivery +at step 9. That is [`ack_mode: async_retry`](#how-the-reliability-works), the +default; `sync` holds the response open instead and lets Hookdeck's own retry +rules do the work. + +### Two ways in + +Both modes run the same adapter and the same reliability machinery. They differ +only in how an event crosses your network boundary. + +| | `mode: cli` (default) | `mode: push` | +|---|---|---| +| Reachability | None needed — the connection is outbound | A public HTTPS URL | +| Suits | A laptop, a homelab box, anything behind NAT | A VPS, a container, anything with an address | +| Extra process | One `hookdeck listen` per route | None | +| Gateway-side throttling | Not available — CLI destinations have no rate limit | Delivery rate limits, delivery groups, issue triggers, alerting | +| Buffering while you are down | Only if you **pause** first — see the caveat in the CLI quickstart | Yes; failed deliveries stay queued and retry | + +In `cli` mode the listener binds loopback only and is not reachable from the +network at all. In `push` mode it binds whatever `host` you configure, and the +signature check is the only thing in front of it. + +### Three things here are called "CLI" + +Worth separating once, because the quickstarts use all three: + +- **The Hookdeck CLI** (`hookdeck`) — a binary you install from Hookdeck, and + the thing that makes `cli` mode work. You do not run it by hand: the adapter + spawns `hookdeck listen` itself, one process per route, and supervises it — + restarting with capped backoff if it dies, piping its output into the gateway + log. What you do need is to be logged in (`hookdeck login`) and on version + 2.3.2 or later. +- **`hermes hookdeck …`** — the operator commands this plugin adds: `setup`, + `status`, `pause`, `resume`, `replay`, `doctor`. These call the Hookdeck REST + API rather than the binary above, and work in both modes. +- **`hermes`** — Hermes' own CLI, which hosts all of the above. `hermes gateway` + runs the process the adapter lives in. + ## Install ```bash @@ -269,9 +398,14 @@ than core's exemption, which covers built-in webhook routes even unverified. signing secret entered on the source in the Hookdeck dashboard. `setup` creates the source with the right verification shape but cannot invent the secret. -- The plugin does not poll. Hookdeck has no pull/consumer API — the Events API - is inspection-only, with no ack, lease or consumer group — so if you cannot - run the CLI and cannot expose a URL, it will not help you. +- Delivery is push-only, in both directions. Hookdeck pushes to the adapter and + the adapter pushes retry requests back; there is no lease-and-ack loop, and no + pull API to build one from — the Events API is for inspection, with no ack, + lease or consumer group. Two consequences worth being explicit about. If you + can neither run the CLI nor expose a URL, this plugin cannot help you. And + "the event is safe in Hookdeck" holds because a delivered-but-failed event + stays retryable, not because anything is holding a lock on it — which is why + boot recovery has to reconcile `running` ledger rows itself. - Delivery groups throttle per subject by rate, not by concurrency, because Hookdeck's group-level period is `second|minute|hour`. - Every recovery path is bounded by your plan's retention: 3 days on @@ -285,6 +419,62 @@ than core's exemption, which covers built-in webhook routes even unverified. in the instant before a crash. That is the at-least-once contract the whole design assumes; set `recover_on_boot: false` if it is wrong for your routes. +## Hookdeck can do more than this plugin asks it to + +The list above is what *cannot* be done. This is the other kind of boundary: +things Hookdeck offers that the plugin simply does not wire up yet, so nobody +mistakes the edge of `hookdeck/api.py` for the edge of the product. Each is a +candidate, not a promise — [issues and PRs welcome](https://github.com/hookdeck/hermes-hookdeck/issues). + +**Getting events in.** The +[Publish API](https://hookdeck.com/docs/api/publish.md) — +`POST https://hkdk.events/v1/publish` with an `X-Hookdeck-Source-Name` header — +sends a request to any source, authenticated with the same API key everything +else here uses. Nothing in the plugin calls it, and two uses stand out: a +`hermes hookdeck test ` that puts a real event through the real +connection without waiting for a provider to fire one, and a way for Hermes to +enqueue durable work for itself — the queue, ledger, retry and replay machinery +all apply to a published event exactly as to a provider's. + +**Recovering what the queue calls "ignored".** The CLI-mode caveat above — that +events arriving with no listener attached become `CLI_DISCONNECTED` ignored +events and are discarded — is stated against what the plugin currently does with +them, which is nothing. Hookdeck has +[`POST /bulk/ignored-events/retry`](https://hookdeck.com/docs/api/bulk.md#bulk-retry-ignored-events), +which retries ignored events matching a `cause`. Whether `CLI_DISCONNECTED` is +among the causes it accepts needs confirming against a live project — the docs +show `FILTERED` and `TRANSFORMATION_FAILED` — but if it is, the sharpest edge in +CLI mode has a recovery path and `hermes hookdeck doctor` should be pointing at +it. + +**Bulk operations with the safety catch on.** `hookdeck_bulk_retry` is an +*agent-callable* tool that fires `POST /bulk/events/retry` immediately. Hookdeck +estimates a bulk operation before running it (`GET /bulk/events/retry/plan`), +and can cancel one in flight (`POST /bulk/events/retry/{id}/cancel`). An agent +that could see "this would re-run 4,000 events" before committing is a +materially safer agent. `POST /bulk/events/cancel` is the other half — a way to +stop a flood rather than grind through it. + +**Requests, not just events.** A Hookdeck *request* is what the provider sent; an +*event* is one connection's copy of it. `/bulk/requests/retry` and +`/bulk/requests/replay` re-run the request, producing fresh events for every +matching connection. That is the right instrument after fixing a connection that +was misconfigured when the traffic arrived, and the plugin only knows about +events. + +**Alerting and shaping.** [Issue triggers and +notifications](https://hookdeck.com/docs/api) can tell you a connection is +failing without anyone watching `hermes hookdeck status`; `setup` provisions +none. [Transformations](https://hookdeck.com/docs/api) run JavaScript on an +event before delivery — the documented workaround for the XML limitation above +is to add one by hand, and `setup` could manage it. Destinations can also carry +their own auth (bearer, basic, API key); the plugin pins +`HOOKDECK_SIGNATURE`, which is the right default and currently the only option. + +**Metrics beyond queue depth.** The dashboard tab reads +`GET /metrics/queue-depth`. Hookdeck also exposes request, event, attempt and +events-by-issue metrics, which would turn that panel from a number into a trend. + ## Verified end to end Against a real Hermes 0.20.0 gateway, a real Hookdeck project and the Hookdeck diff --git a/hookdeck/__init__.py b/hookdeck/__init__.py index 8bb8ff7..a16e4ea 100644 --- a/hookdeck/__init__.py +++ b/hookdeck/__init__.py @@ -1,4 +1,4 @@ -"""Hookdeck event gateway plugin for Hermes Agent. +"""Hookdeck Event Gateway plugin for Hermes Agent. ``register(ctx)`` wires up three surfaces: @@ -22,7 +22,7 @@ logger = logging.getLogger(__name__) __version__ = "0.1.0" -__all__ = ["register", "PLATFORM_NAME", "__version__"] +__all__ = ["PLATFORM_NAME", "__version__", "register"] PLATFORM_HINT = ( "You were triggered by a webhook delivered through Hookdeck, not by a " @@ -94,7 +94,7 @@ def _register_cli_commands(ctx: Any) -> None: ctx.register_cli_command( name="hookdeck", - help="Provision and operate the Hookdeck event gateway", + help="Provision and operate the Hookdeck Event Gateway", setup_fn=register_cli, handler_fn=hookdeck_command, description=( diff --git a/hookdeck/adapter.py b/hookdeck/adapter.py index 6035583..ba82560 100644 --- a/hookdeck/adapter.py +++ b/hookdeck/adapter.py @@ -26,7 +26,7 @@ import logging import time from dataclasses import dataclass -from typing import Any, Optional +from typing import Any try: from aiohttp import web @@ -135,8 +135,8 @@ def __init__(self, config: PlatformConfig): self._routes = self.settings.routes self._max_body_bytes = self.settings.max_body_bytes - self._ledger: Optional[DeliveryLedger] = None - self._api: Optional[HookdeckAPI] = None + self._ledger: DeliveryLedger | None = None + self._api: HookdeckAPI | None = None self._tunnels: list[HookdeckTunnel] = [] self._site_runner = None @@ -147,7 +147,7 @@ def __init__(self, config: PlatformConfig): # failures need the same explicit hand-back an async_retry run gets, # because Hookdeck has already recorded the delivery as successful. self._acked_before_completion: set[str] = set() - self._maintenance: Optional[asyncio.Task] = None + self._maintenance: asyncio.Task | None = None self._last_sweep = 0.0 @staticmethod @@ -263,7 +263,7 @@ async def _start_sites(self) -> bool: """Start a listener per address. One family may be absent; both failing is fatal.""" assert self._site_runner is not None started: list[str] = [] - last_error: Optional[OSError] = None + last_error: OSError | None = None for host in self.settings.bind_hosts: try: await web.TCPSite(self._site_runner, host, self._port).start() @@ -513,7 +513,7 @@ async def _handle_delivery(self, request: web.Request) -> web.Response: async def _read_verified_body( self, request: web.Request - ) -> tuple[bytes, Optional[web.Response]]: + ) -> tuple[bytes, web.Response | None]: """Read the body within limits and verify it, before anything parses it.""" too_large = self._respond({"error": "Payload too large"}, status=413) @@ -523,7 +523,7 @@ async def _read_verified_body( raw_body = await request.read() except web.HTTPRequestEntityTooLarge: return b"", too_large - except Exception as exc: + except Exception as exc: # noqa: BLE001 - any read failure is a bad request logger.error("[hookdeck] Failed to read body: %s", exc) return b"", self._respond({"error": "Bad request"}, status=400) if len(raw_body) > self.settings.max_body_bytes: @@ -552,12 +552,17 @@ def _signature_valid(self, request: web.Request, raw_body: bytes) -> bool: def _parse_delivery( self, request: web.Request, raw_body: bytes - ) -> tuple[Delivery, Optional[web.Response]]: + ) -> tuple[Delivery, web.Response | None]: """Build a :class:`Delivery` from a verified request.""" source_name = self._header(request, SOURCE_NAME) - event_id = self._header(request, EVENT_ID) or request.headers.get( - "X-Request-ID", "" - ) + # No fallback. An id that did not come from Hookdeck is worse than + # none: `_admit` has a deliberate branch for a delivery with no event + # id, which warns loudly and names `header_prefix` as the likely cause, + # and a substitute id silences exactly that warning while breaking both + # things the id is for. Dedup keyed on a value Hookdeck did not mint is + # dedup on the wrong thing, and `POST /events/{id}/retry` with it 404s, + # so the failed run is never handed back. + event_id = self._header(request, EVENT_ID) try: attempt = int(self._header(request, ATTEMPT_COUNT) or 0) except ValueError: @@ -625,7 +630,7 @@ def _parse_delivery( async def _reject_if_filtered( self, request: web.Request, delivery: Delivery - ) -> Optional[web.Response]: + ) -> web.Response | None: """Apply the route's own filters. Ignored events answer 200, not an error.""" route = delivery.route @@ -696,7 +701,7 @@ async def _deliver_without_agent(self, delivery: Delivery) -> web.Response: } ) - def _already_handled(self, delivery: Delivery) -> Optional[str]: + def _already_handled(self, delivery: Delivery) -> str | None: """Why this delivery needs no run, if it needs none. Read-only on purpose. Answering this *before* the capacity check is @@ -715,7 +720,7 @@ def _already_handled(self, delivery: Delivery) -> Optional[str]: logger.info("[hookdeck] Skipping %s: %s", delivery.event_id, reason) return reason - def _admit(self, delivery: Delivery) -> Optional[web.Response]: + def _admit(self, delivery: Delivery) -> web.Response | None: """Claim a slot and a ledger entry, or defer. Nothing is recorded for a deferred event: a ledger entry would make @@ -963,6 +968,7 @@ async def _record_outcome( self._ledger.mark_exhausted( event_id, reason, session_chat_id=session_chat_id ) + self._acked_before_completion.discard(event_id) logger.error( "[hookdeck] Event %s failed %d times — giving up. Inspect it in " "Hookdeck and replay with `hermes hookdeck replay %s`.", @@ -986,7 +992,14 @@ async def _record_outcome( ) self._ledger.mark_failed(event_id, reason, session_chat_id=session_chat_id) - if not self._should_hand_back(event_id): + hand_back = self._should_hand_back(event_id) + # The marker has done its job the moment that decision is made, whether + # or not the hand-back then succeeds. Leaving it behind on the failure + # path would grow the set for the lifetime of the process, and a later + # sync-mode run of the same event would be treated as having been acked + # early when it was not. + self._acked_before_completion.discard(event_id) + if not hand_back: return if await self._request_redelivery(event_id): logger.info( @@ -1020,11 +1033,10 @@ async def _request_redelivery(self, event_id: str) -> bool: """ assert self._api is not None delay = _REDELIVERY_RETRY_INITIAL_SECONDS - last: Optional[HookdeckAPIError] = None + last: HookdeckAPIError | None = None for attempt in range(1, _REDELIVERY_ATTEMPTS + 1): try: await self._api.retry_event(event_id) - self._acked_before_completion.discard(event_id) return True except HookdeckAPIError as exc: last = exc @@ -1054,7 +1066,7 @@ def _respond( body: dict, *, status: int = 200, - headers: Optional[dict] = None, + headers: dict | None = None, ) -> web.Response: """Answer a delivery, refusing any status whose retryability is undeclared. @@ -1160,7 +1172,7 @@ def _apply_skills(self, route: dict, prompt: str) -> str: content = build_skill_invocation_message(command, user_instruction=prompt) if content: return content - except Exception as exc: + except Exception as exc: # noqa: BLE001 - a missing skill must not lose the event logger.warning("[hookdeck] Skill loading failed: %s", exc) return prompt @@ -1210,7 +1222,7 @@ def _sweep_inflight(self, *, force: bool = False) -> None: def _validate_startup(self) -> None: self.settings.validate() - def _bind_hosts(self) -> list[Optional[str]]: + def _bind_hosts(self) -> list[str | None]: return self.settings.bind_hosts def _tunnel_plan(self) -> dict[str, str]: @@ -1258,7 +1270,7 @@ def is_connected(config: PlatformConfig) -> bool: return validate_config(config) -def env_enablement() -> Optional[dict]: +def env_enablement() -> dict | None: """Seed ``PlatformConfig.extra`` from the environment. Lets ``hermes gateway status`` report an env-only setup without diff --git a/hookdeck/api.py b/hookdeck/api.py index 26914c8..b0ade5c 100644 --- a/hookdeck/api.py +++ b/hookdeck/api.py @@ -9,8 +9,9 @@ import asyncio import os +from collections.abc import Mapping from datetime import datetime, timedelta, timezone -from typing import TYPE_CHECKING, Any, Mapping, Optional +from typing import TYPE_CHECKING, Any from .constants import API_BASE_URL @@ -81,11 +82,11 @@ class HookdeckAPI: def __init__( self, - api_key: Optional[str] = None, + api_key: str | None = None, *, base_url: str = API_BASE_URL, timeout: float = 20.0, - client: Optional["httpx.AsyncClient"] = None, + client: httpx.AsyncClient | None = None, ): self.api_key = api_key or os.getenv("HOOKDECK_API_KEY", "") self.base_url = base_url.rstrip("/") @@ -97,7 +98,7 @@ def __init__( # Plumbing # ------------------------------------------------------------------ - def _ensure_client(self) -> "httpx.AsyncClient": + def _ensure_client(self) -> httpx.AsyncClient: if self._client is None: self._client = _httpx().AsyncClient(timeout=self._timeout) return self._client @@ -107,7 +108,7 @@ async def aclose(self) -> None: await self._client.aclose() self._client = None - async def __aenter__(self) -> "HookdeckAPI": + async def __aenter__(self) -> HookdeckAPI: return self async def __aexit__(self, *_exc: Any) -> None: @@ -198,7 +199,7 @@ async def bulk_retry_events(self, query: Mapping[str, Any]) -> Any: # ------------------------------------------------------------------ async def queue_depth( - self, *, hours: int = 24, measures: Optional[list[str]] = None + self, *, hours: int = 24, measures: list[str] | None = None ) -> Any: """GET /metrics/queue-depth over the last *hours*. diff --git a/hookdeck/cli.py b/hookdeck/cli.py index 1ac6424..6814b85 100644 --- a/hookdeck/cli.py +++ b/hookdeck/cli.py @@ -15,7 +15,7 @@ import subprocess from dataclasses import dataclass from pathlib import Path -from typing import Any, Optional +from typing import Any from .api import HookdeckAPI, HookdeckAPIError, run_sync from .constants import DEFAULT_PATH, DEFAULT_PORT @@ -44,7 +44,7 @@ def _cli_version(binary: str) -> str: out = subprocess.run( [binary, "version"], capture_output=True, text=True, timeout=10 ).stdout - except Exception: + except Exception: # noqa: BLE001 - an unknown version is reported, not raised return "" match = re.search(r"(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z.-]+))?", out) return match.group(0) if match else "" @@ -300,7 +300,7 @@ def _print_local_state(limit: int) -> None: ledger.close() -async def _resolve_connection_id(api: HookdeckAPI, value: str) -> Optional[str]: +async def _resolve_connection_id(api: HookdeckAPI, value: str) -> str | None: if value.startswith("web_") or value.startswith("con_"): return value result = await api.list_connections(name=value) diff --git a/hookdeck/constants.py b/hookdeck/constants.py index a827258..bb67a4b 100644 --- a/hookdeck/constants.py +++ b/hookdeck/constants.py @@ -17,6 +17,12 @@ SIGNATURE = "signature" SIGNATURE_2 = "signature-2" EVENT_ID = "eventid" +# Hookdeck sends this too, and it is deliberately not used as a delivery +# identity. One request fans out to one event per matching connection, so two +# routes sharing a source produce two events carrying the same request id — +# dedup keyed on it would drop the second as a duplicate. `eventid` is the only +# per-delivery identifier, which is why its absence is treated as "no id" and +# said out loud rather than papered over. REQUEST_ID = "requestid" ATTEMPT_COUNT = "attempt-count" ATTEMPT_TRIGGER = "attempt-trigger" diff --git a/hookdeck/plugin.yaml b/hookdeck/plugin.yaml index 4304f07..9c29a64 100644 --- a/hookdeck/plugin.yaml +++ b/hookdeck/plugin.yaml @@ -3,7 +3,7 @@ label: Hookdeck kind: platform version: 0.1.0 description: > - Hookdeck event gateway adapter for Hermes Agent. Hookdeck sits in front of the + Hookdeck Event Gateway adapter for Hermes Agent. Hookdeck sits in front of the agent as a durable, verified inbox: it verifies provider signatures for ~140 source types, queues events while the gateway is down, throttles delivery so a burst cannot outrun a slow agent run, and retries failures. diff --git a/hookdeck/provision.py b/hookdeck/provision.py index f706206..0bd9bf9 100644 --- a/hookdeck/provision.py +++ b/hookdeck/provision.py @@ -15,7 +15,8 @@ from __future__ import annotations -from typing import Any, Mapping, Optional +from collections.abc import Mapping +from typing import Any from .constants import RETRYABLE_STATUSES @@ -54,7 +55,7 @@ def _http_destination_config( url: str, *, - rate_limit: Optional[int], + rate_limit: int | None, rate_limit_period: str, delivery_group_key: str, group_rate: int, @@ -100,16 +101,16 @@ def build_connection_payload( mode: str = "cli", path: str = "/hookdeck", url: str = "", - events: Optional[list[str]] = None, + events: list[str] | None = None, event_path: str = "", - rate_limit: Optional[int] = None, + rate_limit: int | None = None, rate_limit_period: str = "concurrent", delivery_group_key: str = "", group_rate: int = DEFAULT_GROUP_RATE, group_rate_period: str = DEFAULT_GROUP_RATE_PERIOD, retry_count: int = DEFAULT_RETRY_COUNT, retry_interval_ms: int = DEFAULT_RETRY_INTERVAL_MS, - dedupe_window_ms: Optional[int] = DEFAULT_DEDUPE_WINDOW_MS, + dedupe_window_ms: int | None = DEFAULT_DEDUPE_WINDOW_MS, source_secret: str = "", ) -> dict[str, Any]: """Assemble the ``PUT /connections`` body for one Hermes route. @@ -168,7 +169,7 @@ def _destination_spec( mode: str, path: str, url: str, - rate_limit: Optional[int], + rate_limit: int | None, rate_limit_period: str, delivery_group_key: str, group_rate: int, @@ -220,7 +221,7 @@ def _rules( event_path: str, retry_count: int, retry_interval_ms: int, - dedupe_window_ms: Optional[int], + dedupe_window_ms: int | None, ) -> list[dict[str, Any]]: """Connection rules — where most of the reliability actually lives.""" rules: list[dict[str, Any]] = [ @@ -243,7 +244,7 @@ def _rules( def build_event_filter( events: list[str], source_type: str, event_path: str -) -> Optional[dict[str, Any]]: +) -> dict[str, Any] | None: """Turn a route's ``events`` list into a Hookdeck filter rule. Returns ``None`` when the event name's location is unknown — a wrong filter @@ -290,7 +291,7 @@ def retryable_status_codes() -> list[str]: return ordered -def _code_expression_matches(expression: str, status: int) -> Optional[bool]: +def _code_expression_matches(expression: str, status: int) -> bool | None: """Evaluate one Hookdeck retry-rule code expression against *status*. Returns True/False for a positive match, or None when the expression is an @@ -318,7 +319,7 @@ def _code_expression_matches(expression: str, status: int) -> Optional[bool]: def uncovered_statuses( - codes: Optional[list[str]], statuses: tuple[int, ...] = ADAPTER_RETRYABLE_STATUSES + codes: list[str] | None, statuses: tuple[int, ...] = ADAPTER_RETRYABLE_STATUSES ) -> list[int]: """Which of *statuses* a retry rule's ``response_status_codes`` misses. diff --git a/hookdeck/routing.py b/hookdeck/routing.py index 3436f58..652055f 100644 --- a/hookdeck/routing.py +++ b/hookdeck/routing.py @@ -6,7 +6,8 @@ from __future__ import annotations -from typing import Any, Mapping, Optional +from collections.abc import Mapping +from typing import Any from .payload import dig @@ -37,7 +38,7 @@ def route_name_from_path(path_tail: str) -> str: def resolve( routes: Routes, *, path_tail: str = "", source_name: str = "" -) -> tuple[str, Optional[dict]]: +) -> tuple[str, dict | None]: """Pick the route for a delivery, most explicit signal first. Returns the route name and its config, or the name and ``None`` when diff --git a/hookdeck/settings.py b/hookdeck/settings.py index 2fb1011..ed09ef1 100644 --- a/hookdeck/settings.py +++ b/hookdeck/settings.py @@ -11,9 +11,10 @@ from __future__ import annotations import os +from collections.abc import Mapping from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Mapping, Optional +from typing import Any from .constants import ( ACK_MODES, @@ -62,7 +63,7 @@ def load_hermes_config() -> dict: return {} -def platform_extra(config: Optional[Mapping[str, Any]] = None) -> dict: +def platform_extra(config: Mapping[str, Any] | None = None) -> dict: """``gateway.platforms.hookdeck.extra`` from a parsed config.""" parsed = load_hermes_config() if config is None else config gateway = parsed.get("gateway") or {} @@ -85,7 +86,7 @@ def configured_state_path() -> Path: LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost"}) -def is_loopback(host: Optional[str]) -> bool: +def is_loopback(host: str | None) -> bool: return bool(host) and host in LOOPBACK_HOSTS @@ -97,7 +98,7 @@ class AdapterSettings: # ── Transport ────────────────────────────────────────────────── mode: str = "cli" - host: Optional[str] = None + host: str | None = None port: int = DEFAULT_PORT path: str = DEFAULT_PATH source: str = "" @@ -129,7 +130,7 @@ class AdapterSettings: # ------------------------------------------------------------------ @classmethod - def from_extra(cls, extra: Optional[Mapping[str, Any]]) -> "AdapterSettings": + def from_extra(cls, extra: Mapping[str, Any] | None) -> AdapterSettings: """Build settings from ``platforms.hookdeck.extra``, with env fallbacks. Environment variables are a fallback rather than an override, so a @@ -201,7 +202,7 @@ def verifies_signatures(self) -> bool: return self.signing_secret not in ("", INSECURE_NO_AUTH) @property - def bind_hosts(self) -> list[Optional[str]]: + def bind_hosts(self) -> list[str | None]: """Addresses to listen on. In cli mode that is *both* loopback families. The Hookdeck CLI forwards diff --git a/hookdeck/state.py b/hookdeck/state.py index f99517f..67d80bf 100644 --- a/hookdeck/state.py +++ b/hookdeck/state.py @@ -23,7 +23,6 @@ import time from dataclasses import dataclass from pathlib import Path -from typing import Optional # Terminal and non-terminal statuses for a delivery. STATUS_RUNNING = "running" @@ -108,7 +107,7 @@ def close(self) -> None: @staticmethod def _admits( - row: Optional[sqlite3.Row], attempt: int, *, operator_initiated: bool = False + row: sqlite3.Row | None, attempt: int, *, operator_initiated: bool = False ) -> tuple[bool, str]: """Whether *attempt* is new work, given what is already recorded. @@ -141,7 +140,7 @@ def _admits( def rejection_reason( self, event_id: str, attempt: int, *, operator_initiated: bool = False - ) -> Optional[str]: + ) -> str | None: """Read-only: why :meth:`admit` would reject this delivery, if it would. Lets a caller recognise a repeat *without* recording anything — which @@ -294,7 +293,7 @@ def record_cancelled( ) self._conn.commit() - def get(self, event_id: str) -> Optional[sqlite3.Row]: + def get(self, event_id: str) -> sqlite3.Row | None: with self._lock: return self._conn.execute( "SELECT * FROM deliveries WHERE event_id = ?", (event_id,) @@ -332,7 +331,7 @@ def cancel_scheduled_resume(self, connection_id: str) -> None: ) self._conn.commit() - def due_resumes(self, now: Optional[float] = None) -> list[sqlite3.Row]: + def due_resumes(self, now: float | None = None) -> list[sqlite3.Row]: moment = time.time() if now is None else now with self._lock: return self._conn.execute( diff --git a/hookdeck/tools.py b/hookdeck/tools.py index fe276d5..5a32068 100644 --- a/hookdeck/tools.py +++ b/hookdeck/tools.py @@ -16,7 +16,7 @@ import logging import threading import time -from typing import Any, Optional +from typing import Any from .api import HookdeckAPI, HookdeckAPIError @@ -40,7 +40,7 @@ def _run(coro: Any) -> Any: def _worker() -> None: try: box["value"] = asyncio.run(coro) - except BaseException as exc: + except BaseException as exc: # noqa: BLE001 - re-raised on the caller's thread box["error"] = exc thread = threading.Thread(target=_worker, daemon=True) @@ -115,12 +115,12 @@ def _models(result: Any) -> list[dict]: def _guard(fn: Any) -> Any: """Turn API errors into a message the model can act on.""" - def wrapper(args: Optional[dict] = None, **_: Any) -> str: + def wrapper(args: dict | None = None, **_: Any) -> str: try: return fn(args or {}) except HookdeckAPIError as exc: return f"Hookdeck API error: {exc}" - except Exception as exc: + except Exception as exc: # noqa: BLE001 - the model gets a message, not a traceback return f"Hookdeck tool failed: {exc}" wrapper.__name__ = getattr(fn, "__name__", "hookdeck_tool") diff --git a/hookdeck/tunnel.py b/hookdeck/tunnel.py index 119b508..ca72129 100644 --- a/hookdeck/tunnel.py +++ b/hookdeck/tunnel.py @@ -16,7 +16,6 @@ import logging import os import shutil -from typing import Optional logger = logging.getLogger(__name__) @@ -68,8 +67,8 @@ def __init__( self._api_key = api_key or os.getenv("HOOKDECK_API_KEY", "") self._login_enabled = login self._binary = binary - self._process: Optional[asyncio.subprocess.Process] = None - self._supervisor: Optional[asyncio.Task] = None + self._process: asyncio.subprocess.Process | None = None + self._supervisor: asyncio.Task | None = None self._stopping = False # ------------------------------------------------------------------ @@ -171,7 +170,7 @@ async def _login(self, binary: str) -> None: ) except asyncio.TimeoutError: logger.warning("[hookdeck] `hookdeck ci` timed out after 30s") - except Exception as exc: # pragma: no cover - environment dependent + except Exception as exc: # noqa: BLE001 # pragma: no cover - environment dependent logger.warning("[hookdeck] `hookdeck ci` failed: %s", exc) async def _supervise(self, binary: str) -> None: @@ -182,7 +181,7 @@ async def _supervise(self, binary: str) -> None: await self._run_once(binary) except asyncio.CancelledError: raise - except Exception as exc: + except Exception as exc: # noqa: BLE001 - the supervisor restarts, never dies logger.error("[hookdeck] CLI tunnel error: %s", exc) if self._stopping: diff --git a/hookdeck/verify.py b/hookdeck/verify.py index b81c7d9..53a49d5 100644 --- a/hookdeck/verify.py +++ b/hookdeck/verify.py @@ -17,7 +17,7 @@ import base64 import hashlib import hmac -from typing import Iterable, Mapping +from collections.abc import Iterable, Mapping from .constants import DEFAULT_HEADER_PREFIX, SIGNATURE, SIGNATURE_2, header_name @@ -29,11 +29,26 @@ def compute_signature(body: bytes, secret: str) -> str: def _matches_any(expected: str, provided: Iterable[str]) -> bool: + """Whether any candidate is *expected*, compared as bytes. + + Bytes rather than ``str`` because ``compare_digest`` refuses two ``str`` + arguments unless both are pure ASCII, and a candidate is a header value an + unauthenticated sender controls outright. One non-ASCII byte in it used to + raise ``TypeError`` straight out of the verification path, which aiohttp + turned into a 500 — wrong twice over: it bypassed the + ``EMITTED_STATUS_RETRYABLE`` guard every other response goes through, and + 500 sits inside the provisioned retry rule's ``500-599`` range, so a + malformed request was retried rather than refused. + + Encoding cannot fail on either side: *expected* is base64, and ``replace`` + maps an undecodable candidate to bytes that simply do not match. + """ + wanted = expected.encode("ascii") # compare_digest on every candidate rather than short-circuiting, so the # work done does not depend on which header happened to match. matched = False for candidate in provided: - if candidate and hmac.compare_digest(expected, candidate): + if candidate and hmac.compare_digest(wanted, candidate.encode("utf-8", "replace")): matched = True return matched diff --git a/pyproject.toml b/pyproject.toml index 1fce3cf..3e2f5b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,10 @@ [project] name = "hermes-hookdeck" -version = "0.1.0" -description = "Hookdeck event gateway plugin for Hermes Agent — verified, queued, retryable webhook triggers" +# Read from `hookdeck.__version__` rather than repeated here. Two copies drift, +# and the one that drifts is always the one nobody looks at — `hermes plugins` +# reports the module's, PyPI reports this one. +dynamic = ["version"] +description = "Hookdeck Event Gateway plugin for Hermes Agent — verified, queued, retryable webhook triggers" readme = "README.md" requires-python = ">=3.10" license = { file = "LICENSE" } @@ -22,12 +25,57 @@ hookdeck = "hookdeck" # fastapi is provided by the Hermes web server at runtime, not by this # plugin — it is a dev dependency only, so the dashboard routes can be # tested without a Hermes checkout. -dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "fastapi>=0.110"] +dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "fastapi>=0.110", "ruff>=0.16"] [tool.pytest.ini_options] asyncio_mode = "auto" testpaths = ["tests"] +[tool.ruff] +# 88 is what the code was already written to; the handful of lines over it are +# long string literals, which `E501` is told to leave alone below. +line-length = 88 +target-version = "py310" + +[tool.ruff.lint] +select = [ + "E", # pycodestyle + "W", # pycodestyle warnings + "F", # pyflakes — unused imports and names, the ones that are always bugs + "I", # import sorting + "UP", # pyupgrade, bounded by target-version + "B", # bugbear — mutable defaults, `except` that swallows, loop closures + "C4", # comprehensions + "BLE", # blind `except` — see below + "RUF", # ruff's own, mostly correctness +] +ignore = [ + # Off rather than enforced. `line-length` still guides `ruff format` for + # anyone who runs it, but the repo is not formatter-managed: turning E501 + # into an error means either reformatting every file at once, burying the + # history, or a wave of awkward wraps through argparse calls and log + # statements that read worse split. Nothing here is long by accident. + "E501", + # `raise ... from` is used where the chain is informative and deliberately + # omitted where the original exception is noise. + "B904", +] + +# BLE001 is selected on purpose rather than ignored. This adapter degrades +# instead of aborting in a dozen places — a failed API call at boot, a failed +# maintenance tick, a plugin surface that will not register — and every one of +# those is a decision, not an oversight. Selecting the rule means each blind +# `except` carries a `noqa: BLE001` saying which it is, and a new one has to be +# argued for rather than merged unnoticed. + +[tool.ruff.lint.per-file-ignores] +# The `hermes_stub` import in conftest runs for its side effect and must happen +# before anything imports the adapter; `plugin_api` imports its siblings only +# after `_load_plugin_package()`. Both are load-bearing order, and both already +# say so with an inline `noqa: E402` at the import itself, which is where a +# reader needs it. +"tests/*" = ["E402"] + [tool.setuptools] # The manifest, the dashboard bundle and the bundled skill are the plugin as # much as the Python is — a package that ships only the modules installs @@ -35,6 +83,9 @@ testpaths = ["tests"] packages = ["hookdeck", "hookdeck.dashboard"] include-package-data = true +[tool.setuptools.dynamic] +version = { attr = "hookdeck.__version__" } + [tool.setuptools.package-data] hookdeck = [ "plugin.yaml", diff --git a/scripts/check_upstream_contract.py b/scripts/check_upstream_contract.py new file mode 100644 index 0000000..e20ab96 --- /dev/null +++ b/scripts/check_upstream_contract.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python3 +"""Check that Hermes still provides everything this plugin borrows from it. + +The plugin lives outside the Hermes tree and its tests run against +``tests/hermes_stub.py``, which is the only way to exercise the ingest path +without a Hermes checkout. The cost of that is a blind spot: the stub cannot +notice when the real ``WebhookAdapter`` renames a method the adapter overrides +or calls, and the first symptom would be a gateway that fails to start. + +So this reads the upstream source and asserts each borrowed name is still +there. It parses rather than imports — importing Hermes would mean installing +its whole dependency tree to answer a question about names. + +It is a smoke alarm, not a type checker. A method that keeps its name and +changes its signature or semantics still gets through, which is why the README +points at a real end-to-end run as the thing that actually proves integration. + + python scripts/check_upstream_contract.py path/to/hermes-agent +""" + +from __future__ import annotations + +import ast +import sys +from pathlib import Path + +#: module path -> names the plugin imports or overrides from it. +#: +#: Sources, in order: the `from gateway...` imports at the top of adapter.py, +#: the attributes hermes_stub.py reproduces, and the lazy `agent.skill_commands` +#: import in `_apply_skills`. +CONTRACT: dict[str, dict[str, list[str]]] = { + "gateway/config.py": { + "module": ["Platform", "PlatformConfig"], + }, + "gateway/platforms/base.py": { + "module": [ + "BasePlatformAdapter", + "MessageEvent", + "MessageType", + "ProcessingOutcome", + "SendResult", + "SessionSource", + ], + # Inherited and called by HookdeckAdapter, or driven by its tests. + "BasePlatformAdapter": [ + "build_source", + "handle_message", + "on_processing_complete", + "_mark_connected", + "_mark_disconnected", + ], + }, + "gateway/platforms/webhook.py": { + "module": ["WebhookAdapter"], + "WebhookAdapter": [ + # Overridden. + "on_processing_complete", + # Called from the delivery path; renaming any of these breaks + # ingest without breaking startup, which is the worse failure. + "_render_prompt", + "_render_delivery_extra", + "_direct_deliver", + "_prune_delivery_info", + ], + }, + "agent/skill_commands.py": { + "module": ["build_skill_invocation_message", "get_skill_commands"], + }, +} + +#: Attributes the adapter reads or writes on `self` that the base classes own. +#: Assigned in `__init__` upstream, so they are found by scanning assignments +#: rather than definitions. +INHERITED_ATTRIBUTES: dict[str, list[str]] = { + "gateway/platforms/webhook.py": [ + "_route_processor", + "_delivery_info", + "_delivery_info_created", + "_delivery_info_order", + ], + "gateway/platforms/base.py": [ + "_background_tasks", + ], +} + + +def _module_level_names(tree: ast.Module) -> set[str]: + names: set[str] = set() + for node in tree.body: + if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + names.add(node.name) + elif isinstance(node, ast.Assign): + names.update(t.id for t in node.targets if isinstance(t, ast.Name)) + elif isinstance(node, ast.ImportFrom): + names.update(a.asname or a.name for a in node.names) + return names + + +def _class_member_names(tree: ast.Module, class_name: str) -> set[str]: + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef) and node.name == class_name: + return { + child.name + for child in node.body + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + return set() + + +def _self_assigned_names(tree: ast.Module) -> set[str]: + """Every ``self.x = ...`` in the file, which is where base classes declare state.""" + names: set[str] = set() + for node in ast.walk(tree): + targets: list[ast.expr] = [] + if isinstance(node, ast.Assign): + targets = list(node.targets) + elif isinstance(node, ast.AnnAssign): + targets = [node.target] + for target in targets: + if ( + isinstance(target, ast.Attribute) + and isinstance(target.value, ast.Name) + and target.value.id == "self" + ): + names.add(target.attr) + return names + + +def check(root: Path) -> list[str]: + problems: list[str] = [] + + for rel, expectations in CONTRACT.items(): + path = root / rel + if not path.exists(): + problems.append(f"{rel}: module is gone") + continue + tree = ast.parse(path.read_text(encoding="utf-8")) + + for scope, expected in expectations.items(): + found = ( + _module_level_names(tree) + if scope == "module" + else _class_member_names(tree, scope) + ) + if scope != "module" and not found: + problems.append(f"{rel}: class {scope} is gone") + continue + for name in expected: + if name not in found: + where = rel if scope == "module" else f"{rel}:{scope}" + problems.append(f"{where}: {name} is gone") + + for rel, attributes in INHERITED_ATTRIBUTES.items(): + path = root / rel + if not path.exists(): + continue # already reported above + assigned = _self_assigned_names(ast.parse(path.read_text(encoding="utf-8"))) + for attribute in attributes: + if attribute not in assigned: + problems.append(f"{rel}: self.{attribute} is no longer assigned") + + return problems + + +def main() -> int: + if len(sys.argv) != 2: + print(__doc__) + return 2 + root = Path(sys.argv[1]) + if not (root / "gateway").is_dir(): + print(f"! {root} does not look like a hermes-agent checkout") + return 2 + + problems = check(root) + if not problems: + print(f"✓ every name this plugin borrows is still in {root}") + return 0 + + print(f"✗ {len(problems)} name(s) this plugin depends on have moved:\n") + for problem in problems: + print(f" {problem}") + print( + "\nUpdate the adapter and tests/hermes_stub.py together — the stub" + "\nmatching upstream is the only thing making the test suite mean" + "\nanything." + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/hermes_stub.py b/tests/hermes_stub.py index 2abd921..476a203 100644 --- a/tests/hermes_stub.py +++ b/tests/hermes_stub.py @@ -24,8 +24,7 @@ from collections import deque from dataclasses import dataclass, field from enum import Enum -from typing import Any, Dict, Optional - +from typing import Any # Names the registry has been told about. Real Hermes mints a Platform member # only for a registered plugin name — "arbitrary strings are rejected to @@ -56,7 +55,7 @@ def _missing_(cls, value): @dataclass class PlatformConfig: enabled: bool = True - extra: Dict[str, Any] = field(default_factory=dict) + extra: dict[str, Any] = field(default_factory=dict) class MessageType(Enum): @@ -73,11 +72,11 @@ class ProcessingOutcome(Enum): class SessionSource: platform: Any = None chat_id: str = "" - chat_name: Optional[str] = None + chat_name: str | None = None chat_type: str = "dm" - user_id: Optional[str] = None - user_name: Optional[str] = None - profile: Optional[str] = None + user_id: str | None = None + user_name: str | None = None + profile: str | None = None @dataclass @@ -86,15 +85,15 @@ class MessageEvent: message_type: Any = MessageType.TEXT source: Any = None raw_message: Any = None - message_id: Optional[str] = None - metadata: Dict[str, Any] = field(default_factory=dict) + message_id: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) @dataclass class SendResult: success: bool - message_id: Optional[str] = None - error: Optional[str] = None + message_id: str | None = None + error: str | None = None class _RouteProcessor: @@ -154,11 +153,11 @@ def __init__(self, config: PlatformConfig): extra = config.extra or {} self._host = extra.get("host") or None self._port = int(extra.get("port", 8787)) - self._routes: Dict[str, dict] = dict(extra.get("routes") or {}) + self._routes: dict[str, dict] = dict(extra.get("routes") or {}) self._max_body_bytes = int(extra.get("max_body_bytes", 1_048_576)) self._route_processor = _RouteProcessor() - self._delivery_info: Dict[str, dict] = {} - self._delivery_info_created: Dict[str, float] = {} + self._delivery_info: dict[str, dict] = {} + self._delivery_info_created: dict[str, float] = {} self._delivery_info_order: deque = deque() self.direct_deliveries: list = [] self.direct_deliver_result = SendResult(success=True) diff --git a/tests/test_adapter.py b/tests/test_adapter.py index 5d75e29..e6d9b9d 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py @@ -123,6 +123,24 @@ async def test_delivery_signed_with_the_wrong_secret_is_rejected(client_factory) assert response.status == 401 +async def test_a_malformed_signature_is_refused_not_a_server_error(client_factory): + """A junk signature must answer 401, like any other forged one. + + A non-ASCII byte in the header used to raise out of the verification path + and become a 500 — which skips ``assert_declared_status`` entirely and, + because the provisioned retry rule covers ``500-599``, had Hookdeck retry + an unauthenticated request instead of dropping it. + """ + _adapter, client = await client_factory({"default": {}}) + response = await post( + client, + {"hello": "world"}, + sign=False, + extra_headers={"x-hookdeck-signature": "not-base64-é"}, + ) + assert response.status == 401 + + async def test_unmatched_source_returns_404(client_factory): _adapter, client = await client_factory( {"github": {"source": "github"}, "stripe": {"source": "stripe"}} @@ -891,6 +909,38 @@ async def test_a_delivery_with_no_event_id_is_processed_but_warned_about( assert "without deduplication or retry" in caplog.text +async def test_a_proxys_request_id_is_not_mistaken_for_an_event_id( + client_factory, caplog +): + """`X-Request-ID` is not a Hookdeck identifier and must not stand in. + + Anything in front of the gateway can set it, it is not subject to + `header_prefix`, and one Hookdeck request fans out to one event per + matching connection — so it is not even unique per delivery. Accepting it + would key the ledger on the wrong thing and silence the warning that tells + an operator their `header_prefix` is wrong. + """ + adapter, client = await client_factory({"default": {}}) + seen: list[Any] = [] + adapter.run_agent = lambda event: _record(seen, event) + + raw = json.dumps({"n": 1}).encode() + response = await client.post( + "/hookdeck", + data=raw, + headers={ + "content-type": "application/json", + "x-hookdeck-signature": compute_signature(raw, SECRET), + "X-Request-ID": "req_from_some_proxy", + }, + ) + assert response.status == 202 + await _settle() + assert "without deduplication or retry" in caplog.text + assert adapter._ledger is not None + assert adapter._ledger.get("req_from_some_proxy") is None + + # ---------------------------------------------------------------------- # Deduplication precedence # ---------------------------------------------------------------------- @@ -1032,6 +1082,59 @@ async def _slow_failure(_event): assert adapter._api.retried == ["evt_slow_fail"] +async def test_an_abandoned_hand_back_still_clears_its_sync_marker( + client_factory, monkeypatch +): + """The marker goes once the hand-back decision is made, not once it works. + + Keeping it on the failure path grows the set for the life of the process, + and a later sync-mode run of the same id would be treated as having been + acked early when it was not. + """ + from hookdeck.api import HookdeckAPIError + + monkeypatch.setattr("hookdeck.adapter._REDELIVERY_RETRY_INITIAL_SECONDS", 0.0) + adapter, client = await client_factory( + {"default": {}}, ack_mode="sync", sync_timeout_seconds=0.05 + ) + adapter._api.fail_with = HookdeckAPIError(0, "POST", "/events/x/retry", "no route") + gate = asyncio.Event() + + async def _slow_failure(_event): + await gate.wait() + return ProcessingOutcome.FAILURE + + adapter.run_agent = _slow_failure + assert (await post(client, {"n": 1}, event_id="evt_abandoned")).status == 202 + assert "evt_abandoned" in adapter._acked_before_completion + + gate.set() + await _settle() + assert adapter._api.retried == [] + assert adapter._acked_before_completion == set() + + +async def test_an_exhausted_event_clears_its_sync_marker_too(client_factory): + adapter, client = await client_factory( + {"default": {}}, + ack_mode="sync", + sync_timeout_seconds=0.05, + max_agent_retries=0, + ) + gate = asyncio.Event() + + async def _slow_failure(_event): + await gate.wait() + return ProcessingOutcome.FAILURE + + adapter.run_agent = _slow_failure + assert (await post(client, {"n": 1}, event_id="evt_spent")).status == 202 + + gate.set() + await _settle() + assert adapter._acked_before_completion == set() + + async def test_a_sync_failure_within_the_timeout_is_left_to_hookdeck(client_factory): # The 5xx response *is* the retry request there; asking again would double # the redeliveries. diff --git a/tests/test_cli.py b/tests/test_cli.py index bff21ef..87886ae 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -6,19 +6,19 @@ def _args(**kwargs) -> argparse.Namespace: - defaults = dict( - route="", - all=False, - source="", - source_type="", - mode="", - url="", - path="", - rate_limit=None, - rate_limit_period="concurrent", - group_key="", - dry_run=True, - ) + defaults = { + "route": "", + "all": False, + "source": "", + "source_type": "", + "mode": "", + "url": "", + "path": "", + "rate_limit": None, + "rate_limit_period": "concurrent", + "group_key": "", + "dry_run": True, + } defaults.update(kwargs) return argparse.Namespace(hookdeck_action="setup", **defaults) diff --git a/tests/test_verify.py b/tests/test_verify.py index fcd5f02..289ad3f 100644 --- a/tests/test_verify.py +++ b/tests/test_verify.py @@ -51,6 +51,20 @@ def test_an_empty_secret_never_passes(): assert not verify_signature(signed, BODY, "") +def test_a_non_ascii_signature_is_rejected_not_raised(): + # The sender controls this header outright. compare_digest refuses two + # non-ASCII strs, and letting that TypeError escape turned a forged + # signature into a 500 — which the retry rule then treats as retryable. + assert not verify_signature(headers(**{"x-hookdeck-signature": "abcé"}), BODY, SECRET) + + +def test_a_lone_surrogate_signature_is_rejected_not_raised(): + # Not reachable through aiohttp today, but the guard is one encode call and + # the failure mode it prevents is a 500 on the security-critical path. + signed = headers(**{"x-hookdeck-signature": "\ud800bad"}) + assert not verify_signature(signed, BODY, SECRET) + + def test_honours_a_custom_header_prefix(): signed = headers(**{"x-acme-signature": compute_signature(BODY, SECRET)}) assert verify_signature(signed, BODY, SECRET, prefix="x-acme")