diff --git a/.env.example b/.env.example
index 46ba8fb..2dea4d9 100644
--- a/.env.example
+++ b/.env.example
@@ -62,8 +62,36 @@ PLC_DIRECTORY_URL=https://plc.directory
 # PLC_DIRECTORY_URL=http://127.0.0.1:2582
 
 # --- AppView ingest auth (optional, dev ok if unset) ---
+# Wenn gesetzt, muss die PDS denselben Wert als Header
+# `X-Ingest-Secret` mitschicken; ist er nicht gesetzt, nimmt
+# /internal/ingest-commit unauthentifizierte Writes entgegen (jeder,
+# der den Port erreicht, kann Posts/Follows/Notifications fälschen).
+# Die AppView warnt beim Start, solange er fehlt.
 # APPVIEW_INGEST_SECRET=change-me-to-a-shared-secret-between-pds-and-appview
 
+# --- AppView auth (Bearer-Token der PDS) ---
+# Erzwingt ein gültiges Access-JWT auf den privaten Endpoints
+# (/api/notifications, /api/notifications/count,
+# /api/notifications/seen, /api/timeline/home); `sub` im Token muss
+# dem `did`-Parameter entsprechen, sonst 403. Default: true.
+# Den öffentlichen P-256-Schlüssel holt sich die AppView von
+# `PDS_INTERNAL_URL` (sonst `PDS_PUBLIC_URL`) unter
+# /.well-known/did.json — `PDS_JWT_SECRET` verlässt die PDS nie.
+# Auf `false` verhält sich die AppView wie vor der Auth-Einführung
+# (alles öffentlich): nötig für die fail-open-Integrationstests mit
+# synthetischen DIDs und für eine Instanz, die schon per VPN
+# abgeschottet ist. Die AppView warnt beim Start laut, wenn er aus ist.
+# APPVIEW_AUTH_REQUIRED=true
+# Kommaseparierte Allowlist der Browser-Origins, die `/api/*` aufrufen
+# dürfen. Nicht gesetzt = bisheriges Verhalten
+# (`Access-Control-Allow-Origin: *`) plus Startup-Warnung.
+# Der Tauri-Webview hat je nach Plattform eine eigene Origin:
+# macOS/iOS `tauri://localhost`, Windows `http://tauri.localhost`,
+# im Dev-Modus der Vite-Server `http://127.0.0.1:1430`.
+# `Authorization` steht immer in den erlaubten Headern (sonst
+# scheitert der Preflight), `allow_credentials` bleibt aus.
+# APPVIEW_CORS_ORIGINS=tauri://localhost,http://127.0.0.1:1430
+
 # --- Tauri-Client (Build-/Laufzeit-Overrides des Desktop-Clients) ---
 # MAARCADETWEET_PDS_URL=http://127.0.0.1:2583
 # MAARCADETWEET_APPVIEW_URL=http://127.0.0.1:2584
