diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7ebc8f4..3c9514e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.5.0
+
+### Added
+
+- `halvorai_ingest.device`, the backend half of mobile apps built with `halvorai_ingest_flutter`, whose installs sign their own events:
+  - `register_device_source` registers an install's public Ed25519 key (a key with `d` is refused).
+  - `forward_envelopes` checks each event's owner and environment, then passes the batch to `POST /v1/events` byte for byte and returns HalvorAI's answer unchanged.
+  - Both have `_async` versions for FastAPI and other async frameworks. Same behaviour as `@halvorai/ingest/device` 0.4.0.
+
+### Changed
+
+- `init` registers its source through the same code. A refused registration still raises `RuntimeError` (now `SourceRegistrationError`, a subclass), and a network failure says `POST /v1/sources failed: <reason>`.
+
 ## 0.4.1
 
 ### Fixed
diff --git a/README.md b/README.md
index 8bd3d5a..7c0ed9d 100644
--- a/README.md
+++ b/README.md
@@ -115,6 +115,61 @@ Each 402 challenge becomes a `payment.requested` event. The payment that answers
 
 Stripe, MPP and MCP adapters are in the Node.js SDK, [`@halvorai/ingest`](https://github.com/halvorAI/ingest-js). Guides for each: [docs.halvorai.com](https://docs.halvorai.com).
 
+## Mobile apps (Flutter)
+
+Apps built with [`halvorai_ingest_flutter`](https://github.com/halvorAI/ingest-dart) sign events on the phone, with a key that each install makes for itself. The app never holds your API key. Your backend registers each install's public key and forwards its signed batches; `halvorai_ingest.device` provides both. With FastAPI:
+
+```python
+from fastapi import FastAPI, Request, Response
+
+from halvorai_ingest.device import (
+    DeviceRequestError,
+    forward_envelopes_async,
+    register_device_source_async,
+)
+
+app = FastAPI()
+UNAUTHORIZED = b'{"error":{"code":"unauthorized","message":"sign in first"}}'
+
+
+@app.post("/halvor/sources")  # the app's provisionUrl
+async def halvor_sources(request: Request) -> Response:
+    user = await current_user(request)  # your own session check
+    if user is None:
+        return Response(UNAUTHORIZED, status_code=401, media_type="application/json")
+    body = await request.json()
+    try:
+        source_id = await register_device_source_async(body.get("public_jwk"), name=body.get("name"))
+    except DeviceRequestError as exc:
+        return Response(exc.body(), status_code=exc.status, media_type="application/json")
+    return Response(f'{{"source_id":"{source_id}"}}', media_type="application/json")
+
+
+@app.post("/halvor/events")  # the app's forwardUrl
+async def halvor_events(request: Request) -> Response:
+    user = await current_user(request)
+    if user is None:
+        return Response(UNAUTHORIZED, status_code=401, media_type="application/json")
+    out = await forward_envelopes_async(await request.body(), expected_owner_id=user.id)
+    return Response(out.body, status_code=out.status, media_type=out.content_type)
+```
+
+Both functions read `HALVOR_API_KEY` (and `HALVOR_ENDPOINT`) from the environment, or take `api_key=`. Sync versions without `_async` work the same way in Flask or Django.
+
+What they do:
+
+- **`register_device_source`** accepts only the public half of an Ed25519 key, refusing a key with `d`. It registers the key with `POST /v1/sources` and returns the `src_` id.
+- **`forward_envelopes`**:
+  - Checks that every event's `owner.id` is the signed-in user and its `environment` is the key's.
+  - Sends the request body to `POST /v1/events` **byte for byte**, never a parsed and re-serialized copy, so the device's signatures stay valid.
+  - Returns HalvorAI's status and body unchanged, which the app reads its per-event results from.
+- **Status codes tell the app what to do:**
+  - 401 or 403 means "try later": the app keeps the batch on disk and retries.
+  - 400, 413 and 422 are permanent: the app drops the batch, or splits it on 413.
+  - 502 means HalvorAI couldn't be reached, and the app retries.
+
+Set the signed-in user as `owner` in the app's agent context; events without it are refused.
+
 ## Develop
 
 ```bash
diff --git a/halvorai_ingest/device.py b/halvorai_ingest/device.py
new file mode 100644
index 0000000..71ac061
--- /dev/null
+++ b/halvorai_ingest/device.py
@@ -0,0 +1,269 @@
+"""Server-side helpers for apps whose installs sign their own events (`halvorai_ingest_flutter`).
+
+The app never holds your API key. Your backend:
+
+- registers each install's public key (`register_device_source`) and answers the app's
+  provisioning request with `{"source_id": "src_…"}`;
+- checks and forwards each signed batch (`forward_envelopes`), answering with HalvorAI's response
+  unchanged.
+
+Both come in sync and async forms; use the async ones in FastAPI and other async frameworks.
+The status codes tell the app what to do: 401 or 403 (your own session check) means try later,
+400, 413 and 422 are permanent, and 502 means HalvorAI could not be reached.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+from .client import API_KEY_PATTERN, default_endpoint
+from .sources import SourceRegistrationError, post_source, post_source_async
+
+DEFAULT_MAX_ENVELOPES = 100
+DEFAULT_MAX_BYTES = 1_000_000
+
+_BASE64URL_32_BYTES = re.compile(r"^[A-Za-z0-9_-]{43}$")
+
+
+class DeviceRequestError(Exception):
+    """A request the app sent that your backend must answer with `status`.
+
+    `body()` is HalvorAI's error format, so the Flutter SDK reports `code` as `serverCode`.
+    """
+
+    def __init__(self, status: int, code: str, message: str) -> None:
+        super().__init__(message)
+        self.status = status
+        self.code = code
+        self.message = message
+
+    def body(self) -> bytes:
+        return json.dumps({"error": {"code": self.code, "message": self.message}}).encode("utf-8")
+
+
+@dataclass(frozen=True)
+class ForwardResult:
+    """What to answer the app: HalvorAI's response unchanged, or why the batch was not sent."""
+
+    status: int
+    body: bytes
+    content_type: str
+
+
+@dataclass(frozen=True)
+class _Config:
+    api_key: str
+    environment: str
+    endpoint: str
+
+
+def _config(api_key: str | None, endpoint: str | None) -> _Config:
+    """A misconfigured backend is a programming error, so this raises ValueError."""
+    key = api_key or os.environ.get("HALVOR_API_KEY")
+    if not key or not API_KEY_PATTERN.match(key):
+        raise ValueError("HalvorAI api key is missing or malformed (expected hk_test_… or hk_live_…)")
+    environment = "live" if key.startswith("hk_live_") else "test"
+    base = endpoint or os.environ.get("HALVOR_ENDPOINT") or default_endpoint(key, environment)
+    base = base.rstrip("/")
+    if base.endswith("/v1/events"):
+        base = base[: -len("/v1/events")]
+    return _Config(key, environment, base)
+
+
+def public_device_jwk(jwk: object) -> dict[str, str]:
+    """The public half of an Ed25519 key, reduced to `kty`, `crv` and `x`.
+
+    A key with `d` is refused: the app sent its private key, which must never leave the device.
+    """
+    if not isinstance(jwk, dict):
+        raise DeviceRequestError(400, "invalid_request", "public_jwk must be a JSON object")
+    if "d" in jwk:
+        raise DeviceRequestError(
+            400, "invalid_request", "public_jwk contains a private key (d); only the public half may be sent"
+        )
+    if jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519":
+        raise DeviceRequestError(400, "invalid_request", "public_jwk must be an Ed25519 key")
+    x = jwk.get("x")
+    if not isinstance(x, str) or not _BASE64URL_32_BYTES.match(x):
+        raise DeviceRequestError(400, "invalid_request", "public_jwk.x must be 32 bytes, base64url")
+    return {"kty": "OKP", "crv": "Ed25519", "x": x}
+
+
+def _name(name: str | None) -> str:
+    return ((name or "").strip() or "app install")[:100]
+
+
+def register_device_source(
+    public_jwk: object,
+    *,
+    name: str | None = None,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    transport: httpx.BaseTransport | None = None,
+) -> str:
+    """Register an app install's public key and return the `src_` id the app signs with.
+
+    Call it from the route the app's provisioning request reaches, after checking the user is
+    signed in. Raises `DeviceRequestError`: 400 for a key the app should not have sent, 502 when
+    HalvorAI does not register it (the app retries).
+    """
+    config = _config(api_key, endpoint)
+    jwk = public_device_jwk(public_jwk)
+    try:
+        with httpx.Client(timeout=30.0, transport=transport) as http:
+            return post_source(http, config.endpoint, config.api_key, _name(name), jwk)
+    except SourceRegistrationError as exc:
+        raise DeviceRequestError(502, "upstream_error", str(exc)) from exc
+
+
+async def register_device_source_async(
+    public_jwk: object,
+    *,
+    name: str | None = None,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> str:
+    """`register_device_source` for async frameworks."""
+    config = _config(api_key, endpoint)
+    jwk = public_device_jwk(public_jwk)
+    try:
+        async with httpx.AsyncClient(timeout=30.0, transport=transport) as http:
+            return await post_source_async(http, config.endpoint, config.api_key, _name(name), jwk)
+    except SourceRegistrationError as exc:
+        raise DeviceRequestError(502, "upstream_error", str(exc)) from exc
+
+
+def _check_batch(body: bytes, expected_owner_id: str, environment: str, max_envelopes: int, max_bytes: int) -> None:
+    if len(body) > max_bytes:
+        raise DeviceRequestError(413, "payload_too_large", f"body is over {max_bytes} bytes")
+    try:
+        parsed: Any = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError):
+        raise DeviceRequestError(400, "invalid_request", "body is not UTF-8 JSON") from None
+    events = parsed.get("events") if isinstance(parsed, dict) else None
+    if not isinstance(events, list) or not events:
+        raise DeviceRequestError(400, "invalid_request", 'body must be {"events":[…]} with at least one envelope')
+    if len(events) > max_envelopes:
+        raise DeviceRequestError(413, "payload_too_large", f"more than {max_envelopes} envelopes")
+    for i, envelope in enumerate(events):
+        event = envelope.get("event") if isinstance(envelope, dict) else None
+        if not isinstance(event, dict) or not isinstance(envelope.get("sig"), str):
+            raise DeviceRequestError(400, "invalid_request", f'events[{i}] must be {{"event":{{…}},"sig":"…"}}')
+        owner = event.get("owner")
+        if not isinstance(owner, dict) or owner.get("id") != expected_owner_id:
+            raise DeviceRequestError(422, "owner_mismatch", f"events[{i}].event.owner.id is not the signed-in user")
+        if event.get("environment") != environment:
+            raise DeviceRequestError(
+                422,
+                "environment_mismatch",
+                f"events[{i}].event.environment must be {environment}, the API key's environment",
+            )
+
+
+def _prepare(
+    body: bytes | str,
+    expected_owner_id: str,
+    api_key: str | None,
+    endpoint: str | None,
+    max_envelopes: int,
+    max_bytes: int,
+) -> tuple[_Config, bytes, ForwardResult | None]:
+    config = _config(api_key, endpoint)
+    raw = body.encode("utf-8") if isinstance(body, str) else bytes(body)
+    try:
+        _check_batch(raw, expected_owner_id, config.environment, max_envelopes, max_bytes)
+    except DeviceRequestError as exc:
+        return config, raw, _error(exc)
+    return config, raw, None
+
+
+def _error(exc: DeviceRequestError) -> ForwardResult:
+    return ForwardResult(exc.status, exc.body(), "application/json")
+
+
+def _headers(config: _Config) -> dict[str, str]:
+    return {
+        "Authorization": f"Bearer {config.api_key}",
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+
+
+def _result(res: httpx.Response) -> ForwardResult:
+    return ForwardResult(res.status_code, res.content, res.headers.get("content-type", "application/json"))
+
+
+def _unreachable(exc: Exception) -> ForwardResult:
+    return _error(DeviceRequestError(502, "upstream_error", f"HalvorAI could not be reached: {exc}"))
+
+
+def forward_envelopes(
+    body: bytes | str,
+    *,
+    expected_owner_id: str,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    max_envelopes: int = DEFAULT_MAX_ENVELOPES,
+    max_bytes: int = DEFAULT_MAX_BYTES,
+    transport: httpx.BaseTransport | None = None,
+) -> ForwardResult:
+    """Check a batch an app install sent and pass it to `POST /v1/events` byte for byte.
+
+    Returns HalvorAI's status and body unchanged; the app reads its per-event results from them.
+    The body is parsed only to check it: re-serializing could change the bytes the device signed.
+
+    Refused without calling HalvorAI, in HalvorAI's error format: 400 `invalid_request` for a body
+    that is not `{"events":[{"event":…,"sig":…}]}`, 413 `payload_too_large` above `max_envelopes`
+    or `max_bytes` (the app splits the batch), 422 `owner_mismatch` when an event's `owner.id` is
+    not `expected_owner_id`, 422 `environment_mismatch` when its `environment` is not the key's.
+    502 when HalvorAI cannot be reached. Answer 401 or 403 yourself for a session that is not valid.
+    """
+    config, raw, refused = _prepare(body, expected_owner_id, api_key, endpoint, max_envelopes, max_bytes)
+    if refused is not None:
+        return refused
+    try:
+        with httpx.Client(timeout=30.0, transport=transport) as http:
+            res = http.post(f"{config.endpoint}/v1/events", content=raw, headers=_headers(config))
+    except httpx.HTTPError as exc:
+        return _unreachable(exc)
+    return _result(res)
+
+
+async def forward_envelopes_async(
+    body: bytes | str,
+    *,
+    expected_owner_id: str,
+    api_key: str | None = None,
+    endpoint: str | None = None,
+    max_envelopes: int = DEFAULT_MAX_ENVELOPES,
+    max_bytes: int = DEFAULT_MAX_BYTES,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> ForwardResult:
+    """`forward_envelopes` for async frameworks."""
+    config, raw, refused = _prepare(body, expected_owner_id, api_key, endpoint, max_envelopes, max_bytes)
+    if refused is not None:
+        return refused
+    try:
+        async with httpx.AsyncClient(timeout=30.0, transport=transport) as http:
+            res = await http.post(f"{config.endpoint}/v1/events", content=raw, headers=_headers(config))
+    except httpx.HTTPError as exc:
+        return _unreachable(exc)
+    return _result(res)
+
+
+__all__ = [
+    "DeviceRequestError",
+    "ForwardResult",
+    "forward_envelopes",
+    "forward_envelopes_async",
+    "public_device_jwk",
+    "register_device_source",
+    "register_device_source_async",
+]
diff --git a/halvorai_ingest/init.py b/halvorai_ingest/init.py
index cbaabd8..3e98e8f 100644
--- a/halvorai_ingest/init.py
+++ b/halvorai_ingest/init.py
@@ -13,6 +13,7 @@
 import httpx
 
 from .client import API_KEY_PATTERN, HalvorClient, default_endpoint
+from .sources import post_source
 from .sign import generate_source_key
 
 DASHBOARD_URL = "https://app.halvorai.com"
@@ -67,22 +68,6 @@ def upsert_env_file(path: Path, values: dict[str, str]) -> None:
     path.write_text(content, encoding="utf-8")
 
 
-def _register_source(
-    http: httpx.Client, endpoint: str, api_key: str, name: str, public_jwk: dict[str, str]
-) -> str:
-    res = http.post(
-        f"{endpoint}/v1/sources",
-        json={"name": name, "adapter": "custom", "public_jwk": public_jwk},
-        headers={"Authorization": f"Bearer {api_key}", "Accept": "application/json"},
-    )
-    if res.status_code != 201:
-        raise RuntimeError(f"POST /v1/sources failed ({res.status_code}): {res.text or res.reason_phrase}")
-    source_id = res.json().get("id")
-    if not isinstance(source_id, str) or not source_id.startswith("src_"):
-        raise RuntimeError("POST /v1/sources returned no source id")
-    return source_id
-
-
 def run_init(
     api_key: str,
     *,
@@ -110,7 +95,7 @@ def run_init(
 
     log(f"Registering source at {base}…")
     with httpx.Client(timeout=30.0, transport=transport) as http:
-        source_id = _register_source(http, base, api_key, source_name, public_jwk)
+        source_id = post_source(http, base, api_key, source_name, public_jwk)
     private_jwk["kid"] = source_id
 
     env_path = Path(cwd).resolve() / ".env.local"
diff --git a/halvorai_ingest/sources.py b/halvorai_ingest/sources.py
new file mode 100644
index 0000000..235cc24
--- /dev/null
+++ b/halvorai_ingest/sources.py
@@ -0,0 +1,68 @@
+"""Register a public Ed25519 key as a HalvorAI source (`POST /v1/sources`).
+
+Used by `init` for a server's own key and by `halvorai_ingest.device` for an app install's key.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+import httpx
+
+
+class SourceRegistrationError(RuntimeError):
+    """`POST /v1/sources` did not register the key. `status` is None when there was no answer."""
+
+    def __init__(self, message: str, status: int | None = None) -> None:
+        super().__init__(message)
+        self.status = status
+
+
+def sources_url(endpoint: str) -> str:
+    base = endpoint.rstrip("/")
+    if base.endswith("/v1/events"):
+        base = base[: -len("/v1/events")]
+    return f"{base}/v1/sources"
+
+
+def _request(api_key: str, name: str, public_jwk: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "json": {"name": name, "adapter": "custom", "public_jwk": dict(public_jwk)},
+        "headers": {"Authorization": f"Bearer {api_key}", "Accept": "application/json"},
+    }
+
+
+def _source_id(res: httpx.Response) -> str:
+    if res.status_code != 201:
+        raise SourceRegistrationError(
+            f"POST /v1/sources failed ({res.status_code}): {res.text or res.reason_phrase}",
+            res.status_code,
+        )
+    try:
+        body = res.json()
+    except ValueError:
+        body = None
+    source_id = body.get("id") if isinstance(body, dict) else None
+    if not isinstance(source_id, str) or not source_id.startswith("src_"):
+        raise SourceRegistrationError("POST /v1/sources returned no source id", res.status_code)
+    return source_id
+
+
+def post_source(
+    http: httpx.Client, endpoint: str, api_key: str, name: str, public_jwk: Mapping[str, Any]
+) -> str:
+    try:
+        res = http.post(sources_url(endpoint), **_request(api_key, name, public_jwk))
+    except httpx.HTTPError as exc:
+        raise SourceRegistrationError(f"POST /v1/sources failed: {exc}") from exc
+    return _source_id(res)
+
+
+async def post_source_async(
+    http: httpx.AsyncClient, endpoint: str, api_key: str, name: str, public_jwk: Mapping[str, Any]
+) -> str:
+    try:
+        res = await http.post(sources_url(endpoint), **_request(api_key, name, public_jwk))
+    except httpx.HTTPError as exc:
+        raise SourceRegistrationError(f"POST /v1/sources failed: {exc}") from exc
+    return _source_id(res)
diff --git a/pyproject.toml b/pyproject.toml
index 92f29cd..149594a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "halvorai-ingest"
-version = "0.4.1"
+version = "0.5.0"
 description = "Signed, attributed payment events for AI agents. Never in the payment path."
 readme = "README.md"
 requires-python = ">=3.10"
diff --git a/tests/test_device.py b/tests/test_device.py
new file mode 100644
index 0000000..ed3e3de
--- /dev/null
+++ b/tests/test_device.py
@@ -0,0 +1,205 @@
+"""Server-side helpers for app installs that sign their own events (halvorai_ingest_flutter)."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Callable
+
+import httpx
+import pytest
+
+from halvorai_ingest import generate_source_key
+from halvorai_ingest.device import (
+    DeviceRequestError,
+    forward_envelopes,
+    forward_envelopes_async,
+    register_device_source,
+    register_device_source_async,
+)
+
+API_KEY = "hk_live_" + "a" * 32
+ENDPOINT = "https://api.example"
+PUBLIC_JWK = {k: v for k, v in generate_source_key().items() if k not in ("d", "kid")}
+
+
+class FakeApi:
+    """A fake HalvorAI that records requests and answers with `respond`."""
+
+    def __init__(self, respond: Callable[[httpx.Request], httpx.Response]) -> None:
+        self.requests: list[httpx.Request] = []
+        self._respond = respond
+        self.transport = httpx.MockTransport(self._handle)
+
+    def _handle(self, request: httpx.Request) -> httpx.Response:
+        self.requests.append(request)
+        return self._respond(request)
+
+
+def accepted(request: httpx.Request) -> httpx.Response:
+    events = json.loads(request.content)["events"]
+    return httpx.Response(200, json={"results": [{"id": e["event"]["id"], "status": "accepted"} for e in events]})
+
+
+def envelope(**over: Any) -> dict[str, Any]:
+    event: dict[str, Any] = {
+        "id": "evt_01JA8Q4K3T9V2C7M5N6P8R0S1W",
+        "environment": "live",
+        "owner": {"id": "own_u_1", "type": "person"},
+    }
+    event.update(over)
+    return {"event": event, "sig": "eyJhbGciOiJFZERTQSJ9..c2ln"}
+
+
+# register_device_source
+
+
+def test_registers_only_the_public_half_and_returns_the_source_id() -> None:
+    api = FakeApi(lambda _: httpx.Response(201, json={"id": "src_device_1"}))
+    source_id = register_device_source(
+        {**PUBLIC_JWK, "kid": "src_pending", "alg": "EdDSA"},
+        name="iOS app",
+        api_key=API_KEY,
+        endpoint=ENDPOINT,
+        transport=api.transport,
+    )
+    assert source_id == "src_device_1"
+    request = api.requests[0]
+    assert str(request.url) == "https://api.example/v1/sources"
+    assert request.headers["authorization"] == f"Bearer {API_KEY}"
+    assert json.loads(request.content) == {
+        "name": "iOS app",
+        "adapter": "custom",
+        "public_jwk": {"kty": "OKP", "crv": "Ed25519", "x": PUBLIC_JWK["x"]},
+    }
+
+
+@pytest.mark.parametrize(
+    ("jwk", "message"),
+    [
+        ({**PUBLIC_JWK, "d": "secret"}, "private key"),
+        ({**PUBLIC_JWK, "crv": "P-256"}, "Ed25519"),
+        ({**PUBLIC_JWK, "x": "abc"}, "32 bytes"),
+        ("key", "JSON object"),
+    ],
+)
+def test_refuses_a_bad_key_with_400_and_never_calls_halvorai(jwk: object, message: str) -> None:
+    api = FakeApi(lambda _: httpx.Response(201, json={"id": "src_x"}))
+    with pytest.raises(DeviceRequestError) as info:
+        register_device_source(jwk, api_key=API_KEY, endpoint=ENDPOINT, transport=api.transport)
+    assert info.value.status == 400
+    assert message in info.value.message
+    assert api.requests == []
+
+
+def test_answers_502_when_halvorai_does_not_register_the_key() -> None:
+    api = FakeApi(lambda _: httpx.Response(503, text="down"))
+    with pytest.raises(DeviceRequestError) as info:
+        register_device_source(PUBLIC_JWK, api_key=API_KEY, endpoint=ENDPOINT, transport=api.transport)
+    assert info.value.status == 502
+    assert info.value.code == "upstream_error"
+    assert json.loads(info.value.body())["error"]["code"] == "upstream_error"
+
+
+def test_refuses_to_start_without_a_well_formed_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("HALVOR_API_KEY", raising=False)
+    with pytest.raises(ValueError, match="api key"):
+        register_device_source(PUBLIC_JWK, api_key="hk_nope")
+
+
+async def test_async_registration_works_the_same() -> None:
+    api = FakeApi(lambda _: httpx.Response(201, json={"id": "src_device_2"}))
+    source_id = await register_device_source_async(
+        PUBLIC_JWK, api_key=API_KEY, endpoint=ENDPOINT, transport=api.transport
+    )
+    assert source_id == "src_device_2"
+
+
+# forward_envelopes
+
+
+def test_passes_the_body_byte_for_byte_and_returns_the_answer_unchanged() -> None:
+    api = FakeApi(accepted)
+    # Unusual spacing and key order: re-serializing would change these bytes.
+    raw = '{ "events" : [ ' + json.dumps(envelope()) + ' ] , "extra": 1 }'
+    result = forward_envelopes(
+        raw, expected_owner_id="own_u_1", api_key=API_KEY, endpoint=ENDPOINT, transport=api.transport
+    )
+    request = api.requests[0]
+    assert str(request.url) == "https://api.example/v1/events"
+    assert request.headers["authorization"] == f"Bearer {API_KEY}"
+    assert request.content == raw.encode("utf-8")
+    assert result.status == 200
+    assert json.loads(result.body) == {"results": [{"id": "evt_01JA8Q4K3T9V2C7M5N6P8R0S1W", "status": "accepted"}]}
+
+
+def test_returns_halvorai_errors_unchanged() -> None:
+    api = FakeApi(lambda _: httpx.Response(503, content=b'{"error":{"code":"unavailable"}}'))
+    result = forward_envelopes(
+        json.dumps({"events": [envelope()]}),
+        expected_owner_id="own_u_1",
+        api_key=API_KEY,
+        endpoint=ENDPOINT,
+        transport=api.transport,
+    )
+    assert result.status == 503
+    assert result.body == b'{"error":{"code":"unavailable"}}'
+
+
+@pytest.mark.parametrize(
+    ("body", "status", "code"),
+    [
+        ({"events": [envelope(owner={"id": "own_someone_else", "type": "person"})]}, 422, "owner_mismatch"),
+        ({"events": [envelope(owner=None)]}, 422, "owner_mismatch"),
+        ({"events": [envelope(environment="test")]}, 422, "environment_mismatch"),
+        ({"events": []}, 400, "invalid_request"),
+        ({"events": [{"event": envelope()["event"]}]}, 400, "invalid_request"),
+        ({"events": [envelope() for _ in range(101)]}, 413, "payload_too_large"),
+    ],
+)
+def test_refuses_a_bad_batch_without_calling_halvorai(body: Any, status: int, code: str) -> None:
+    api = FakeApi(accepted)
+    result = forward_envelopes(
+        json.dumps(body), expected_owner_id="own_u_1", api_key=API_KEY, endpoint=ENDPOINT, transport=api.transport
+    )
+    assert result.status == status
+    assert json.loads(result.body)["error"]["code"] == code
+    assert api.requests == []
+
+
+def test_refuses_a_body_that_is_not_json_or_too_large() -> None:
+    api = FakeApi(accepted)
+    options: dict[str, Any] = {
+        "expected_owner_id": "own_u_1",
+        "api_key": API_KEY,
+        "endpoint": ENDPOINT,
+        "transport": api.transport,
+    }
+    assert forward_envelopes(b"not json", **options).status == 400
+    assert forward_envelopes(b"\xff\xfe", **options).status == 400
+    assert forward_envelopes(b"x" * 1_000_001, **options).status == 413
+    assert api.requests == []
+
+
+def test_answers_502_when_halvorai_cannot_be_reached() -> None:
+    def fail(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("refused", request=request)
+
+    result = forward_envelopes(
+        json.dumps({"events": [envelope()]}),
+        expected_owner_id="own_u_1",
+        api_key=API_KEY,
+        endpoint=ENDPOINT,
+        transport=httpx.MockTransport(fail),
+    )
+    assert result.status == 502
+    assert b"upstream_error" in result.body
+
+
+async def test_async_forwarding_works_the_same() -> None:
+    api = FakeApi(accepted)
+    raw = json.dumps({"events": [envelope()]}).encode("utf-8")
+    result = await forward_envelopes_async(
+        raw, expected_owner_id="own_u_1", api_key=API_KEY, endpoint=ENDPOINT, transport=api.transport
+    )
+    assert result.status == 200
+    assert api.requests[0].content == raw
diff --git a/uv.lock b/uv.lock
index 7927e88..91d074b 100644
--- a/uv.lock
+++ b/uv.lock
@@ -323,7 +323,7 @@ wheels = [
 
 [[package]]
 name = "halvorai-ingest"
-version = "0.4.1"
+version = "0.5.0"
 source = { editable = "." }
 dependencies = [
     { name = "cryptography" },