diff --git a/Cargo.lock b/Cargo.lock
index 2d95a93..56bf80e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -53,6 +53,7 @@ dependencies = [
  "chrono",
  "dotenvy",
  "futures",
+ "p256",
  "reqwest",
  "rustls",
  "serde",
diff --git a/crates/appview/Cargo.toml b/crates/appview/Cargo.toml
index bff2966..42b1ae5 100644
--- a/crates/appview/Cargo.toml
+++ b/crates/appview/Cargo.toml
@@ -35,6 +35,7 @@ at-shared = { workspace = true }
 at-firehose = { workspace = true }
 at-crypto = { workspace = true }
 at-identity = { workspace = true }
+reqwest = { workspace = true }
 uuid = { workspace = true }
 rustls = { version = "0.23", default-features = false, features = ["ring", "std", "logging", "tls12"] }
 base64 = { workspace = true }
@@ -45,3 +46,7 @@ tokio = { workspace = true }
 reqwest = { workspace = true }
 serde_json = { workspace = true }
 uuid = { workspace = true }
+# Unit tests in `src/auth.rs` build a P-256 verification key in the
+# same `0x8012 + uncompressed point` encoding the PDS publishes, which
+# needs the curve's `ToEncodedPoint`.
+p256 = { workspace = true }
diff --git a/crates/appview/src/auth.rs b/crates/appview/src/auth.rs
new file mode 100644
index 0000000..c7b70ea
--- /dev/null
+++ b/crates/appview/src/auth.rs
@@ -0,0 +1,659 @@
+//! Bearer-token authentication for the AppView's private endpoints.
+//!
+//! ## What is being proven
+//!
+//! The PDS already issues an ES256 access JWT on
+//! `com.atproto.server.createSession` / `createAccount`. That token
+//! carries `sub = <did>`, `scope = "com.atproto.access"` and
+//! `aud = "did:web:appview.maarcadetweet.local"` — it was always meant
+//! to be presented *here*. All the AppView has to do is verify the
+//! signature and read `sub`.
+//!
+//! ## Where the key comes from
+//!
+//! Verifying an ES256 signature needs only the public half of the PDS's
+//! P-256 key. The PDS publishes it in its DID document at
+//! `GET /.well-known/did.json` (see `pds-server/src/main.rs`), so
+//! `PDS_JWT_SECRET` never leaves the PDS process. We fetch that
+//! document once, cache the `publicKeyMultibase` behind an `RwLock`,
+//! and re-fetch on a verification failure so an operator can rotate
+//! `PDS_JWT_SECRET` without restarting the AppView.
+//!
+//! Two failure modes are deliberately handled differently:
+//!
+//!  - **PDS unreachable at startup.** The AppView has always booted
+//!    independently of the PDS (it indexes the firehose, which has
+//!    nothing to do with the local PDS), and a crash-loop on a
+//!    colocated service that happens to boot second would be a
+//!    self-inflicted outage. So startup only *warns*; the first
+//!    authenticated request retries the fetch.
+//!  - **PDS unreachable when a token must be checked.** There is no
+//!    safe way to guess, so the request gets a `503`. Fail closed —
+//!    never fail open.
+//!
+//! ## Error contract
+//!
+//! The body shape matches every other AppView error
+//! (`{"error": …, "message": …}`). The `error` codes are load-bearing:
+//!
+//! | case                                   | status | `error`           |
+//! |----------------------------------------|--------|-------------------|
+//! | no / malformed `Authorization` header  | 401    | `AuthMissing`     |
+//! | bad signature, expired, wrong `scope`  | 401    | `TokenInvalid`    |
+//! | valid token, but `sub` ≠ requested did | 403    | `Forbidden`       |
+//! | PDS key not obtainable                 | 503    | `AuthUnavailable` |
+//!
+//! **`TokenInvalid` is a contract with the desktop client**: seeing it,
+//! the client refreshes its access JWT (`com.atproto.server.refreshSession`)
+//! and retries the request once. Renaming it silently logs every user
+//! out an hour after login.
+
+use at_crypto::jwt::JwtClaims;
+use axum::async_trait;
+use axum::extract::FromRequestParts;
+use axum::http::request::Parts;
+use axum::http::{HeaderMap, StatusCode};
+use axum::Json;
+use serde_json::{json, Value};
+use std::time::{Duration, Instant};
+use tokio::sync::RwLock;
+use tracing::{debug, warn};
+
+use crate::state::AppState;
+
+/// The scope an *access* token carries. Refresh tokens are minted with
+/// `com.atproto.refresh` by the same key, so without this check a
+/// refresh token — which lives for 90 days instead of an hour — would
+/// be accepted as a session credential everywhere.
+const ACCESS_SCOPE: &str = "com.atproto.access";
+
+/// Minimum spacing between two key re-fetches triggered by a failed
+/// verification.
+///
+/// Re-fetching on failure is what makes key rotation work without a
+/// restart. Doing it on *every* failure would also hand anyone who can
+/// reach the AppView a free amplifier: a stream of garbage tokens
+/// becomes a stream of requests to the PDS. One re-fetch per minute is
+/// far quicker than any plausible rotation cadence needs and costs the
+/// PDS nothing.
+const KEY_REFETCH_MIN_INTERVAL: Duration = Duration::from_secs(60);
+
+/// HTTP timeout for the DID-document fetch. The PDS is colocated; if it
+/// doesn't answer in two seconds it isn't answering, and a request
+/// blocked on auth is a request the user is staring at.
+const DID_DOC_TIMEOUT: Duration = Duration::from_secs(2);
+
+// -- error type --------------------------------------------------------------
+
+/// Everything that can go wrong between "a request arrived" and "we know
+/// which DID it belongs to". Converted into the AppView's standard error
+/// body by [`AuthError::into_response_parts`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AuthError {
+    /// No `Authorization` header, or one that isn't `Bearer <token>`.
+    Missing,
+    /// Signature, expiry or scope check failed. The client's cue to
+    /// refresh its access token and retry.
+    Invalid(String),
+    /// Authenticated fine, but the token belongs to somebody else.
+    Forbidden,
+    /// We could not obtain the PDS's public key, so we cannot decide.
+    Unavailable(String),
+}
+
+impl AuthError {
+    pub fn into_response_parts(self) -> (StatusCode, Json<Value>) {
+        let (status, code, message) = match self {
+            AuthError::Missing => (
+                StatusCode::UNAUTHORIZED,
+                "AuthMissing",
+                "missing Authorization: Bearer header".to_string(),
+            ),
+            AuthError::Invalid(detail) => {
+                (StatusCode::UNAUTHORIZED, "TokenInvalid", detail)
+            }
+            AuthError::Forbidden => (
+                StatusCode::FORBIDDEN,
+                "Forbidden",
+                "token sub does not match the requested did".to_string(),
+            ),
+            AuthError::Unavailable(detail) => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "AuthUnavailable",
+                format!("cannot verify tokens: {detail}"),
+            ),
+        };
+        (
+            status,
+            Json(json!({
+                "error": code,
+                "message": message,
+            })),
+        )
+    }
+}
+
+impl From<AuthError> for (StatusCode, Json<Value>) {
+    fn from(e: AuthError) -> Self {
+        e.into_response_parts()
+    }
+}
+
+// -- key cache ---------------------------------------------------------------
+
+#[derive(Default)]
+struct CachedKey {
+    multibase: Option<String>,
+    /// When we last *attempted* a fetch — successful or not. Throttles
+    /// the on-failure refresh path; see [`KEY_REFETCH_MIN_INTERVAL`].
+    last_attempt: Option<Instant>,
+}
+
+/// The PDS signing key, lazily fetched and cached.
+///
+/// Lives in [`AppState`] behind an `Arc`, so all handlers share one
+/// cache and one HTTP client.
+pub struct PdsKeys {
+    http: reqwest::Client,
+    /// Fully-qualified URL of the PDS's DID document.
+    did_doc_url: String,
+    inner: RwLock<CachedKey>,
+}
+
+impl PdsKeys {
+    /// Build a cache pointed at `base_url` (no trailing slash required).
+    pub fn new(base_url: &str) -> Self {
+        let http = reqwest::Client::builder()
+            .timeout(DID_DOC_TIMEOUT)
+            .build()
+            .expect("reqwest client build should never fail");
+        Self {
+            http,
+            did_doc_url: format!(
+                "{}/.well-known/did.json",
+                base_url.trim_end_matches('/')
+            ),
+            inner: RwLock::new(CachedKey::default()),
+        }
+    }
+
+    /// Same PDS the handle-sync worker talks to: `PDS_INTERNAL_URL`
+    /// when set, else `PDS_PUBLIC_URL`.
+    pub fn from_config(cfg: &at_shared::config::AppConfig) -> Self {
+        Self::new(&cfg.pds_base_url())
+    }
+
+    pub fn did_doc_url(&self) -> &str {
+        &self.did_doc_url
+    }
+
+    /// The cached key, if we have ever fetched one.
+    pub async fn cached(&self) -> Option<String> {
+        self.inner.read().await.multibase.clone()
+    }
+
+    /// Fetch the DID document and replace the cached key.
+    ///
+    /// Called once at startup (best effort), on the first authenticated
+    /// request if startup failed, and — throttled — after a failed
+    /// verification.
+    pub async fn refresh(&self) -> anyhow::Result<String> {
+        // Record the attempt before the await so two concurrent
+        // failures can't both decide they're the first one.
+        self.inner.write().await.last_attempt = Some(Instant::now());
+
+        let resp = self.http.get(&self.did_doc_url).send().await?;
+        if !resp.status().is_success() {
+            anyhow::bail!(
+                "{} returned {}",
+                self.did_doc_url,
+                resp.status().as_u16()
+            );
+        }
+        let doc: Value = resp.json().await?;
+        let key = extract_public_key_multibase(&doc)?;
+        {
+            let mut guard = self.inner.write().await;
+            guard.multibase = Some(key.clone());
+        }
+        debug!(url = %self.did_doc_url, "loaded PDS signing key");
+        Ok(key)
+    }
+
+    /// The key to verify with, fetching it if we don't have one yet.
+    async fn key_or_fetch(&self) -> Result<String, AuthError> {
+        if let Some(k) = self.cached().await {
+            return Ok(k);
+        }
+        self.refresh().await.map_err(|e| {
+            warn!(url = %self.did_doc_url, error = %e, "PDS signing key unavailable");
+            AuthError::Unavailable(format!(
+                "PDS did document at {} not reachable: {e}",
+                self.did_doc_url
+            ))
+        })
+    }
+
+    /// Re-fetch after a verification failure, at most once per
+    /// [`KEY_REFETCH_MIN_INTERVAL`]. Returns the new key only when it
+    /// actually differs from `stale` — re-running the same failing
+    /// verification against an unchanged key proves nothing.
+    async fn refetch_if_stale(&self, stale: &str) -> Option<String> {
+        {
+            let guard = self.inner.read().await;
+            if let Some(last) = guard.last_attempt {
+                if last.elapsed() < KEY_REFETCH_MIN_INTERVAL {
+                    return None;
+                }
+            }
+        }
+        match self.refresh().await {
+            Ok(fresh) if fresh != stale => {
+                warn!("PDS signing key changed; re-verifying with the rotated key");
+                Some(fresh)
+            }
+            Ok(_) => None,
+            Err(e) => {
+                warn!(error = %e, "PDS key re-fetch after verification failure failed");
+                None
+            }
+        }
+    }
+
+    /// Verify an access token and return its claims.
+    ///
+    /// Retries exactly once against a freshly fetched key, so a rotated
+    /// `PDS_JWT_SECRET` costs one extra HTTP round trip rather than a
+    /// restart.
+    pub async fn verify_access_token(&self, token: &str) -> Result<JwtClaims, AuthError> {
+        let key = self.key_or_fetch().await?;
+        match verify_with_key(token, &key) {
+            Ok(claims) => Ok(claims),
+            Err(first) => {
+                let Some(fresh) = self.refetch_if_stale(&key).await else {
+                    return Err(first);
+                };
+                verify_with_key(token, &fresh).map_err(|_| first)
+            }
+        }
+    }
+}
+
+/// Pull `verificationMethod[].publicKeyMultibase` out of a DID document.
+///
+/// We take the first entry that carries a `publicKeyMultibase` rather
+/// than insisting on a fragment name: the PDS writes `#atproto`, but a
+/// document served by a proxy or a future PDS version may order or name
+/// its methods differently, and any key in the document is a key the
+/// controller published for itself. A document with none is an error,
+/// not an empty key — silently caching `""` would turn every later
+/// verification into a confusing signature failure.
+fn extract_public_key_multibase(doc: &Value) -> anyhow::Result<String> {
+    let methods = doc
+        .get("verificationMethod")
+        .and_then(|v| v.as_array())
+        .ok_or_else(|| anyhow::anyhow!("did document has no verificationMethod array"))?;
+    for m in methods {
+        if let Some(k) = m.get("publicKeyMultibase").and_then(|v| v.as_str()) {
+            if !k.is_empty() {
+                return Ok(k.to_string());
+            }
+        }
+    }
+    anyhow::bail!("did document has no verificationMethod with publicKeyMultibase")
+}
+
+/// Signature + expiry + scope check against one specific key.
+///
+/// `verify_jwt` handles the ES256 signature and `exp` (with a 30 s
+/// leeway for clock skew); the scope check is ours, and it is the line
+/// that keeps a 90-day refresh token from working as a session
+/// credential.
+fn verify_with_key(token: &str, pubkey_multibase: &str) -> Result<JwtClaims, AuthError> {
+    let claims = at_crypto::jwt::verify_jwt(token, pubkey_multibase)
+        .map_err(|e| AuthError::Invalid(format!("invalid token: {e}")))?;
+    match claims.scope.as_deref() {
+        Some(ACCESS_SCOPE) => Ok(claims),
+        other => Err(AuthError::Invalid(format!(
+            "token scope {:?} is not {ACCESS_SCOPE}",
+            other.unwrap_or("<none>")
+        ))),
+    }
+}
+
+/// Extract the bearer token from an `Authorization` header.
+///
+/// The scheme match is case-insensitive (RFC 7235 says it is) — some
+/// HTTP clients send `bearer`.
+fn bearer_token(headers: &HeaderMap) -> Result<String, AuthError> {
+    let raw = headers
+        .get("authorization")
+        .and_then(|v| v.to_str().ok())
+        .ok_or(AuthError::Missing)?;
+    let (scheme, token) = raw.split_once(' ').ok_or(AuthError::Missing)?;
+    if !scheme.eq_ignore_ascii_case("bearer") {
+        return Err(AuthError::Missing);
+    }
+    let token = token.trim();
+    if token.is_empty() {
+        return Err(AuthError::Missing);
+    }
+    Ok(token.to_string())
+}
+
+// -- extractor ---------------------------------------------------------------
+
+/// The authenticated DID of the caller.
+///
+/// `None` means auth is switched off for this instance
+/// (`APPVIEW_AUTH_REQUIRED=false`) — *not* "anonymous but allowed
+/// through". When auth is on, this is always `Some` by construction:
+/// the extractor rejects the request otherwise.
+///
+/// Handlers must still call [`AuthedDid::ensure_matches`] with the DID
+/// the request asks about. Proving *who you are* is not the same as
+/// proving *whose notifications you may read*.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AuthedDid(pub Option<String>);
+
+impl AuthedDid {
+    /// 403 unless the token's `sub` is the DID the request targets.
+    ///
+    /// With auth disabled this is a no-op, which is exactly what
+    /// `APPVIEW_AUTH_REQUIRED=false` means.
+    pub fn ensure_matches(&self, did: &str) -> Result<(), (StatusCode, Json<Value>)> {
+        match self.0.as_deref() {
+            None => Ok(()),
+            Some(sub) if sub == did => Ok(()),
+            Some(_) => Err(AuthError::Forbidden.into_response_parts()),
+        }
+    }
+
+    /// The caller's DID, if authenticated.
+    ///
+    /// No handler needs it yet — every private endpoint takes the DID
+    /// as a parameter and compares it — but an endpoint that acts
+    /// purely on "whoever is calling" would read it here instead of
+    /// trusting a query parameter.
+    #[allow(dead_code)]
+    pub fn did(&self) -> Option<&str> {
+        self.0.as_deref()
+    }
+}
+
+#[async_trait]
+impl FromRequestParts<AppState> for AuthedDid {
+    type Rejection = (StatusCode, Json<Value>);
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        state: &AppState,
+    ) -> Result<Self, Self::Rejection> {
+        if !state.cfg.appview_auth_required {
+            // Opt-out instance. The startup log says so in capitals.
+            return Ok(AuthedDid(None));
+        }
+        let token = bearer_token(&parts.headers).map_err(AuthError::into_response_parts)?;
+        let claims = state
+            .pds_keys
+            .verify_access_token(&token)
+            .await
+            .map_err(AuthError::into_response_parts)?;
+        Ok(AuthedDid(Some(claims.sub)))
+    }
+}
+
+// -- startup posture ---------------------------------------------------------
+
+/// Log, once at boot, every way this instance is configured to be less
+/// strict than the defaults.
+///
+/// All three of these are legitimate configurations — a VPN-only
+/// deployment, a dev box, a single-machine setup — and all three are
+/// also exactly what an accidentally-public instance looks like. The
+/// only defence that survives a hurried deployment is a log line the
+/// operator cannot miss, so each one is a `warn!` naming the variable
+/// that turns it back on.
+pub fn log_startup_posture(cfg: &at_shared::config::AppConfig) {
+    if !cfg.appview_auth_required {
+        warn!(
+            "APPVIEW_AUTH_REQUIRED=false — /api/notifications* and \
+             /api/timeline/home are served to ANY caller for ANY did. \
+             Only safe when this instance is unreachable from untrusted \
+             networks."
+        );
+    }
+    if cfg.appview_cors_origins.is_empty() {
+        warn!(
+            "APPVIEW_CORS_ORIGINS unset — sending Access-Control-Allow-Origin: * \
+             so any web page can call this AppView from a browser. Set it to the \
+             origins your client actually uses, e.g. \
+             'tauri://localhost,http://127.0.0.1:1430'."
+        );
+    }
+    if cfg.appview_ingest_secret.is_none() {
+        warn!(
+            "APPVIEW_INGEST_SECRET unset — POST /internal/ingest-commit accepts \
+             unauthenticated writes into the index (anyone who can reach this port \
+             can forge posts, follows and notifications). Set the same value here \
+             and on the PDS."
+        );
+    }
+}
+
+// -- tests -------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use at_crypto::ecdsa::P256Keypair;
+    use at_crypto::jwt::{issue_jwt, JwtClaims};
+    use axum::http::HeaderValue;
+
+    /// A throwaway server key plus its `publicKeyMultibase`, in the
+    /// exact encoding the PDS publishes: `0x8012` (P-256) followed by
+    /// the *uncompressed* affine coordinates. `P256Keypair::generate`
+    /// stores a compressed point, which `verify_jwt` cannot decode — so
+    /// this mirrors `pds-server`'s `server_p256_keypair`.
+    fn test_key() -> (P256Keypair, String) {
+        let kp = P256Keypair::generate().unwrap();
+        let vk = kp.verifying_key().unwrap();
+        let pt = vk.to_encoded_point(false);
+        let mut raw = vec![0x80u8, 0x12u8];
+        raw.extend_from_slice(pt.x().unwrap());
+        raw.extend_from_slice(pt.y().unwrap());
+        let multibase = at_crypto::multibase_util::encode_b58btc(&raw);
+        (kp, multibase)
+    }
+
+    fn mint(kp: &P256Keypair, did: &str, scope: &str, ttl_secs: i64) -> String {
+        let now = chrono::Utc::now().timestamp();
+        issue_jwt(
+            kp,
+            &JwtClaims {
+                iss: "did:web:127.0.0.1%3A2583".into(),
+                sub: did.into(),
+                aud: "did:web:appview.maarcadetweet.local".into(),
+                iat: now - 1,
+                exp: now + ttl_secs,
+                jti: None,
+                scope: Some(scope.into()),
+            },
+        )
+        .unwrap()
+    }
+
+    fn header_map(value: &str) -> HeaderMap {
+        let mut h = HeaderMap::new();
+        h.insert("authorization", HeaderValue::from_str(value).unwrap());
+        h
+    }
+
+    #[test]
+    fn missing_header_is_auth_missing() {
+        assert_eq!(bearer_token(&HeaderMap::new()), Err(AuthError::Missing));
+    }
+
+    #[test]
+    fn malformed_header_is_auth_missing() {
+        // No scheme, wrong scheme, and an empty token all read as
+        // "the client never presented a credential" — 401 AuthMissing,
+        // not TokenInvalid, so the client re-authenticates instead of
+        // burning a refresh round trip.
+        assert_eq!(bearer_token(&header_map("abc.def.ghi")), Err(AuthError::Missing));
+        assert_eq!(bearer_token(&header_map("Basic dXNlcjpwdw==")), Err(AuthError::Missing));
+        assert_eq!(bearer_token(&header_map("Bearer   ")), Err(AuthError::Missing));
+    }
+
+    #[test]
+    fn bearer_scheme_is_case_insensitive() {
+        assert_eq!(bearer_token(&header_map("bearer tok")).unwrap(), "tok");
+        assert_eq!(bearer_token(&header_map("Bearer tok")).unwrap(), "tok");
+    }
+
+    #[test]
+    fn error_bodies_carry_the_documented_codes() {
+        // These strings are a contract: the desktop client keys its
+        // "refresh and retry" behaviour off `TokenInvalid`.
+        let (s, b) = AuthError::Missing.into_response_parts();
+        assert_eq!(s, StatusCode::UNAUTHORIZED);
+        assert_eq!(b.0["error"], "AuthMissing");
+        let (s, b) = AuthError::Invalid("x".into()).into_response_parts();
+        assert_eq!(s, StatusCode::UNAUTHORIZED);
+        assert_eq!(b.0["error"], "TokenInvalid");
+        let (s, b) = AuthError::Forbidden.into_response_parts();
+        assert_eq!(s, StatusCode::FORBIDDEN);
+        assert_eq!(b.0["error"], "Forbidden");
+        let (s, b) = AuthError::Unavailable("pds down".into()).into_response_parts();
+        assert_eq!(s, StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(b.0["error"], "AuthUnavailable");
+        // Every body carries both fields the client parses.
+        assert!(b.0["message"].as_str().unwrap().contains("pds down"));
+    }
+
+    #[test]
+    fn valid_access_token_verifies() {
+        let (kp, mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", ACCESS_SCOPE, 3600);
+        let claims = verify_with_key(&token, &mb).unwrap();
+        assert_eq!(claims.sub, "did:plc:alice");
+    }
+
+    #[test]
+    fn garbage_token_is_invalid() {
+        let (_, mb) = test_key();
+        let err = verify_with_key("not-a-jwt", &mb).unwrap_err();
+        assert!(matches!(err, AuthError::Invalid(_)));
+    }
+
+    #[test]
+    fn token_signed_by_another_key_is_invalid() {
+        let (kp, _) = test_key();
+        let (_, other_mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", ACCESS_SCOPE, 3600);
+        assert!(matches!(
+            verify_with_key(&token, &other_mb).unwrap_err(),
+            AuthError::Invalid(_)
+        ));
+    }
+
+    #[test]
+    fn refresh_scope_is_rejected() {
+        // The refresh token is signed by the same key and lives 90
+        // days. Without the scope check it would be a session token.
+        let (kp, mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", "com.atproto.refresh", 3600);
+        let err = verify_with_key(&token, &mb).unwrap_err();
+        match err {
+            AuthError::Invalid(msg) => assert!(msg.contains("com.atproto.refresh")),
+            other => panic!("expected Invalid, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn expired_token_is_rejected() {
+        // Beyond `verify_jwt`'s 30 s leeway.
+        let (kp, mb) = test_key();
+        let token = mint(&kp, "did:plc:alice", ACCESS_SCOPE, -120);
+        assert!(matches!(
+            verify_with_key(&token, &mb).unwrap_err(),
+            AuthError::Invalid(_)
+        ));
+    }
+
+    #[test]
+    fn ensure_matches_enforces_sub_equals_did() {
+        let me = AuthedDid(Some("did:plc:alice".into()));
+        assert!(me.ensure_matches("did:plc:alice").is_ok());
+        let (status, body) = me.ensure_matches("did:plc:bob").unwrap_err();
+        assert_eq!(status, StatusCode::FORBIDDEN);
+        assert_eq!(body.0["error"], "Forbidden");
+        // A prefix of the real DID must not pass.
+        assert!(me.ensure_matches("did:plc:ali").is_err());
+    }
+
+    #[test]
+    fn ensure_matches_is_a_noop_when_auth_disabled() {
+        let off = AuthedDid(None);
+        assert!(off.ensure_matches("did:plc:anyone").is_ok());
+        assert_eq!(off.did(), None);
+    }
+
+    #[test]
+    fn did_doc_key_extraction() {
+        let doc = json!({
+            "id": "did:web:127.0.0.1%3A2583",
+            "verificationMethod": [{
+                "id": "did:web:127.0.0.1%3A2583#atproto",
+                "type": "Multikey",
+                "controller": "did:web:127.0.0.1%3A2583",
+                "publicKeyMultibase": "zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme"
+            }]
+        });
+        assert_eq!(
+            extract_public_key_multibase(&doc).unwrap(),
+            "zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme"
+        );
+
+        // A document without a usable key must error rather than cache
+        // an empty string.
+        assert!(extract_public_key_multibase(&json!({})).is_err());
+        assert!(extract_public_key_multibase(&json!({
+            "verificationMethod": [{"id": "#x", "type": "Multikey"}]
+        }))
+        .is_err());
+        assert!(extract_public_key_multibase(&json!({
+            "verificationMethod": [{"publicKeyMultibase": ""}]
+        }))
+        .is_err());
+    }
+
+    #[test]
+    fn did_doc_url_is_built_from_the_base_url() {
+        assert_eq!(
+            PdsKeys::new("http://127.0.0.1:2583").did_doc_url(),
+            "http://127.0.0.1:2583/.well-known/did.json"
+        );
+        // A trailing slash must not produce a double slash — some
+        // servers 404 on it.
+        assert_eq!(
+            PdsKeys::new("http://pds:3000/").did_doc_url(),
+            "http://pds:3000/.well-known/did.json"
+        );
+    }
+
+    #[tokio::test]
+    async fn verification_fails_closed_when_the_pds_is_unreachable() {
+        // Port 1 on loopback: nothing listens there, so the fetch fails
+        // fast. The result must be a 503, never a pass-through.
+        let keys = PdsKeys::new("http://127.0.0.1:1");
+        let err = keys.verify_access_token("whatever").await.unwrap_err();
+        assert!(
+            matches!(err, AuthError::Unavailable(_)),
+            "expected Unavailable, got {err:?}"
+        );
+        let (status, body) = err.into_response_parts();
+        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(body.0["error"], "AuthUnavailable");
+    }
+}
diff --git a/crates/appview/src/ingest.rs b/crates/appview/src/ingest.rs
index 612781d..465f394 100644
--- a/crates/appview/src/ingest.rs
+++ b/crates/appview/src/ingest.rs
@@ -15,8 +15,31 @@
 //! }
 //! ```
 //!
-//! In production this endpoint would be protected with mTLS and a token
-//! minted by the PDS; for now it's open inside the cluster.
+//! ## Who may call this
+//!
+//! This is the AppView's only write path, and it is not a browser
+//! endpoint: it is excluded from the CORS layer in
+//! [`crate::routes::router`], because an `Access-Control-Allow-Origin`
+//! header here would only ever help a web page forge index entries.
+//!
+//! Authentication is the shared secret `APPVIEW_INGEST_SECRET`,
+//! compared in constant time against the caller's `X-Ingest-Secret`
+//! header:
+//!
+//! - **set** → enforced. A missing or wrong header is `401
+//!   AuthenticationRequired`.
+//! - **unset** → anonymous writes are accepted, and the AppView shouts
+//!   about it once at startup (see
+//!   [`crate::auth::log_startup_posture`]). Refusing to start would
+//!   break every existing single-machine dev setup for a service that,
+//!   in that configuration, is bound to loopback anyway; accepting
+//!   silently is how an internet-facing deployment ends up letting
+//!   anyone forge posts, follows and notifications. So: keep working,
+//!   but never quietly.
+//!
+//! A future hardening step is mTLS or a PDS-minted token, at which
+//! point the shared secret becomes the fallback rather than the only
+//! line.
 
 use crate::indexer;
 use crate::state::AppState;
@@ -51,8 +74,13 @@ pub struct IngestCommitReq {
 }
 
 /// Authenticate internal ingest requests.
-/// - If `APPVIEW_INGEST_SECRET` env var is unset: dev mode, accept anything.
-/// - If set: require `X-Ingest-Secret: <value>` header to match.
+///
+/// - `APPVIEW_INGEST_SECRET` unset: accept anonymous writes (dev mode —
+///   the startup log warns, see the module docs for why this isn't a
+///   hard failure).
+/// - Set: require a matching `X-Ingest-Secret` header. The comparison
+///   is constant-time so a caller can't recover the secret byte by byte
+///   from response timings.
 pub fn check_ingest_secret(
     headers: &HeaderMap,
     configured: Option<&str>,
diff --git a/crates/appview/src/lib.rs b/crates/appview/src/lib.rs
index 6d50c89..9440bc6 100644
--- a/crates/appview/src/lib.rs
+++ b/crates/appview/src/lib.rs
@@ -3,6 +3,7 @@
 //! tests under `tests/` import from here so they can build a worker
 //! against a stub resolver without booting the binary.
 
+pub mod auth;
 pub mod firehose;
 pub mod handle_sync;
 pub mod indexer;
diff --git a/crates/appview/src/main.rs b/crates/appview/src/main.rs
index 885c80d..1087f24 100644
--- a/crates/appview/src/main.rs
+++ b/crates/appview/src/main.rs
@@ -7,6 +7,7 @@ use tokio::sync::mpsc;
 use tracing::info;
 use tracing_subscriber::EnvFilter;
 
+mod auth;
 mod firehose;
 mod handle_sync;
 mod indexer;
@@ -83,6 +84,33 @@ async fn main() -> Result<()> {
 
     let state = AppState::new(cfg.clone(), db.clone(), stats.clone());
 
+    // Announce every relaxed security switch before we serve anything.
+    auth::log_startup_posture(&cfg);
+
+    // Pre-load the PDS's published signing key so the first
+    // authenticated request doesn't pay for the round trip.
+    //
+    // Best effort on purpose: the AppView has always started
+    // independently of the PDS, and in a compose file the two race. A
+    // hard failure here would turn "the PDS booted two seconds later"
+    // into "the AppView is in a crash loop". If the fetch fails, the
+    // first authenticated request retries it — and answers `503
+    // AuthUnavailable` if the PDS is still unreachable. Never open.
+    if cfg.appview_auth_required {
+        match state.pds_keys.refresh().await {
+            Ok(_) => info!(
+                url = %state.pds_keys.did_doc_url(),
+                "loaded PDS signing key for token verification"
+            ),
+            Err(e) => tracing::warn!(
+                url = %state.pds_keys.did_doc_url(),
+                error = %e,
+                "could not load the PDS signing key at startup; will retry on the \
+                 first authenticated request (which fails with 503 until it works)"
+            ),
+        }
+    }
+
     // Back-fill the `handle` column on posts that the Jetstream
     // indexer inserted with an empty placeholder. The worker dispatches
     // by DID method: `did:plc:` → PLC directory, `did:web:` → a
@@ -102,12 +130,12 @@ async fn main() -> Result<()> {
     // inside docker compose) — `pds_public_url` may not be reachable
     // from inside the cluster when TLS / DNS is set up for outside
     // clients only.
-    let pds_base_url = cfg
-        .pds_internal_url
-        .clone()
-        .unwrap_or_else(|| cfg.pds_public_url.clone());
+    //
+    // `AppConfig::pds_base_url()` owns that fallback so the handle
+    // resolver and the signing-key fetch in `auth.rs` can never end up
+    // pointed at different PDS instances.
     let pds_resolver: Arc<dyn at_identity::DidHandleResolver> = Arc::new(
-        at_identity::pds_handle::PdsHandleResolver::new(pds_base_url),
+        at_identity::pds_handle::PdsHandleResolver::new(cfg.pds_base_url()),
     );
 
     let handle_sync = handle_sync::HandleSyncWorker {
diff --git a/crates/appview/src/routes.rs b/crates/appview/src/routes.rs
index f8da41a..9e497f9 100644
--- a/crates/appview/src/routes.rs
+++ b/crates/appview/src/routes.rs
@@ -8,6 +8,30 @@
 //! - `ingest_commit`: the internal-only writer used by the PDS, owned
 //!   in `crate::ingest`.
 //!
+//! ## Public vs. private
+//!
+//! Most of what the AppView serves is public by construction: in AT
+//! Proto a post, a profile, a follow edge and a like are records in a
+//! public repo, replicated over the firehose. Those endpoints
+//! (`/api/profile*`, `/api/search`, `/api/post/*`, `/api/thread*`,
+//! `/api/followers`, `/api/following`) need no credential.
+//!
+//! Two things are *not* public, and they are the reason this service
+//! has authentication at all:
+//!
+//!  - `/api/notifications*` — who interacted with you, and the read
+//!    state of that list. Nothing in the protocol makes it readable by
+//!    anyone but the recipient.
+//!  - `/api/timeline/home` — the timeline is assembled from the
+//!    viewer's follow graph, so serving it to an arbitrary `did`
+//!    parameter answers "what does this person's feed look like" for
+//!    any DID a caller cares to type.
+//!
+//! Both require a valid PDS-issued access token whose `sub` equals the
+//! `did` in the request ([`crate::auth`]). The check is two steps on
+//! purpose: the extractor proves *who* the caller is, and
+//! `ensure_matches` proves they are asking about themselves.
+//!
 //! The cursor format used by `timeline_home` is opaque: it's a
 //! `base64url(micros):uri` pair, which is what [`cursor::encode`] and
 //! [`cursor::decode`] produce/consume.
@@ -19,11 +43,14 @@ use axum::{
     routing::{get, post},
     Json, Router,
 };
+use axum::http::{header, HeaderValue, Method};
 use chrono::{DateTime, TimeZone, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
-use tower_http::cors::{Any, CorsLayer};
+use tower_http::cors::{AllowOrigin, Any, CorsLayer};
+use tracing::warn;
 
+use crate::auth::AuthedDid;
 use crate::state::AppState;
 
 pub mod cursor;
@@ -35,25 +62,73 @@ use types::{
     ProfileResponse, SearchResponse, ThreadFullResponse, TimelineResponse,
 };
 
-pub fn router(state: AppState) -> Router {
-    // CORS: the Tauri webview's origin is the Vite dev server
-    // (`http://127.0.0.1:1430`) in dev or the bundled `tauri://` /
-    // `asset://` origin in production. Either way it's a cross-origin
-    // fetch against this service's `http://127.0.0.1:2584` listen
-    // address, so the browser blocks the response without an explicit
-    // allow-origin header. We allow any origin — the AppView's
-    // public read endpoints (`/api/...`) carry no auth cookie and
-    // the AppView runs alongside the user's own PDS, not on the
-    // open internet; production deployments behind a reverse proxy
-    // can tighten this via the proxy itself.
-    let cors = CorsLayer::new()
-        .allow_origin(Any)
-        .allow_methods(Any)
-        .allow_headers(Any);
+/// Build the CORS layer for the browser-facing routes.
+///
+/// The Tauri webview is a cross-origin caller: in dev its origin is the
+/// Vite server (`http://127.0.0.1:1430`), in a packaged build it is a
+/// platform-specific scheme — `tauri://localhost` on macOS/iOS,
+/// `http://tauri.localhost` on Windows. Either way the browser drops
+/// the response unless we send `Access-Control-Allow-Origin`.
+///
+/// `APPVIEW_CORS_ORIGINS` is a comma-separated allowlist, e.g.
+/// `tauri://localhost,http://127.0.0.1:1430`. When it is unset we keep
+/// the historic wildcard so no existing deployment breaks on upgrade —
+/// [`crate::auth::log_startup_posture`] warns about that at startup.
+///
+/// `Authorization` has to be in `allow_headers`: it is not a
+/// CORS-safelisted header, so without it the browser's preflight fails
+/// and the authenticated endpoints become unreachable from the webview
+/// — with an error that looks nothing like an auth problem.
+///
+/// `allow_credentials` stays off. We authenticate with a bearer token
+/// the client attaches deliberately, never with an ambient cookie, so
+/// there is nothing for a hostile page to replay — and turning it on
+/// would additionally make the wildcard origin illegal.
+fn cors_layer(cfg: &at_shared::config::AppConfig) -> CorsLayer {
+    let base = CorsLayer::new()
+        .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
+        .allow_headers([
+            header::AUTHORIZATION,
+            header::CONTENT_TYPE,
+            header::ACCEPT,
+        ]);
 
-    Router::new()
+    if cfg.appview_cors_origins.is_empty() {
+        return base.allow_origin(Any);
+    }
+
+    // Anything unparseable as a header value is dropped with a warning
+    // rather than panicking the process — a stray quote in the env var
+    // must not take the service down. If *every* entry is bad we fall
+    // back to the wildcard and say so, because an empty allowlist would
+    // silently break every browser client.
+    let mut origins: Vec<HeaderValue> = Vec::new();
+    for raw in &cfg.appview_cors_origins {
+        match HeaderValue::from_str(raw) {
+            Ok(v) => origins.push(v),
+            Err(_) => warn!(origin = %raw, "APPVIEW_CORS_ORIGINS: ignoring unparseable origin"),
+        }
+    }
+    if origins.is_empty() {
+        warn!("APPVIEW_CORS_ORIGINS contained no usable origin; falling back to allow-any");
+        return base.allow_origin(Any);
+    }
+    base.allow_origin(AllowOrigin::list(origins))
+}
+
+pub fn router(state: AppState) -> Router {
+    let cors = cors_layer(&state.cfg);
+
+    // Browser-facing surface. Everything here answers a `fetch()` from
+    // the desktop client, so it carries the CORS layer.
+    let api = Router::new()
         .route("/", get(root))
+        // --- private: token required, `sub` must equal `did` ---
         .route("/api/timeline/home", get(timeline_home))
+        .route("/api/notifications", get(notifications))
+        .route("/api/notifications/count", get(notifications_count))
+        .route("/api/notifications/seen", post(notifications_seen))
+        // --- public: AT Proto public records ---
         .route("/api/profile", get(profile_query))
         .route("/api/profile/:handle", get(profile_path))
         .route("/api/search", get(search))
@@ -66,15 +141,20 @@ pub fn router(state: AppState) -> Router {
         // implementation, so they can't drift.
         .route("/api/thread", get(thread_query))
         .route("/api/thread/*uri", get(thread_path))
-        .route("/api/notifications", get(notifications))
-        .route("/api/notifications/count", get(notifications_count))
-        .route("/api/notifications/seen", post(notifications_seen))
         .route("/api/followers", get(followers))
         .route("/api/following", get(following))
         .route("/healthz", get(healthz))
-        .route("/internal/ingest-commit", post(crate::ingest::ingest_commit))
-        .layer(cors)
-        .with_state(state)
+        .layer(cors);
+
+    // Server-to-server surface. `/internal/ingest-commit` is called by
+    // the PDS with a shared secret, never by a browser, so it stays
+    // outside the CORS layer: handing it an
+    // `Access-Control-Allow-Origin` header would only ever help a web
+    // page try to write to the index.
+    let internal = Router::new()
+        .route("/internal/ingest-commit", post(crate::ingest::ingest_commit));
+
+    api.merge(internal).with_state(state)
 }
 
 async fn root() -> Json<Value> {
@@ -115,13 +195,21 @@ const MAX_LIMIT: i64 = 100;
 /// surface.
 const MAX_FOLLOWED_DIDS: usize = 1000;
 
+/// `GET /api/timeline/home?did=…`
+///
+/// **Authenticated.** The timeline is derived from the viewer's follow
+/// graph, so `did` must be the caller's own DID — otherwise this
+/// endpoint would answer "what does this account's feed look like" for
+/// any DID at all.
 async fn timeline_home(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Query(q): Query<TimelineQuery>,
 ) -> Result<Json<TimelineResponse>, (StatusCode, Json<Value>)> {
     if q.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&q.did)?;
     let limit = clamp_limit(q.limit);
 
     // Look up the set of DIDs this user follows, then build the
@@ -1046,13 +1134,19 @@ struct NotificationsQuery {
 /// of the list. The tiebreak here is the row's `id` rather than a URI
 /// (a notification has no URI of its own), which the shared
 /// [`cursor`] codec carries in its string slot.
+///
+/// **Authenticated**: `did` is the recipient, so the caller has to be
+/// that recipient. This is the endpoint that made authentication
+/// necessary in the first place.
 async fn notifications(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Query(q): Query<NotificationsQuery>,
 ) -> Result<Json<NotificationsResponse>, (StatusCode, Json<Value>)> {
     if q.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&q.did)?;
     let limit = clamp_limit(q.limit);
     let keyset = decode_cursor(q.cursor.as_deref())?;
     // The cursor's string slot holds the row id. A client that hands
@@ -1121,13 +1215,18 @@ struct NotificationCountQuery {
 /// scales with the number of *unread* rows, not the user's lifetime
 /// notification history. That matters because the client polls this
 /// for its tray badge.
+///
+/// **Authenticated**, same rule as the list itself: an unread count is
+/// still information about someone else's inbox.
 async fn notifications_count(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Query(q): Query<NotificationCountQuery>,
 ) -> Result<Json<NotificationCountResponse>, (StatusCode, Json<Value>)> {
     if q.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&q.did)?;
     let count: i64 = sqlx::query_scalar(
         "SELECT COUNT(*)::BIGINT FROM notifications \
           WHERE recipient_did = $1 AND read_at IS NULL",
@@ -1165,13 +1264,20 @@ struct NotificationsSeenReq {
 /// updates nothing and reports `updated: 0`. `read_at` is set to
 /// `now()` (when we recorded the ack), not to `seenAt` (which is a
 /// client-supplied watermark and could be arbitrarily far in the past).
+///
+/// **Authenticated**, and the only *write* among the private
+/// endpoints: without the check anyone could clear another user's
+/// unread badge. `AuthedDid` runs before `Json` because the body
+/// extractor consumes the request — axum requires body extractors last.
 async fn notifications_seen(
     State(state): State<AppState>,
+    auth: AuthedDid,
     Json(req): Json<NotificationsSeenReq>,
 ) -> Result<Json<NotificationSeenResponse>, (StatusCode, Json<Value>)> {
     if req.did.is_empty() {
         return Err(bad_request("did is required"));
     }
+    auth.ensure_matches(&req.did)?;
     let res = sqlx::query(
         r#"UPDATE notifications
               SET read_at = now()
diff --git a/crates/appview/src/state.rs b/crates/appview/src/state.rs
index 6b179cd..18bbc38 100644
--- a/crates/appview/src/state.rs
+++ b/crates/appview/src/state.rs
@@ -2,18 +2,30 @@ use at_shared::config::AppConfig;
 use sqlx::PgPool;
 use std::sync::Arc;
 
+use crate::auth::PdsKeys;
 use crate::firehose::Stats;
 
 #[derive(Clone)]
 pub struct AppState {
-    #[allow(dead_code)]
     pub cfg: AppConfig,
     pub db: PgPool,
     pub stats: Arc<Stats>,
+    /// Cache of the PDS's published signing key, used by the
+    /// [`crate::auth::AuthedDid`] extractor. Shared (`Arc`) so every
+    /// handler verifies against the same cached key and one HTTP
+    /// client, and so a key rotation picked up by one request is
+    /// immediately visible to the rest.
+    pub pds_keys: Arc<PdsKeys>,
 }
 
 impl AppState {
     pub fn new(cfg: AppConfig, db: PgPool, stats: Arc<Stats>) -> Self {
-        Self { cfg, db, stats }
+        let pds_keys = Arc::new(PdsKeys::from_config(&cfg));
+        Self {
+            cfg,
+            db,
+            stats,
+            pds_keys,
+        }
     }
 }
diff --git a/crates/appview/tests/api_integration.rs b/crates/appview/tests/api_integration.rs
index 99bb9d5..e2d0ee6 100644
--- a/crates/appview/tests/api_integration.rs
+++ b/crates/appview/tests/api_integration.rs
@@ -7,6 +7,9 @@
 //! rather than panicking — so `cargo test --workspace` stays green in
 //! environments where the appview hasn't been started.
 
+mod common;
+
+use common::TestAuth;
 use serde_json::{json, Value};
 use std::time::Duration;
 
@@ -46,6 +49,36 @@ async fn db_reachable() -> bool {
     )
 }
 
+/// How this suite authenticates against `/api/timeline/home`, which is
+/// no longer public. `None` means the AppView enforces auth but the
+/// test process cannot mint a token (no `PDS_JWT_SECRET`), in which
+/// case the test skips like it does for a missing service.
+async fn auth_or_skip() -> Option<TestAuth> {
+    TestAuth::probe(&client().await, APPVIEW_URL).await
+}
+
+/// `GET /api/timeline/home` as `did`, with the bearer token attached
+/// when the instance requires one. The seeded DIDs are synthetic, so
+/// the token is minted from the PDS's own signing secret — see
+/// `tests/common/mod.rs`.
+async fn get_timeline(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    did: &str,
+    extra: &[(&str, &str)],
+) -> reqwest::Response {
+    let mut params: Vec<(&str, &str)> = vec![("did", did)];
+    params.extend_from_slice(extra);
+    auth.apply(
+        c.get(format!("{APPVIEW_URL}/api/timeline/home"))
+            .query(&params),
+        did,
+    )
+    .send()
+    .await
+    .unwrap()
+}
+
 async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response {
     c.post(format!("{APPVIEW_URL}/internal/ingest-commit"))
         .json(&body)
@@ -120,6 +153,7 @@ async fn timeline_returns_seeded_posts() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("tl");
 
     // Seed 3 posts with distinct rkeys.
@@ -157,12 +191,7 @@ async fn timeline_returns_seeded_posts() {
     // machine that has run this suite twice) the three rows we just
     // seeded fall outside a 10-row window and the assertions below
     // fail for reasons that have nothing to do with the timeline.
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "100")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "100")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");
@@ -222,6 +251,7 @@ async fn timeline_paginates_with_cursor() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("pg");
 
     // Seed 50 posts.
@@ -246,28 +276,14 @@ async fn timeline_paginates_with_cursor() {
     tokio::time::sleep(Duration::from_millis(100)).await;
 
     // Page 1: limit=20.
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "20")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "20")]).await;
     let body: Value = resp.json().await.unwrap();
     let page1 = body["posts"].as_array().unwrap().clone();
     let cursor1 = body["cursor"].as_str().expect("page1 cursor");
     assert_eq!(page1.len(), 20, "page1 should be exactly 20");
 
     // Page 2: with cursor.
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[
-            ("did", did.as_str()),
-            ("limit", "20"),
-            ("cursor", cursor1),
-        ])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "20"), ("cursor", cursor1)]).await;
     let body: Value = resp.json().await.unwrap();
     let page2 = body["posts"].as_array().unwrap().clone();
     assert_eq!(page2.len(), 20, "page2 should be exactly 20");
@@ -285,16 +301,7 @@ async fn timeline_paginates_with_cursor() {
 
     // Page 3: tail — fewer than 20 expected, cursor=null.
     let cursor2 = body["cursor"].as_str().expect("page2 cursor");
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[
-            ("did", did.as_str()),
-            ("limit", "20"),
-            ("cursor", cursor2),
-        ])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "20"), ("cursor", cursor2)]).await;
     let body: Value = resp.json().await.unwrap();
     let page3 = body["posts"].as_array().unwrap().clone();
     assert!(page3.len() <= 20, "page3 should be <= 20");
@@ -484,6 +491,7 @@ async fn timeline_filters_to_followees() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let url = std::env::var("DATABASE_URL_APPVIEW").unwrap();
     let pool = sqlx::PgPool::connect(&url).await.unwrap();
 
@@ -503,12 +511,7 @@ async fn timeline_filters_to_followees() {
 
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", alice.as_str()), ("limit", "100")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &alice, &[("limit", "100")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");
@@ -555,18 +558,14 @@ async fn timeline_includes_own_posts() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let alice = did_for_test("alone");
 
     // Alice posts without seeding any follows.
     seed_posts(&c, &alice, &["alice's first post", "alice's second post"]).await;
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", alice.as_str()), ("limit", "100")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &alice, &[("limit", "100")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");
@@ -605,6 +604,7 @@ async fn timeline_caps_followee_list() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let url = std::env::var("DATABASE_URL_APPVIEW").unwrap();
     let pool = sqlx::PgPool::connect(&url).await.unwrap();
 
@@ -624,12 +624,7 @@ async fn timeline_caps_followee_list() {
     seed_posts(&c, &alice, &["poweruser post"]).await;
     tokio::time::sleep(Duration::from_millis(100)).await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", alice.as_str()), ("limit", "50")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &alice, &[("limit", "50")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().expect("posts is array");
diff --git a/crates/appview/tests/auth_integration.rs b/crates/appview/tests/auth_integration.rs
new file mode 100644
index 0000000..ceb7854
--- /dev/null
+++ b/crates/appview/tests/auth_integration.rs
@@ -0,0 +1,311 @@
+//! End-to-end tests for AppView authentication.
+//!
+//! Unlike the other suites in this directory these use a **real**
+//! token: the test creates an account on the running PDS
+//! (`com.atproto.server.createAccount`, the same call the desktop
+//! client makes at signup) and presents the access JWT it gets back to
+//! the AppView. That is the whole point — it exercises the actual key
+//! distribution path (PDS signs → publishes its public key at
+//! `/.well-known/did.json` → AppView fetches and verifies), which a
+//! self-minted token would bypass.
+//!
+//! Fail-open like the rest of the suites: if the PDS or the AppView
+//! isn't running, or the AppView runs with `APPVIEW_AUTH_REQUIRED=false`,
+//! the test prints a notice and returns successfully.
+
+use serde_json::{json, Value};
+use std::time::Duration;
+
+const PDS_URL: &str = "http://127.0.0.1:2583";
+
+fn appview_url() -> String {
+    std::env::var("APPVIEW_TEST_URL").unwrap_or_else(|_| "http://127.0.0.1:2584".to_string())
+}
+
+/// 30 s, not the 5 s the sibling suites use.
+///
+/// The authenticated happy path for `/api/timeline/home` runs the
+/// cold-start query (a brand-new account follows nobody, so the handler
+/// falls back to the global recent feed). On a developer machine whose
+/// AppView has been indexing the public firehose for a while that scan
+/// takes seconds — a timeout there would look like an auth failure and
+/// is nothing of the sort.
+fn client() -> reqwest::Client {
+    reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .build()
+        .unwrap()
+}
+
+async fn up(c: &reqwest::Client, base: &str) -> bool {
+    for _ in 0..20 {
+        if let Ok(r) = c.get(format!("{base}/healthz")).send().await {
+            if r.status().is_success() {
+                return true;
+            }
+        }
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+    false
+}
+
+/// A freshly created PDS account.
+struct Account {
+    did: String,
+    access_jwt: String,
+    refresh_jwt: String,
+}
+
+/// Guard for every test here: both services up **and** the AppView
+/// actually enforcing auth. Returns the client plus a new account.
+async fn ready() -> Option<(reqwest::Client, Account)> {
+    let c = client();
+    if !up(&c, &appview_url()).await {
+        eprintln!("appview not running, skipping");
+        return None;
+    }
+    if !up(&c, PDS_URL).await {
+        eprintln!("pds not running, skipping");
+        return None;
+    }
+    // Probe: does this instance enforce auth? An operator running with
+    // `APPVIEW_AUTH_REQUIRED=false` (the mode the other suites use)
+    // would otherwise see every assertion here fail for the one reason
+    // that isn't a bug.
+    let probe = c
+        .get(format!("{}/api/notifications/count", appview_url()))
+        .query(&[("did", "did:plc:auth_probe")])
+        .send()
+        .await
+        .ok()?;
+    if probe.status().as_u16() != 401 {
+        eprintln!(
+            "appview does not enforce auth (probe returned {}), skipping",
+            probe.status()
+        );
+        return None;
+    }
+    let acc = create_account(&c).await?;
+    Some((c, acc))
+}
+
+async fn create_account(c: &reqwest::Client) -> Option<Account> {
+    let handle = format!("auth_{}.maarcadetweet.local", uuid::Uuid::new_v4().simple());
+    let r: Value = c
+        .post(format!("{PDS_URL}/xrpc/com.atproto.server.createAccount"))
+        .json(&json!({ "handle": handle, "password": "hunter2hunter2" }))
+        .send()
+        .await
+        .ok()?
+        .json()
+        .await
+        .ok()?;
+    Some(Account {
+        did: r["did"].as_str()?.to_string(),
+        access_jwt: r["access_jwt"].as_str()?.to_string(),
+        refresh_jwt: r["refresh_jwt"].as_str()?.to_string(),
+    })
+}
+
+/// `GET /api/notifications?did=…` with an optional bearer token.
+async fn get_notifications(
+    c: &reqwest::Client,
+    did: &str,
+    token: Option<&str>,
+) -> reqwest::Response {
+    let mut rb = c
+        .get(format!("{}/api/notifications", appview_url()))
+        .query(&[("did", did)]);
+    if let Some(t) = token {
+        rb = rb.bearer_auth(t);
+    }
+    rb.send().await.unwrap()
+}
+
+/// Assert the AppView's standard error envelope: the status, and the
+/// `error` code the desktop client branches on.
+async fn assert_error(r: reqwest::Response, status: u16, code: &str) {
+    let got = r.status().as_u16();
+    let body: Value = r.json().await.unwrap();
+    assert_eq!(got, status, "unexpected status; body = {body}");
+    assert_eq!(body["error"], json!(code), "unexpected error code: {body}");
+    assert!(
+        body["message"].is_string(),
+        "error body must carry a message: {body}"
+    );
+}
+
+#[tokio::test]
+async fn own_token_reads_own_notifications() {
+    let Some((c, acc)) = ready().await else { return };
+    let r = get_notifications(&c, &acc.did, Some(&acc.access_jwt)).await;
+    assert_eq!(r.status().as_u16(), 200);
+    let body: Value = r.json().await.unwrap();
+    // A brand-new account has no notifications, but the shape must be
+    // the normal list response, not an error.
+    assert!(body["notifications"].is_array(), "body = {body}");
+}
+
+#[tokio::test]
+async fn no_token_is_401_auth_missing() {
+    let Some((c, acc)) = ready().await else { return };
+    let r = get_notifications(&c, &acc.did, None).await;
+    assert_error(r, 401, "AuthMissing").await;
+}
+
+#[tokio::test]
+async fn foreign_did_is_403() {
+    let Some((c, acc)) = ready().await else { return };
+    // A valid token, but asking about somebody else's inbox. This is
+    // the case the endpoint used to answer with a 200.
+    let r = get_notifications(&c, "did:plc:somebodyelse", Some(&acc.access_jwt)).await;
+    assert_error(r, 403, "Forbidden").await;
+}
+
+#[tokio::test]
+async fn garbage_token_is_401_token_invalid() {
+    let Some((c, acc)) = ready().await else { return };
+    let r = get_notifications(&c, &acc.did, Some("not.a.jwt")).await;
+    assert_error(r, 401, "TokenInvalid").await;
+}
+
+#[tokio::test]
+async fn refresh_token_is_not_accepted() {
+    let Some((c, acc)) = ready().await else { return };
+    // Signed by the same key and valid for 90 days — only the `scope`
+    // claim separates it from a session credential.
+    let r = get_notifications(&c, &acc.did, Some(&acc.refresh_jwt)).await;
+    assert_error(r, 401, "TokenInvalid").await;
+}
+
+#[tokio::test]
+async fn seen_write_requires_matching_token() {
+    let Some((c, acc)) = ready().await else { return };
+    let url = format!("{}/api/notifications/seen", appview_url());
+
+    // Somebody else's unread badge: 403, nothing written.
+    let r = c
+        .post(&url)
+        .bearer_auth(&acc.access_jwt)
+        .json(&json!({ "did": "did:plc:somebodyelse" }))
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 403, "Forbidden").await;
+
+    // No credential at all: 401.
+    let r = c
+        .post(&url)
+        .json(&json!({ "did": &acc.did }))
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 401, "AuthMissing").await;
+
+    // Own inbox: allowed (zero rows updated — the account is new).
+    let r = c
+        .post(&url)
+        .bearer_auth(&acc.access_jwt)
+        .json(&json!({ "did": &acc.did }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status().as_u16(), 200);
+    let body: Value = r.json().await.unwrap();
+    assert_eq!(body["ok"], json!(true));
+}
+
+#[tokio::test]
+async fn timeline_home_is_authenticated() {
+    let Some((c, acc)) = ready().await else { return };
+    let base = appview_url();
+
+    let r = c
+        .get(format!("{base}/api/timeline/home"))
+        .query(&[("did", acc.did.as_str())])
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 401, "AuthMissing").await;
+
+    let r = c
+        .get(format!("{base}/api/timeline/home"))
+        .query(&[("did", "did:plc:somebodyelse")])
+        .bearer_auth(&acc.access_jwt)
+        .send()
+        .await
+        .unwrap();
+    assert_error(r, 403, "Forbidden").await;
+
+    let r = c
+        .get(format!("{base}/api/timeline/home"))
+        .query(&[("did", acc.did.as_str())])
+        .bearer_auth(&acc.access_jwt)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(r.status().as_u16(), 200);
+}
+
+#[tokio::test]
+async fn public_endpoints_stay_public() {
+    let Some((c, acc)) = ready().await else { return };
+    let base = appview_url();
+
+    // Profiles, search, follower lists and the health probe are public
+    // records in AT Proto. Putting them behind auth would be a
+    // behaviour change no protocol rule asks for — assert they still
+    // answer without a token.
+    for (path, params) in [
+        ("/api/profile", vec![("did", acc.did.as_str())]),
+        ("/api/search", vec![("q", "hello")]),
+        ("/api/followers", vec![("did", acc.did.as_str())]),
+        ("/api/following", vec![("did", acc.did.as_str())]),
+        ("/healthz", vec![]),
+    ] {
+        let r = c
+            .get(format!("{base}{path}"))
+            .query(&params)
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(
+            r.status().as_u16(),
+            200,
+            "{path} must remain public, got {}",
+            r.status()
+        );
+    }
+}
+
+/// The AppView can only verify anything because the PDS publishes its
+/// public key. If this document ever stops carrying a usable key, every
+/// authenticated request degrades to `503 AuthUnavailable` — so assert
+/// the shape the AppView parses.
+#[tokio::test]
+async fn pds_publishes_a_usable_signing_key() {
+    let c = client();
+    if !up(&c, PDS_URL).await {
+        eprintln!("pds not running, skipping");
+        return;
+    }
+    let doc: Value = c
+        .get(format!("{PDS_URL}/.well-known/did.json"))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let id = doc["id"].as_str().expect("did document needs an id");
+    assert!(id.starts_with("did:web:"), "id = {id}");
+    let vm = &doc["verificationMethod"][0];
+    assert_eq!(vm["type"], json!("Multikey"));
+    assert_eq!(vm["controller"], json!(id));
+    let key = vm["publicKeyMultibase"]
+        .as_str()
+        .expect("verificationMethod needs publicKeyMultibase");
+    // base58-btc multibase: the `z` prefix is what the AppView's
+    // decoder expects.
+    assert!(key.starts_with('z'), "key = {key}");
+}
diff --git a/crates/appview/tests/common/mod.rs b/crates/appview/tests/common/mod.rs
new file mode 100644
index 0000000..1d71881
--- /dev/null
+++ b/crates/appview/tests/common/mod.rs
@@ -0,0 +1,167 @@
+//! Shared test support for the AppView integration suites.
+//!
+//! ## Why the suites need this
+//!
+//! `/api/notifications*` and `/api/timeline/home` require a
+//! PDS-issued access token whose `sub` equals the `did` in the request.
+//! The suites, however, seed synthetic DIDs (`did:plc:ntf_…`) through
+//! `/internal/ingest-commit` — accounts the PDS has never heard of, so
+//! there is no `createSession` that would hand out a token for them.
+//!
+//! The way out is that a token is just an ES256 JWT signed with the
+//! server key derived from `PDS_JWT_SECRET`. A test that can read that
+//! secret (from the process environment, or from the repo `.env` the
+//! dev stack itself was started with) can mint a token for any DID it
+//! likes — the same thing `pds-server/src/jwt_issuer.rs` does.
+//!
+//! ## Fail-open, like the rest of the suites
+//!
+//! [`TestAuth::probe`] asks the running AppView whether it enforces
+//! auth at all:
+//!
+//! - not enforcing (`APPVIEW_AUTH_REQUIRED=false`) → no header needed;
+//! - enforcing and we have the secret → mint per-DID tokens;
+//! - enforcing and we don't → `None`, and the caller skips, exactly as
+//!   it already skips when the service or the database is down.
+
+#![allow(dead_code)]
+
+use at_crypto::ecdsa::P256Keypair;
+use at_crypto::jwt::{issue_jwt, JwtClaims};
+
+/// Audience the PDS stamps into access tokens. Not validated by
+/// `verify_jwt` today (`validate_aud = false`), but minting a token
+/// that differs from the real thing would make this helper a poor
+/// stand-in for the client.
+const APPVIEW_AUD: &str = "did:web:appview.maarcadetweet.local";
+
+/// The scope the AppView insists on. A token with any other scope —
+/// `com.atproto.refresh`, say — is rejected with `TokenInvalid`.
+pub const ACCESS_SCOPE: &str = "com.atproto.access";
+
+/// How the suite should authenticate against the AppView under test.
+///
+/// `Clone` because a test that pages through results in a closure has
+/// to hand each iteration its own copy, exactly like the client and the
+/// base URL next to it.
+#[derive(Clone)]
+pub enum TestAuth {
+    /// `APPVIEW_AUTH_REQUIRED=false`: send no `Authorization` header.
+    Disabled,
+    /// Auth is enforced; mint tokens with this hex secret.
+    Secret(String),
+}
+
+impl TestAuth {
+    /// Decide how (or whether) this suite can talk to the AppView.
+    ///
+    /// Returns `None` when the AppView enforces auth but no
+    /// `PDS_JWT_SECRET` is reachable — the caller should print a notice
+    /// and return, keeping `cargo test --workspace` green on a machine
+    /// without the dev stack's environment.
+    pub async fn probe(c: &reqwest::Client, base_url: &str) -> Option<Self> {
+        // An unauthenticated probe against a private endpoint. We only
+        // look at the status: 401 means the extractor is active. A DID
+        // that doesn't exist is fine — the auth check runs first.
+        let status = c
+            .get(format!("{base_url}/api/notifications/count"))
+            .query(&[("did", "did:plc:auth_probe")])
+            .send()
+            .await
+            .ok()?
+            .status()
+            .as_u16();
+        if status != 401 && status != 503 {
+            return Some(TestAuth::Disabled);
+        }
+        match pds_jwt_secret() {
+            Some(secret) => Some(TestAuth::Secret(secret)),
+            None => {
+                eprintln!(
+                    "appview enforces auth (probe returned {status}) but PDS_JWT_SECRET \
+                     is not set and no .env was found — skipping"
+                );
+                None
+            }
+        }
+    }
+
+    /// Attach an `Authorization: Bearer` header for `did`, if needed.
+    pub fn apply(&self, rb: reqwest::RequestBuilder, did: &str) -> reqwest::RequestBuilder {
+        match self {
+            TestAuth::Disabled => rb,
+            TestAuth::Secret(secret) => match mint_access_jwt(secret, did, ACCESS_SCOPE, 3600) {
+                Some(token) => rb.bearer_auth(token),
+                None => rb,
+            },
+        }
+    }
+
+    /// A token for `did` — for tests that want to send a *wrong* one on
+    /// purpose. `None` when auth is disabled, in which case the test
+    /// that needs it should skip.
+    pub fn token_for(&self, did: &str) -> Option<String> {
+        match self {
+            TestAuth::Disabled => None,
+            TestAuth::Secret(secret) => mint_access_jwt(secret, did, ACCESS_SCOPE, 3600),
+        }
+    }
+}
+
+/// `PDS_JWT_SECRET` from the environment, falling back to the repo
+/// `.env` — the same file the running dev stack loaded at startup, so
+/// the minted tokens verify against the key the PDS actually publishes.
+pub fn pds_jwt_secret() -> Option<String> {
+    if let Ok(v) = std::env::var("PDS_JWT_SECRET") {
+        if !v.trim().is_empty() {
+            return Some(v);
+        }
+    }
+    // `dotenvy::dotenv` walks up from the current directory, which for
+    // a test binary is the crate root — so this finds the workspace
+    // `.env` two levels up. It never overrides a real env var.
+    let _ = dotenvy::dotenv();
+    std::env::var("PDS_JWT_SECRET")
+        .ok()
+        .filter(|v| !v.trim().is_empty())
+}
+
+/// Mint an access JWT for `did`, signed with the PDS's server key.
+///
+/// Mirrors `pds-server/src/jwt_issuer.rs`: the P-256 secret scalar is
+/// the **first 32 bytes** of `PDS_JWT_SECRET` (the config allows a
+/// longer secret), i.e. the first 64 hex characters.
+///
+/// `ttl_secs` may be negative to build a deliberately expired token.
+pub fn mint_access_jwt(
+    secret_hex: &str,
+    did: &str,
+    scope: &str,
+    ttl_secs: i64,
+) -> Option<String> {
+    let hex = secret_hex.trim().trim_start_matches("0x");
+    if hex.len() < 64 || !hex.chars().all(|c| c.is_ascii_hexdigit()) {
+        eprintln!("PDS_JWT_SECRET is not ≥32 bytes of hex; cannot mint a test token");
+        return None;
+    }
+    let kp = P256Keypair {
+        secret_hex: hex[..64].to_string(),
+        // Only the signing half is used by `issue_jwt`; the verifier
+        // fetches the public key from the PDS's DID document.
+        public_multibase: String::new(),
+    };
+    let now = chrono::Utc::now().timestamp();
+    issue_jwt(
+        &kp,
+        &JwtClaims {
+            iss: "did:web:test".into(),
+            sub: did.to_string(),
+            aud: APPVIEW_AUD.into(),
+            iat: now - 1,
+            exp: now + ttl_secs,
+            jti: None,
+            scope: Some(scope.to_string()),
+        },
+    )
+    .ok()
+}
diff --git a/crates/appview/tests/embeds_integration.rs b/crates/appview/tests/embeds_integration.rs
index eaccec1..d21dc0a 100644
--- a/crates/appview/tests/embeds_integration.rs
+++ b/crates/appview/tests/embeds_integration.rs
@@ -15,6 +15,9 @@
 //! and returns rather than panicking. The point of the tests is to
 //! catch regressions in CI where the service IS up.
 
+mod common;
+
+use common::TestAuth;
 use serde_json::{json, Value};
 use std::time::Duration;
 
@@ -50,6 +53,32 @@ async fn db_reachable() -> bool {
     )
 }
 
+/// `/api/timeline/home` requires a token whose `sub` is the requested
+/// DID. The DIDs here are synthetic, so the token is minted from the
+/// PDS signing secret — see `tests/common/mod.rs`. `None` → skip.
+async fn auth_or_skip() -> Option<TestAuth> {
+    TestAuth::probe(&client().await, APPVIEW_URL).await
+}
+
+/// `GET /api/timeline/home` as `did`, authenticated when required.
+async fn get_timeline(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    did: &str,
+    extra: &[(&str, &str)],
+) -> reqwest::Response {
+    let mut params: Vec<(&str, &str)> = vec![("did", did)];
+    params.extend_from_slice(extra);
+    auth.apply(
+        c.get(format!("{APPVIEW_URL}/api/timeline/home"))
+            .query(&params),
+        did,
+    )
+    .send()
+    .await
+    .unwrap()
+}
+
 async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response {
     c.post(format!("{APPVIEW_URL}/internal/ingest-commit"))
         .json(&body)
@@ -125,6 +154,7 @@ async fn timeline_includes_embed() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("img");
     let uri = seed_post(
         &c,
@@ -160,12 +190,7 @@ async fn timeline_includes_embed() {
     )
     .await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "10")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().unwrap();
@@ -199,6 +224,7 @@ async fn timeline_includes_external_embed() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("ext");
     let uri = seed_post(
         &c,
@@ -222,12 +248,7 @@ async fn timeline_includes_external_embed() {
     )
     .await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "10")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let posts = body["posts"].as_array().unwrap();
@@ -241,12 +262,7 @@ async fn timeline_includes_external_embed() {
             break;
         }
         tokio::time::sleep(Duration::from_millis(50)).await;
-        let resp = c
-            .get(format!("{APPVIEW_URL}/api/timeline/home"))
-            .query(&[("did", did.as_str()), ("limit", "10")])
-            .send()
-            .await
-            .unwrap();
+        let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
         let body: Value = resp.json().await.unwrap();
         our = body["posts"]
             .as_array()
@@ -410,6 +426,7 @@ async fn timeline_post_without_embed_has_null_embed() {
         return;
     }
     let c = client().await;
+    let Some(auth) = auth_or_skip().await else { return };
     let did = did_for_test("plain");
     let uri = seed_post(
         &c,
@@ -421,12 +438,7 @@ async fn timeline_post_without_embed_has_null_embed() {
     )
     .await;
 
-    let resp = c
-        .get(format!("{APPVIEW_URL}/api/timeline/home"))
-        .query(&[("did", did.as_str()), ("limit", "10")])
-        .send()
-        .await
-        .unwrap();
+    let resp = get_timeline(&c, &auth, &did, &[("limit", "10")]).await;
     assert_eq!(resp.status().as_u16(), 200);
     let body: Value = resp.json().await.unwrap();
     let our = body["posts"]
diff --git a/crates/appview/tests/notifications_integration.rs b/crates/appview/tests/notifications_integration.rs
index 666ecbb..c8c94c9 100644
--- a/crates/appview/tests/notifications_integration.rs
+++ b/crates/appview/tests/notifications_integration.rs
@@ -14,6 +14,9 @@
 //! API. That's deliberate: it's the only way to catch a mismatch
 //! between what the write path stores and what the read path joins.
 
+mod common;
+
+use common::TestAuth;
 use serde_json::{json, Value};
 use std::time::Duration;
 
@@ -55,8 +58,12 @@ async fn db_pool() -> Option<sqlx::PgPool> {
 }
 
 /// Guard used at the top of every test. Returns `None` (→ skip) unless
-/// both the HTTP service and the database are up.
-async fn ready() -> Option<(reqwest::Client, sqlx::PgPool)> {
+/// the HTTP service is up, the database is reachable, **and** we know
+/// how to authenticate: `/api/notifications*` now requires a token
+/// whose `sub` is the requested DID, and the DIDs seeded here are
+/// synthetic, so the token has to be minted from the PDS's signing
+/// secret. See `tests/common/mod.rs`.
+async fn ready() -> Option<(reqwest::Client, sqlx::PgPool, TestAuth)> {
     if !wait_for_appview_db().await {
         eprintln!("appview not running, skipping");
         return None;
@@ -65,7 +72,32 @@ async fn ready() -> Option<(reqwest::Client, sqlx::PgPool)> {
         eprintln!("appview DB unreachable, skipping");
         return None;
     };
-    Some((client().await, pool))
+    let c = client().await;
+    let auth = TestAuth::probe(&c, &appview_url()).await?;
+    Some((c, pool, auth))
+}
+
+/// `GET <url>` against a private endpoint, carrying the token for
+/// `did` when the instance enforces auth.
+fn authed_get(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    url: String,
+    did: &str,
+) -> reqwest::RequestBuilder {
+    auth.apply(c.get(url), did)
+}
+
+/// `POST <url>` against a private endpoint. Same rule as
+/// [`authed_get`] — `/api/notifications/seen` is a write into one
+/// user's read state.
+fn authed_post(
+    c: &reqwest::Client,
+    auth: &TestAuth,
+    url: String,
+    did: &str,
+) -> reqwest::RequestBuilder {
+    auth.apply(c.post(url), did)
 }
 
 async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response {
@@ -181,7 +213,7 @@ async fn seed_follow(c: &reqwest::Client, follower: &str, subject: &str) {
 #[tokio::test]
 async fn notifications_list_count_and_seen() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("alice");
@@ -196,8 +228,7 @@ async fn notifications_list_count_and_seen() {
     let reply_uri = seed_reply(&c, &carol, &post_uri, &post_uri, "carol's reply").await;
     seed_follow(&c, &bob, &alice).await;
 
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str()), ("limit", "50")])
         .send()
         .await
@@ -253,8 +284,7 @@ async fn notifications_list_count_and_seen() {
     }
 
     // The unread count agrees with the list.
-    let resp = c
-        .get(format!("{base}/api/notifications/count"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications/count"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -264,8 +294,7 @@ async fn notifications_list_count_and_seen() {
     assert_eq!(body["count"], json!(3));
 
     // Mark everything seen.
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice }))
         .send()
         .await
@@ -276,8 +305,7 @@ async fn notifications_list_count_and_seen() {
     assert_eq!(body["updated"], json!(3));
 
     // Idempotent: a second call updates nothing and still succeeds.
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice }))
         .send()
         .await
@@ -286,8 +314,7 @@ async fn notifications_list_count_and_seen() {
     assert_eq!(body["updated"], json!(0));
 
     // Count is now zero and the rows carry a read_at.
-    let resp = c
-        .get(format!("{base}/api/notifications/count"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications/count"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -295,8 +322,7 @@ async fn notifications_list_count_and_seen() {
     let body: Value = resp.json().await.unwrap();
     assert_eq!(body["count"], json!(0));
 
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -311,8 +337,7 @@ async fn notifications_list_count_and_seen() {
         "/api/notifications",
         "/api/notifications/count",
     ] {
-        let resp = c
-            .get(format!("{base}{path}"))
+        let resp = authed_get(&c, &auth, format!("{base}{path}"), &alice)
             .query(&[("did", "")])
             .send()
             .await
@@ -326,7 +351,7 @@ async fn notifications_list_count_and_seen() {
 #[tokio::test]
 async fn notifications_skip_self_interactions() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("solo");
@@ -335,8 +360,7 @@ async fn notifications_skip_self_interactions() {
     seed_reply(&c, &alice, &post_uri, &post_uri, "and replying too").await;
     seed_follow(&c, &alice, &alice).await;
 
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -355,7 +379,7 @@ async fn notifications_skip_self_interactions() {
 #[tokio::test]
 async fn notifications_paginate_with_cursor() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("popular");
@@ -372,9 +396,9 @@ async fn notifications_paginate_with_cursor() {
         let c = c.clone();
         let alice = alice.clone();
         let base = base.clone();
+        let auth = auth.clone();
         async move {
-            let mut req = c
-                .get(format!("{base}/api/notifications"))
+            let mut req = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
                 .query(&[("did", alice.as_str()), ("limit", "5")]);
             if let Some(cur) = cursor {
                 req = req.query(&[("cursor", cur.as_str())]);
@@ -417,8 +441,7 @@ async fn notifications_paginate_with_cursor() {
     assert_eq!(all.len(), 12);
 
     // A mangled cursor is a 400, not a silent restart at page 1.
-    let resp = c
-        .get(format!("{base}/api/notifications"))
+    let resp = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str()), ("cursor", "!!!garbage!!!")])
         .send()
         .await
@@ -431,7 +454,7 @@ async fn notifications_paginate_with_cursor() {
 #[tokio::test]
 async fn notifications_seen_respects_watermark() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, auth)) = ready().await else {
         return;
     };
     let alice = did_for_test("watermark");
@@ -442,8 +465,7 @@ async fn notifications_seen_respects_watermark() {
 
     // Read back the first notification's indexed_at — that's the
     // watermark a client would echo after rendering page 1.
-    let body: Value = c
-        .get(format!("{base}/api/notifications"))
+    let body: Value = authed_get(&c, &auth, format!("{base}/api/notifications"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -461,8 +483,7 @@ async fn notifications_seen_respects_watermark() {
     let second = did_for_test("late");
     seed_like(&c, &second, &post_uri).await;
 
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice, "seenAt": watermark }))
         .send()
         .await
@@ -475,8 +496,7 @@ async fn notifications_seen_respects_watermark() {
     );
 
     // The later one is still unread.
-    let body: Value = c
-        .get(format!("{base}/api/notifications/count"))
+    let body: Value = authed_get(&c, &auth, format!("{base}/api/notifications/count"), &alice)
         .query(&[("did", alice.as_str())])
         .send()
         .await
@@ -487,8 +507,7 @@ async fn notifications_seen_respects_watermark() {
     assert_eq!(body["count"], json!(1));
 
     // snake_case spelling must work identically.
-    let resp = c
-        .post(format!("{base}/api/notifications/seen"))
+    let resp = authed_post(&c, &auth, format!("{base}/api/notifications/seen"), &alice)
         .json(&json!({ "did": alice, "seen_at": null }))
         .send()
         .await
@@ -502,7 +521,7 @@ async fn notifications_seen_respects_watermark() {
 #[tokio::test]
 async fn followers_and_following_lists() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, _auth)) = ready().await else {
         return;
     };
     let hub = did_for_test("hub");
@@ -591,7 +610,8 @@ async fn followers_and_following_lists() {
     assert_eq!(unique.len(), seen.len(), "paged followers repeat: {seen:?}");
     assert_eq!(seen.len(), 3, "paging lost a follower: {seen:?}");
 
-    // `did` is mandatory.
+    // `did` is mandatory. These two stay public — a follow edge is a
+    // public record — so no token is involved.
     for path in ["/api/followers", "/api/following"] {
         let resp = c
             .get(format!("{base}{path}"))
@@ -610,7 +630,7 @@ async fn followers_and_following_lists() {
 #[tokio::test]
 async fn thread_returns_parents_and_replies() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, _auth)) = ready().await else {
         return;
     };
     let a = did_for_test("root");
@@ -689,7 +709,7 @@ async fn thread_returns_parents_and_replies() {
 #[tokio::test]
 async fn post_by_uri_stays_backwards_compatible() {
     let base = appview_url();
-    let Some((c, _pool)) = ready().await else {
+    let Some((c, _pool, _auth)) = ready().await else {
         return;
     };
     let a = did_for_test("compat_a");