diff --git a/cmd/broker/main.go b/cmd/broker/main.go
index 261c2ce..0f7bd61 100644
--- a/cmd/broker/main.go
+++ b/cmd/broker/main.go
@@ -18,6 +18,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 	"time"
 
@@ -30,6 +31,7 @@ func main() {
 	window := flag.Duration("window", 5*time.Minute, "signed-request freshness window")
 	ipHeader := flag.String("ip-header", envOr("BROKER_IP_HEADER", "X-Real-IP"), "header carrying the real source IP (set by the front proxy; client X-Forwarded-For is never trusted)")
 	meterInterval := flag.Duration("meter-interval", 60*time.Second, "usage-metering tick for provisioned apps")
+	brokerName := flag.String("broker-name", envOr("BROKER_NAME", defaultHostname()), "process label on every access event")
 	flag.Parse()
 
 	reg, err := broker.LoadRegistry(*registryPath, os.Getenv)
@@ -37,6 +39,17 @@ func main() {
 		log.Fatalf("broker: %v", err)
 	}
 
+	// Access keys come from the environment only (never the registry file, which
+	// is world-readable config and lands in git).
+	accessKeys := broker.NewAccessKeys(strings.Split(os.Getenv("BROKER_ACCESS_KEYS"), ","))
+	// Refuse to boot if an app demands a key that does not exist. Without this the
+	// broker would come up "healthy" and 401 every caller — a config typo would
+	// read as a total outage with no explanation.
+	if n := reg.AppsRequiringAccessKey(); len(n) > 0 && accessKeys.Len() == 0 {
+		log.Fatalf("broker: apps %v require an access key but BROKER_ACCESS_KEYS is empty — "+
+			"set it (comma-separated, entries may be \"label:key\") or clear require_access_key", n)
+	}
+
 	// Durable store when BROKER_DB is set (prod); in-memory otherwise (dev).
 	var store interface {
 		broker.Store
@@ -61,6 +74,8 @@ func main() {
 	b := broker.New(reg, store)
 	b.Verify = broker.VerifyConfig{Window: *window}
 	b.IPTrust = broker.IPTrust{Header: *ipHeader}
+	b.AccessKeys = accessKeys
+	log.Printf("broker: %d access key(s) configured", accessKeys.Len())
 
 	// Usage meter: drain per-user credit by real machine usage, stopping machines
 	// at zero. Runs for provisioned apps whose registry entry carries a rate card.
@@ -91,15 +106,27 @@ func main() {
 	})
 	mux.Handle("/", b)
 
-	log.Printf("broker: listening on %s", *addr)
+	// Access logging: one structured `ACCESS {json}` line per forwarded request
+	// to stdout → journald (read by the monitoring log crawler). /gw/ and
+	// unrouted (scanner-noise) requests are skipped.
+	handler := broker.WithAccessLog(mux, *brokerName, broker.StdoutSink{W: os.Stdout}, nil)
+
+	log.Printf("broker: listening on %s (name=%s)", *addr, *brokerName)
 	srv := &http.Server{
 		Addr:              *addr,
-		Handler:           mux,
+		Handler:           handler,
 		ReadHeaderTimeout: 10 * time.Second,
 	}
 	log.Fatal(srv.ListenAndServe())
 }
 
+func defaultHostname() string {
+	if h, err := os.Hostname(); err == nil && h != "" {
+		return h
+	}
+	return "broker"
+}
+
 func envOr(k, def string) string {
 	if v := os.Getenv(k); v != "" {
 		return v
diff --git a/deploy/agentphone-tenancy.py b/deploy/agentphone-tenancy.py
new file mode 100644
index 0000000..732642a
--- /dev/null
+++ b/deploy/agentphone-tenancy.py
@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+"""Rewrite the io.pilot.agentphone registry entry for broker-enforced tenancy.
+
+The agentphone partner account is a single shared account (bound to a provider
+campaign that permits number generation), so the partner cannot tell pilot users
+apart and cannot isolate them. The broker therefore owns tenancy: this script
+declares which path/body fields name a resource, how each resource is claimed,
+and which list responses must be filtered to the caller's own rows.
+
+Run on the broker host; it rewrites /opt/pilot/registry/apps.json in place after
+taking a timestamped backup, and validates the JSON before writing.
+"""
+import json
+import re
+import shutil
+import sys
+import time
+
+P = "/opt/pilot/registry/apps.json"
+
+# Routes that cannot be made per-tenant on a shared account.
+#
+# The webhook routes are ACCOUNT-level: one tenant setting a webhook redirects
+# every tenant's events to their endpoint. There is no per-caller scoping to
+# apply, so they are removed rather than "checked" — the per-agent route
+# /v1/agents/{agent_id}/webhook stays and is gated by agent ownership.
+#
+# /v1/messages/{message_id}/reactions names a message id that no route can
+# establish ownership of, and an unmapped param is an unchecked resource. It is
+# dropped until messages carry an ownable link.
+DROP = {
+    "/v1/webhooks",
+    "/v1/webhooks/deliveries",
+    "/v1/webhooks/deliveries/stats",
+    "/v1/webhooks/test",
+    "/v1/messages/{message_id}/reactions",
+}
+
+# Spend routes. These are ENABLED: an AI agent cannot use AgentPhone without
+# buying a number and sending, so the app is unusable without them. They are now
+# safe to serve because (a) tenancy binds every send to a number/agent the caller
+# owns, and (b) the per-IP grant cap bounds how much free budget one network can
+# mint. Each still debits the caller's own $5 budget at its listed cost.
+SPEND = ["/v1/messages", "/v1/calls", "/v1/numbers"]
+
+TENANCY = {
+    # Every {param} that appears in an allow pattern MUST be mapped here. A param
+    # with no mapping is never ownership-checked, which is the same as public.
+    "param_types": {
+        "agent_id": "agent",
+        "number_id": "number",
+        "call_id": "call",
+        "conversation_id": "conversation",
+        "contact_id": "contact",
+    },
+    # Body/query fields that name a resource the call acts THROUGH. Sending names
+    # the sending agent in the body while the path is entirely the caller's own,
+    # so without these the path checks above prove nothing. Aliases are listed
+    # because the partner accepts more than one spelling.
+    "body_refs": {
+        "agent_id": "agent",
+        "agentId": "agent",
+        "number_id": "number",
+        "numberId": "number",
+        "phoneNumberId": "number",
+        "from_number": "number",
+        "conversation_id": "conversation",
+        "conversationId": "conversation",
+        "call_id": "call",
+        "callId": "call",
+        "contact_id": "contact",
+        "contactId": "contact",
+    },
+    "create": [
+        {"method": "POST", "path": "/v1/agents", "type": "agent", "id_field": "id"},
+        {"method": "POST", "path": "/v1/numbers", "type": "number", "id_field": "id"},
+        {"method": "POST", "path": "/v1/calls", "type": "call", "id_field": "id"},
+        {"method": "POST", "path": "/v1/contacts", "type": "contact", "id_field": "id"},
+    ],
+    "delete": [
+        {"method": "DELETE", "path": "/v1/numbers/{number_id}", "type": "number", "param": "number_id"},
+        {"method": "DELETE", "path": "/v1/agents/{agent_id}", "type": "agent", "param": "agent_id"},
+        {"method": "DELETE", "path": "/v1/contacts/{contact_id}", "type": "contact", "param": "contact_id"},
+    ],
+    # Only ACCOUNT-WIDE lists are declared here.
+    #
+    # A sub-resource list under an owned parent (e.g. /v1/numbers/{number_id}/messages)
+    # is already authorized by the parent's ownership check, and its elements carry
+    # no ownable link of their own — declaring it would drop every row and hide the
+    # owner's own messages from them.
+    "list": [
+        {"method": "GET", "path": "/v1/agents", "array": "data",
+         "owner_by": [{"field": "id", "type": "agent"}], "claim_as": "agent",
+         "count_fields": ["total"]},
+        {"method": "GET", "path": "/v1/numbers", "array": "data",
+         "owner_by": [{"field": "id", "type": "number"}], "claim_as": "number",
+         "count_fields": ["total"]},
+        {"method": "GET", "path": "/v1/contacts", "array": "data",
+         "owner_by": [{"field": "id", "type": "contact"}], "claim_as": "contact",
+         "count_fields": ["total"]},
+        # Inbound calls/conversations are created by the partner, so they are
+        # attributable only through the number/agent they hang off. claim_as makes
+        # them fetchable by id afterwards.
+        {"method": "GET", "path": "/v1/calls", "array": "data",
+         "owner_by": [{"field": "phoneNumberId", "type": "number"},
+                      {"field": "agentId", "type": "agent"}], "claim_as": "call",
+         "count_fields": ["total"]},
+        {"method": "GET", "path": "/v1/conversations", "array": "data",
+         "owner_by": [{"field": "phoneNumberId", "type": "number"},
+                      {"field": "agentId", "type": "agent"}], "claim_as": "conversation",
+         "count_fields": ["total"]},
+        {"method": "GET", "path": "/v1/usage/by-number", "array": "data",
+         "owner_by": [{"field": "numberId", "type": "number"}]},
+        {"method": "GET", "path": "/v1/usage/by-agent", "array": "data",
+         "owner_by": [{"field": "agentId", "type": "agent"}]},
+    ],
+    # /v1/usage summarises the SHARED account: its number count and message/call
+    # totals span every tenant. Left alone it is a side-channel — a caller who can
+    # see none of the resources can still read how many exist and watch the totals
+    # move. numbers.used is recomputed from the caller's own ledger; the stats
+    # block cannot be attributed to a single tenant, so it is dropped rather than
+    # reported wrongly.
+    "object": [
+        {"method": "GET", "path": "/v1/usage",
+         "owned_counts": {"numbers.used": "number"},
+         # numbers.remaining is the partner's limit minus the ACCOUNT-WIDE used, so
+         # it discloses the very count numbers.used was just scoped to hide.
+         # A derived field leaks whatever it was derived from.
+         "redact": ["stats", "numbers.remaining"]},
+    ],
+}
+
+
+def main():
+    apps = json.load(open(P))
+    shutil.copy(P, P + ".bak-" + time.strftime("%Y%m%d-%H%M%S"))
+
+    for a in apps:
+        if a["id"] != "io.pilot.agentphone":
+            continue
+        before = len(a["allow"])
+        # Drop unsafe routes; then ensure spend routes are present (containment
+        # removed them, this re-enables them). Order-preserving, no duplicates.
+        allow = [p for p in a["allow"] if p not in DROP and p not in SPEND]
+        allow += [p for p in SPEND if p not in allow]
+        a["allow"] = allow
+        a["tenancy"] = TENANCY
+        # Per-IP GRANT cap: at most this many funded identities per source IP.
+        # It bounds free budget minted from one network; the (N+1)th identity is
+        # recorded with a zero grant (reads still work, spend 402s), so a shared
+        # NAT is not hard-locked out. It is a speed bump, not a boundary — a caller
+        # with many source IPs is not constrained — so it is never the only control.
+        a.setdefault("credit", {})["max_identities_per_ip"] = 3
+        print("allow: %d -> %d  (spend enabled: %s)" % (before, len(allow), ", ".join(SPEND)))
+
+        # Fail loudly if any {param} still lacks an ownership mapping: an unmapped
+        # param is an unchecked resource, which is exactly the bug class this fixes.
+        params = set()
+        for p in a["allow"]:
+            params |= set(re.findall(r"{(\w+)}", p))
+        unmapped = params - set(TENANCY["param_types"])
+        if unmapped:
+            sys.exit("REFUSING: unmapped path params would be unchecked: %s" % sorted(unmapped))
+        print("all path params mapped: %s" % sorted(params))
+
+    json.dump(apps, open(P, "w"), indent=2)
+    json.load(open(P))  # re-parse: never leave invalid JSON behind
+    print("registry written + valid")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/deploy/setup-broker-tls.sh b/deploy/setup-broker-tls.sh
index be84028..35344ed 100755
--- a/deploy/setup-broker-tls.sh
+++ b/deploy/setup-broker-tls.sh
@@ -65,6 +65,11 @@ server {
     location / {
         proxy_pass $ORIGIN;
         proxy_set_header Host \$host;
+        # X-Real-IP is the ONLY source-IP signal the broker trusts (a
+        # client-supplied X-Forwarded-For is ignored). The per-IP grant cap
+        # depends on it: omit it and every caller reads as 127.0.0.1, collapsing
+        # every identity into one IP bucket. It must be set on EVERY location.
+        proxy_set_header X-Real-IP \$remote_addr;
         proxy_set_header X-Forwarded-For \$proxy_add_x_forwarded_for;
         proxy_set_header X-Forwarded-Proto https;
     }
diff --git a/internal/broker/accesskey.go b/internal/broker/accesskey.go
new file mode 100644
index 0000000..9fb2c48
--- /dev/null
+++ b/internal/broker/accesskey.go
@@ -0,0 +1,123 @@
+package broker
+
+import (
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/hex"
+	"net/http"
+	"strings"
+)
+
+// Access keys are the broker's front door.
+//
+// A signed pilot identity (identity.go) proves the caller holds the private key
+// for a public key, but the caller generates that keypair itself. It is a
+// pseudonym: it attributes usage, it does not authorize it.
+//
+// The access key is the other half. Identity says WHO you are; the access key
+// says you are ALLOWED TO BE HERE. Both are required, and neither substitutes
+// for the other: a shared key cannot attribute usage to a user, and a
+// self-generated identity cannot establish that a caller is a pilot client at
+// all. Any app that grants something of value on first contact needs both.
+//
+// Keys are compared as SHA-256 digests in constant time, and only the digests
+// are held in memory, so the broker never keeps a usable secret it could leak
+// through a crash dump or a debug endpoint.
+type AccessKeys struct {
+	digests map[string]string // hex(sha256(key)) -> label, for revocation + attribution
+}
+
+// AccessKeyHeader is the dedicated header. Authorization: Bearer <key> is also
+// accepted so ordinary HTTP clients work without a custom header.
+const AccessKeyHeader = "X-Pilot-Access-Key"
+
+// NewAccessKeys builds the key set from "label:key" (or bare "key") entries.
+// Blank entries are ignored so a trailing comma in config is not a silent
+// "no keys configured" — which would otherwise disable the gate.
+func NewAccessKeys(entries []string) *AccessKeys {
+	ak := &AccessKeys{digests: map[string]string{}}
+	for _, e := range entries {
+		e = strings.TrimSpace(e)
+		if e == "" {
+			continue
+		}
+		label, key := "", e
+		if i := strings.Index(e, ":"); i > 0 {
+			label, key = e[:i], e[i+1:]
+		}
+		key = strings.TrimSpace(key)
+		if key == "" {
+			continue
+		}
+		sum := sha256.Sum256([]byte(key))
+		ak.digests[hex.EncodeToString(sum[:])] = label
+	}
+	return ak
+}
+
+// Len reports how many keys are configured (diagnostics; never logs the keys).
+func (a *AccessKeys) Len() int {
+	if a == nil {
+		return 0
+	}
+	return len(a.digests)
+}
+
+// presented pulls the candidate key off a request.
+func presented(h func(string) string) string {
+	if v := strings.TrimSpace(h(AccessKeyHeader)); v != "" {
+		return v
+	}
+	// Authorization: Bearer <key>
+	if v := strings.TrimSpace(h("Authorization")); v != "" {
+		if len(v) > 7 && strings.EqualFold(v[:7], "bearer ") {
+			return strings.TrimSpace(v[7:])
+		}
+	}
+	return ""
+}
+
+// Check verifies the presented key and returns its label.
+//
+// It FAILS CLOSED in the case that matters most: an AccessKeys with no keys
+// configured authorizes NOTHING. A missing or blank BROKER_ACCESS_KEYS must
+// never degrade to "open to everyone" — an absent control has to read as "deny",
+// not as "allow". main refuses to boot in that state rather than serve.
+func (a *AccessKeys) Check(h func(string) string) (label string, ok bool) {
+	if a == nil || len(a.digests) == 0 {
+		return "", false
+	}
+	key := presented(h)
+	if key == "" {
+		return "", false
+	}
+	sum := sha256.Sum256([]byte(key))
+	want := hex.EncodeToString(sum[:])
+	// Constant-time scan over every digest: comparing hex digests of a hash means
+	// a timing signal cannot reveal the key, and not short-circuiting keeps the
+	// work independent of which key matched.
+	found, lbl := 0, ""
+	for d, l := range a.digests {
+		if subtle.ConstantTimeCompare([]byte(d), []byte(want)) == 1 {
+			found, lbl = 1, l
+		}
+	}
+	return lbl, found == 1
+}
+
+// requireAccessKey enforces the gate for an app that demands one. It answers 401
+// with a WWW-Authenticate hint and no detail about why: a caller learns only
+// that a valid key is required, never whether a key exists or was close.
+func (b *Broker) requireAccessKey(w http.ResponseWriter, r *http.Request, app *AppEntry) bool {
+	if !app.RequireAccessKey {
+		return true
+	}
+	if _, ok := b.AccessKeys.Check(r.Header.Get); ok {
+		return true
+	}
+	w.Header().Set("WWW-Authenticate", `Bearer realm="pilot-broker"`)
+	writeJSON(w, http.StatusUnauthorized, map[string]string{
+		"error": "this app requires a pilot access key — upgrade your pilot client (`pilotctl appstore upgrade`) or see https://pilotprotocol.network/docs/access-keys",
+	})
+	return false
+}
diff --git a/internal/broker/accesslog.go b/internal/broker/accesslog.go
new file mode 100644
index 0000000..b473d1c
--- /dev/null
+++ b/internal/broker/accesslog.go
@@ -0,0 +1,130 @@
+package broker
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// AccessEvent is one broker request, emitted per request for monitoring. The
+// JSON tags are the wire/DB contract — keep them stable.
+type AccessEvent struct {
+	TS         time.Time `json:"ts"`
+	Broker     string    `json:"broker"`
+	AppID      string    `json:"app_id"`
+	HTTPMethod string    `json:"http_method"`
+	Path       string    `json:"path"`
+	Status     int       `json:"status"`
+	DurationMs int64     `json:"duration_ms"`
+	ReqBytes   int64     `json:"req_bytes"`
+	RespBytes  int64     `json:"resp_bytes"`
+	Method     string    `json:"method,omitempty"` // pilotctl command, from X-Pilot-Method
+	Source     string    `json:"source,omitempty"`
+}
+
+// AccessSink receives one event per request; MUST NOT block the request path.
+type AccessSink interface{ Emit(AccessEvent) }
+
+// AccessLogPrefix marks stdout access lines so the crawler can select them out
+// of the broker's mixed journald stream.
+const AccessLogPrefix = "ACCESS "
+
+// PilotMethodHeader carries the pilotctl command (`<app>.<method>`) an app
+// adapter stamps on a brokered call, making stats command-oriented.
+const PilotMethodHeader = "X-Pilot-Method"
+
+// UnroutedHeader is set by the broker on requests it could not route to a
+// managed app (unknown app / bad route); the access log skips these so
+// internet-scanner noise never counts as broker traffic.
+const UnroutedHeader = "X-Pilot-Unrouted"
+
+// StdoutSink writes one prefixed JSON line per event to W (os.Stdout in prod).
+type StdoutSink struct{ W io.Writer }
+
+func (s StdoutSink) Emit(e AccessEvent) {
+	b, err := json.Marshal(e)
+	if err != nil {
+		return
+	}
+	line := make([]byte, 0, len(AccessLogPrefix)+len(b)+1)
+	line = append(line, AccessLogPrefix...)
+	line = append(line, b...)
+	line = append(line, '\n')
+	_, _ = s.W.Write(line) // single Write so concurrent goroutines don't interleave
+}
+
+// statusRecorder captures the response status code and byte count.
+type statusRecorder struct {
+	http.ResponseWriter
+	status int
+	nbytes int64
+}
+
+func (r *statusRecorder) WriteHeader(code int) {
+	r.status = code
+	r.ResponseWriter.WriteHeader(code)
+}
+
+func (r *statusRecorder) Write(b []byte) (int, error) {
+	if r.status == 0 {
+		r.status = http.StatusOK
+	}
+	n, err := r.ResponseWriter.Write(b)
+	r.nbytes += int64(n)
+	return n, err
+}
+
+// appIDFromPath splits /<app-id>/<rest> exactly as ServeHTTP does.
+func appIDFromPath(p string) (appID, method string) {
+	appID, rest, ok := strings.Cut(strings.TrimPrefix(p, "/"), "/")
+	if !ok {
+		return appID, "/"
+	}
+	return appID, "/" + rest
+}
+
+// WithAccessLog wraps next, emitting one AccessEvent per forwarded request to
+// sink. broker labels the process; now is injectable for tests (nil → time.Now).
+// /gw/ paths, a nil sink, and requests the broker flagged X-Pilot-Unrouted pass
+// through unlogged.
+func WithAccessLog(next http.Handler, broker string, sink AccessSink, now func() time.Time) http.Handler {
+	if now == nil {
+		now = time.Now
+	}
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if sink == nil || strings.HasPrefix(r.URL.Path, "/gw/") {
+			next.ServeHTTP(w, r)
+			return
+		}
+		start := now()
+		pilotMethod := r.Header.Get(PilotMethodHeader)
+		rec := &statusRecorder{ResponseWriter: w}
+		next.ServeHTTP(rec, r)
+
+		if rec.Header().Get(UnroutedHeader) != "" { // scanner noise / unknown app
+			return
+		}
+		appID, method := appIDFromPath(r.URL.Path)
+		if rec.status == 0 {
+			rec.status = http.StatusOK
+		}
+		reqBytes := r.ContentLength
+		if reqBytes < 0 {
+			reqBytes = 0
+		}
+		sink.Emit(AccessEvent{
+			TS:         start,
+			Broker:     broker,
+			AppID:      appID,
+			HTTPMethod: r.Method,
+			Path:       method,
+			Method:     pilotMethod,
+			Status:     rec.status,
+			DurationMs: now().Sub(start).Milliseconds(),
+			ReqBytes:   reqBytes,
+			RespBytes:  rec.nbytes,
+		})
+	})
+}
diff --git a/internal/broker/accesslog_test.go b/internal/broker/accesslog_test.go
new file mode 100644
index 0000000..671b0bb
--- /dev/null
+++ b/internal/broker/accesslog_test.go
@@ -0,0 +1,104 @@
+package broker
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+type capSink struct{ events []AccessEvent }
+
+func (c *capSink) Emit(e AccessEvent) { c.events = append(c.events, e) }
+
+// fixedNow returns a clock that advances by one fixed step per call, so
+// DurationMs is deterministic (start=T0, end=T0+step).
+func fixedNow(base time.Time, step time.Duration) func() time.Time {
+	n := -1
+	return func() time.Time {
+		n++
+		if n == 0 {
+			return base
+		}
+		return base.Add(step)
+	}
+}
+
+// TestAccessLog_EmitsPerForwardedRequest: a routed request produces exactly one
+// event carrying the app id, method-path, status, byte counts, the pilotctl
+// command from X-Pilot-Method, and a duration.
+func TestAccessLog_EmitsPerForwardedRequest(t *testing.T) {
+	inner := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(200)
+		_, _ = w.Write([]byte(`{"ok":true}`))
+	})
+	sink := &capSink{}
+	h := WithAccessLog(inner, "pilot-broker", sink, fixedNow(time.Unix(1_800_000_000, 0), 7*time.Millisecond))
+
+	req := httptest.NewRequest("POST", "/io.pilot.agentphone/v1/messages", strings.NewReader(`{"x":1}`))
+	req.Header.Set(PilotMethodHeader, "agentphone.send_message")
+	h.ServeHTTP(httptest.NewRecorder(), req)
+
+	if len(sink.events) != 1 {
+		t.Fatalf("events = %d, want 1", len(sink.events))
+	}
+	e := sink.events[0]
+	if e.Broker != "pilot-broker" || e.AppID != "io.pilot.agentphone" || e.Path != "/v1/messages" {
+		t.Errorf("bad routing fields: %+v", e)
+	}
+	if e.HTTPMethod != "POST" || e.Status != 200 {
+		t.Errorf("bad http fields: %+v", e)
+	}
+	if e.Method != "agentphone.send_message" {
+		t.Errorf("X-Pilot-Method not captured: %q", e.Method)
+	}
+	if e.DurationMs != 7 {
+		t.Errorf("DurationMs = %d, want 7", e.DurationMs)
+	}
+	if e.RespBytes == 0 {
+		t.Error("RespBytes not recorded")
+	}
+}
+
+// TestAccessLog_SkipsUnroutedAndHealth: /gw/ health and requests the broker
+// flagged X-Pilot-Unrouted (unknown app / bad route) emit nothing, so scanner
+// noise never counts as broker traffic.
+func TestAccessLog_SkipsUnroutedAndHealth(t *testing.T) {
+	unrouted := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set(UnroutedHeader, "1")
+		w.WriteHeader(404)
+	})
+	sink := &capSink{}
+	WithAccessLog(unrouted, "b", sink, nil).ServeHTTP(httptest.NewRecorder(),
+		httptest.NewRequest("GET", "/HNAP1", nil))
+	if len(sink.events) != 0 {
+		t.Errorf("unrouted request logged: %+v", sink.events)
+	}
+
+	health := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(200) })
+	WithAccessLog(health, "b", sink, nil).ServeHTTP(httptest.NewRecorder(),
+		httptest.NewRequest("GET", "/gw/health", nil))
+	if len(sink.events) != 0 {
+		t.Errorf("/gw/health logged: %+v", sink.events)
+	}
+}
+
+// TestStdoutSink_PrefixedJSONLine: the sink writes one AccessLogPrefix-tagged
+// JSON line, which the crawler selects out of the mixed journald stream.
+func TestStdoutSink_PrefixedJSONLine(t *testing.T) {
+	var buf strings.Builder
+	StdoutSink{W: &buf}.Emit(AccessEvent{Broker: "b", AppID: "io.pilot.x", Status: 200})
+	out := buf.String()
+	if !strings.HasPrefix(out, AccessLogPrefix) || !strings.HasSuffix(out, "\n") {
+		t.Fatalf("bad line framing: %q", out)
+	}
+	var e AccessEvent
+	if err := json.Unmarshal([]byte(strings.TrimPrefix(strings.TrimSpace(out), AccessLogPrefix)), &e); err != nil {
+		t.Fatalf("payload not JSON: %v", err)
+	}
+	if e.AppID != "io.pilot.x" {
+		t.Errorf("round-trip lost app_id: %+v", e)
+	}
+}
diff --git a/internal/broker/broker.go b/internal/broker/broker.go
index 1c8925a..37ed9cf 100644
--- a/internal/broker/broker.go
+++ b/internal/broker/broker.go
@@ -3,9 +3,12 @@ package broker
 import (
 	"bytes"
 	"context"
+	"crypto/rand"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"math"
 	"net/http"
 	"strconv"
@@ -27,6 +30,10 @@ type Broker struct {
 	// IPTrust says which header carries the real source IP (set by the front
 	// proxy). Client-supplied X-Forwarded-For is never trusted.
 	IPTrust IPTrust
+	// AccessKeys authorizes callers of apps with require_access_key. A signed
+	// identity only proves possession of a self-minted key; this proves the
+	// caller is an authorized pilot client at all.
+	AccessKeys *AccessKeys
 
 	reg atomic.Pointer[Registry] // hot-swappable so the registry can reload without dropping traffic
 }
@@ -78,15 +85,12 @@ func (b *Broker) serveCreditBalance(w http.ResponseWriter, r *http.Request, app
 		return
 	}
 	ip := clientIP(r.Header.Get, r.RemoteAddr, b.IPTrust)
-	rec, err := ps.Provision(app.ID, caller, ip, app.creditSeed, app.creditMaxPerIP, app.creditMintCooldown, b.now())
+	rec, err := b.seedCaller(ps, app.ID, caller, ip, app)
 	if err != nil {
-		switch err {
-		case ErrIPCap:
-			writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "per-IP identity cap reached — too many pilot identities have claimed a budget from this network"})
-		case ErrCooldown:
+		if err == ErrCooldown {
 			writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "re-provision cooldown — retry shortly"})
-		default:
-			writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "balance: " + err.Error()})
+		} else {
+			b.internalError(w, http.StatusInternalServerError, app.ID, "balance", err)
 		}
 		return
 	}
@@ -101,10 +105,35 @@ func (b *Broker) serveCreditBalance(w http.ResponseWriter, r *http.Request, app
 	})
 }
 
+// seedCaller records the caller in the credit ledger, enforcing the per-IP
+// GRANT cap without ever hard-blocking access.
+//
+// On first sight it grants the app's seed budget, unless the caller is the
+// (N+1)th distinct identity on its source IP — in which case it is recorded with
+// a ZERO grant instead of being refused. The distinction matters: the cap exists
+// to bound how much free budget one network can mint, not to decide who may call
+// the broker. A zero-grant caller can still make free (read) calls; every priced
+// call it attempts hits 402 for lack of budget. A caller already in the ledger
+// keeps whatever balance it has (no re-seed, no second cap check).
+//
+// This is what lets the cap be set aggressively low: a tight economic bound
+// costs a shared-NAT newcomer only its free grant, not its access.
+func (b *Broker) seedCaller(ps ProvisionStore, app, caller, ip string, a *AppEntry) (ProvisionRecord, error) {
+	rec, err := ps.Provision(app, caller, ip, a.creditSeed, a.creditMaxPerIP, a.creditMintCooldown, b.now())
+	if err == ErrIPCap {
+		// Over the per-IP grant cap: record the identity with no budget and no cap
+		// (seed 0, maxPerIP 0). Idempotent, so a subsequent call takes the normal
+		// repeat path and stays at zero rather than ever back-filling a grant.
+		return ps.Provision(app, caller, ip, 0, 0, 0, b.now())
+	}
+	return rec, err
+}
+
 // ServeHTTP is the forward path for /<app-id>/<method-path>.
 func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	appID, mpath, ok := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), "/")
 	if !ok || appID == "" {
+		w.Header().Set(UnroutedHeader, "1") // not a managed-app call — monitoring skips it
 		writeJSON(w, http.StatusNotFound, map[string]string{"error": "route must be /<app-id>/<path>"})
 		return
 	}
@@ -114,10 +143,18 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// the app's artifact limit before reading.
 	app := b.reg.Load().Get(appID)
 	if app == nil {
+		w.Header().Set(UnroutedHeader, "1") // unknown app (typo / internet scanner) — not real broker usage
 		writeJSON(w, http.StatusNotFound, map[string]string{"error": "unknown app: " + appID})
 		return
 	}
 
+	// Access key BEFORE anything else that costs or grants: an unauthorized caller
+	// must not be able to reach the credit ledger, seed a grant, or touch the
+	// master key. This is the gate that self-minted identities cannot walk past.
+	if !b.requireAccessKey(w, r, app) {
+		return
+	}
+
 	maxBody := b.MaxBody
 	if app.Provision != nil && mpath == app.Provision.PushPath {
 		maxBody = app.Provision.ArtifactMaxBytes
@@ -154,11 +191,26 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// 3. Is this an allowed method? (no open proxy onto the master key)
-	if !app.allowed(mpath) {
+	if !app.allowed(r.Method, mpath) {
 		writeJSON(w, http.StatusForbidden, map[string]string{"error": "method not allowed for this app: " + mpath})
 		return
 	}
 
+	// 3b. TENANCY: on a shared partner account the broker is the only thing that
+	//     knows which pilot user owns what. Every resource named in the path,
+	//     query, or body must belong to this caller. This runs BEFORE the forward,
+	//     so a refused write never reaches the partner.
+	//
+	//     The answer is 404, never 403: "not yours" and "does not exist" must be
+	//     indistinguishable, otherwise the broker is an oracle for enumerating
+	//     other tenants' resource ids.
+	if app.Tenancy != nil {
+		if _, ok := app.Tenancy.EnforceRequest(b.ownerStore(), appID, app.allowSegs, r.Method, mpath, r.URL.RawQuery, body, string(caller)); !ok {
+			writeJSON(w, http.StatusNotFound, map[string]string{"error": "not found"})
+			return
+		}
+	}
+
 	// 4. Circuit breaker: if the partner has been failing, fail fast and don't
 	//    spend a credit or touch the master key.
 	if !app.breaker.Allow() {
@@ -187,18 +239,17 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		ip := clientIP(r.Header.Get, r.RemoteAddr, b.IPTrust)
-		// Seed the caller on first sight, enforcing the per-IP identity cap: once a
-		// caller depletes their budget (402 below) they can't farm a fresh $5 grant by
-		// minting a new pilot identity from the same IP — the (N+1)th distinct caller
-		// on that IP is refused here with 429.
-		if _, err := ps.Provision(appID, string(caller), ip, app.creditSeed, app.creditMaxPerIP, app.creditMintCooldown, b.now()); err != nil {
-			switch err {
-			case ErrIPCap:
-				writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "per-IP identity cap reached — too many pilot identities have claimed a budget from this network"})
-			case ErrCooldown:
+		// Seed the caller on first sight, enforcing the per-IP GRANT cap. The cap
+		// bounds free money, not access: the (N+1)th distinct identity on one source
+		// IP is recorded with a ZERO grant rather than refused outright. It can still
+		// make free (read) calls, but every priced call hits 402 — so nobody can farm
+		// a fresh budget by minting identities from one network, and a legitimate user
+		// behind a shared NAT is not hard-locked out of the app.
+		if _, err := b.seedCaller(ps, appID, string(caller), ip, app); err != nil {
+			if err == ErrCooldown {
 				writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "re-provision cooldown — retry shortly"})
-			default:
-				writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "provision: " + err.Error()})
+			} else {
+				b.internalError(w, http.StatusInternalServerError, appID, "provision", err)
 			}
 			return
 		}
@@ -211,7 +262,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			if billable {
 				bal, derr := ps.Credit(appID, string(caller))
 				if derr != nil {
-					writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "credit: " + derr.Error()})
+					b.internalError(w, http.StatusInternalServerError, appID, "credit", derr)
 					return
 				}
 				if bal <= 0 {
@@ -227,7 +278,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			creditCost = app.costForCall(r.Method, mpath)
 			admittedCredit, remaining, derr := ps.Debit(appID, string(caller), creditCost)
 			if derr != nil {
-				writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "debit: " + derr.Error()})
+				b.internalError(w, http.StatusInternalServerError, appID, "debit", derr)
 				return
 			}
 			if !admittedCredit {
@@ -262,7 +313,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ureq, err := http.NewRequestWithContext(ctx, r.Method, target, bytes.NewReader(body))
 	if err != nil {
 		refundCredit()
-		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "build upstream: " + err.Error()})
+		b.internalError(w, http.StatusBadGateway, appID, "build upstream", err)
 		return
 	}
 	ureq.Header.Set("Content-Type", "application/json")
@@ -272,7 +323,7 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		app.breaker.Record(false)
 		refundCredit()
-		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upstream: " + err.Error()})
+		b.internalError(w, http.StatusBadGateway, appID, "upstream", err)
 		return
 	}
 	defer resp.Body.Close()
@@ -296,6 +347,21 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	// 7. TENANCY on the way back out. Two jobs, both only on success:
+	//    - claim what this call created, so the caller owns it going forward;
+	//    - filter list answers down to the caller's own resources, because the
+	//      partner answered for the WHOLE shared account.
+	if app.Tenancy != nil && resp.StatusCode/100 == 2 {
+		os := b.ownerStore()
+		app.Tenancy.ClaimFrom(os, appID, r.Method, mpath, rb, string(caller), b.now())
+		app.Tenancy.ReleaseFrom(os, appID, app.allowSegs, r.Method, mpath)
+		if filtered, did := app.Tenancy.FilterResponse(os, appID, r.Method, mpath, rb, string(caller), b.now()); did {
+			rb = filtered
+		} else if filtered, did := app.Tenancy.FilterObject(os, appID, r.Method, mpath, rb, string(caller)); did {
+			rb = filtered
+		}
+	}
+
 	// Surface the caller's remaining budget on every metered response.
 	if app.creditEnabled() {
 		if bal, err := ps.Credit(appID, string(caller)); err == nil {
@@ -307,3 +373,29 @@ func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(resp.StatusCode)
 	_, _ = w.Write(rb)
 }
+
+// internalError logs the real cause server-side and returns an opaque reference
+// to the caller.
+//
+// Raw error strings from the store or the upstream dialer describe internal
+// topology — hostnames, private addresses, database paths, driver states. None
+// of that helps a legitimate caller (it is never something they can fix), and
+// all of it helps someone mapping the deployment. The caller gets a correlation
+// id to quote in a bug report; the detail stays in the log.
+func (b *Broker) internalError(w http.ResponseWriter, code int, appID, stage string, err error) {
+	ref := errorRef()
+	log.Printf("broker: %s: %s: ref=%s: %v", appID, stage, ref, err)
+	writeJSON(w, code, map[string]string{
+		"error": "the broker could not complete this call",
+		"ref":   ref,
+	})
+}
+
+// errorRef mints a short, non-guessable correlation id for one failure.
+func errorRef() string {
+	var b [6]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return "unknown"
+	}
+	return hex.EncodeToString(b[:])
+}
diff --git a/internal/broker/owner.go b/internal/broker/owner.go
new file mode 100644
index 0000000..b76414f
--- /dev/null
+++ b/internal/broker/owner.go
@@ -0,0 +1,77 @@
+package broker
+
+import (
+	"crypto/subtle"
+	"errors"
+	"time"
+)
+
+// constantTimeEqual compares two identity strings without leaking their contents
+// through timing. Owner ids are public keys, but comparing them in constant time
+// costs nothing and keeps the authorization predicate free of an oracle.
+func constantTimeEqual(a, b string) bool {
+	return subtle.ConstantTimeCompare([]byte(a), []byte(b)) == 1
+}
+
+// OwnerStore is the broker-side resource-ownership ledger: it records WHICH
+// caller owns WHICH upstream resource id, for apps whose partner account cannot
+// be split per user.
+//
+// Why this exists: some managed apps are one shared partner account that cannot
+// be split per user (e.g. an account bound to a provider campaign). Upstream,
+// every pilot user is the same customer, so the partner cannot tell which user a
+// resource belongs to and cannot enforce isolation. The broker therefore has to:
+// every resource is claimed by its creator, and anything not provably owned by
+// the caller is invisible to them.
+//
+// Deny-by-default is the whole point: a resource with NO ledger row is owned by
+// NOBODY and is refused to EVERYONE (including pre-existing resources created
+// before this ledger shipped). That is deliberate — it is what makes the legacy
+// shared test number unreachable rather than up for grabs.
+type OwnerStore interface {
+	// Claim records caller as the owner of (app, rtype, rid). It is idempotent
+	// and FIRST-WRITER-WINS: if a different caller already owns the resource the
+	// existing owner is kept and ErrOwned is returned. It must never silently
+	// transfer ownership — that would be a takeover primitive.
+	Claim(app, rtype, rid, caller string, now time.Time) error
+	// OwnerOf returns the recorded owner of a resource, if any.
+	OwnerOf(app, rtype, rid string) (caller string, found bool, err error)
+	// OwnedSet returns the set of resource ids of rtype owned by caller. Used to
+	// filter list responses down to the caller's own resources.
+	OwnedSet(app, rtype, caller string) (map[string]bool, error)
+	// Release drops a resource row (e.g. a number was released upstream) so the
+	// id can be re-claimed if the partner recycles it.
+	Release(app, rtype, rid string) error
+}
+
+// ErrOwned is returned by Claim when the resource already belongs to a different
+// caller. It signals an attempted takeover, not a routine race.
+var ErrOwned = errors.New("broker: resource already owned by another caller")
+
+// Owns reports whether caller owns (app, rtype, rid). It is the single
+// authorization predicate for tenancy and FAILS CLOSED: any store error, or an
+// unknown/unclaimed resource, returns false. Callers must treat false as "deny".
+func Owns(s OwnerStore, app, rtype, rid, caller string) bool {
+	if s == nil || rid == "" || caller == "" {
+		return false
+	}
+	owner, found, err := s.OwnerOf(app, rtype, rid)
+	if err != nil || !found {
+		return false
+	}
+	return constantTimeEqual(owner, caller)
+}
+
+// ownerStore returns the tenancy ledger if the configured Store implements it.
+//
+// It returns nil when the store cannot record ownership. Every tenancy check
+// treats a nil ledger as "deny" (see Owns / EnforceRequest), so an app that
+// declares tenancy against a store that cannot enforce it fails CLOSED — it
+// refuses traffic instead of silently serving it unisolated.
+func (b *Broker) ownerStore() OwnerStore {
+	os, ok := b.Store.(OwnerStore)
+	if !ok {
+		return nil
+	}
+	return os
+}
diff --git a/internal/broker/registry.go b/internal/broker/registry.go
index 15e229b..205c430 100644
--- a/internal/broker/registry.go
+++ b/internal/broker/registry.go
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"sort"
 	"strings"
 	"time"
 )
@@ -41,10 +42,23 @@ type AppEntry struct {
 	// minting. nil ⇒ no budget (call-count Quota still applies).
 	Credit *CreditSpec `json:"credit,omitempty"`
 
-	master        string       // resolved from KeyEnv at load (managed: partner key; provisioned: cloud master, e.g. smk_)
-	injector      AuthInjector // built from AuthHeader/Scheme
-	allowSet      map[string]bool
-	allowPatterns [][]string // templated allow entries split on "/" ("{x}" matches any one segment)
+	// Tenancy, when set, makes the broker enforce per-caller resource isolation
+	// on a SHARED partner account: every resource is claimed by its creator and a
+	// caller may only reference/see resources it owns (see tenancy.go). Required
+	// for any app whose partner account cannot be split per user.
+	Tenancy *Tenancy `json:"tenancy,omitempty"`
+
+	// RequireAccessKey gates this app behind a shared pilot access key (see
+	// accesskey.go), on top of the signed identity. Set it for any app that hands
+	// out something of value (a credit grant, a phone number) to a caller who is
+	// otherwise just a self-minted keypair.
+	RequireAccessKey bool `json:"require_access_key"`
+
+	master        string          // resolved from KeyEnv at load (managed: partner key; provisioned: cloud master, e.g. smk_)
+	injector      AuthInjector    // built from AuthHeader/Scheme
+	allowSet      map[string]bool // key = costKey(method, path); method "" = any method
+	allowPatterns []allowPattern  // templated allow entries ("{x}" matches any one segment)
+	allowSegs     [][]string      // every templated allow path (method-independent), for tenancy param extraction
 	breaker       *Breaker
 
 	creditSeed         int            // Credit.SeedCredits (0 ⇒ no budget)
@@ -165,6 +179,17 @@ type CreditSpec struct {
 	BalancePath string `json:"balance_path"`
 }
 
+// Safe defaults for the credit/Sybil guards. These apply when a registry omits
+// the field, so a new app cannot ship with the guard silently off.
+const (
+	// defaultMaxIdentitiesPerIP caps distinct pilot identities that may claim a
+	// budget from one source IP. Low enough to make bulk identity creation
+	// costly, high enough for a shared NAT / office egress to onboard real users.
+	// It is a speed bump, not a boundary: a caller with many source addresses is
+	// not constrained by it, so it must never be the only control on a grant.
+	defaultMaxIdentitiesPerIP = 3
+)
+
 // costPattern is a templated cost key split into segments (like allowPatterns),
 // optionally scoped to one HTTP method ("" = any method).
 type costPattern struct {
@@ -174,7 +199,13 @@ type costPattern struct {
 }
 
 // creditEnabled reports whether this app meters a per-caller micro-dollar budget.
-func (a *AppEntry) creditEnabled() bool { return a.creditSeed > 0 }
+// creditEnabled reports whether the per-caller budget ledger is active.
+//
+// It keys off the PRESENCE of the credit block, never off the seed amount, so
+// that the ledger's existence and the size of a grant stay independent settings.
+// `seed_credits: 0` therefore means exactly what it reads like — a zero budget,
+// in which every priced path is refused with 402 — rather than "no metering".
+func (a *AppEntry) creditEnabled() bool { return a.Credit != nil }
 
 // costKey is the exact-map key for a (method, path) cost entry ("" method = any).
 func costKey(method, path string) string { return method + " " + path }
@@ -224,22 +255,36 @@ func (a *AppEntry) costForCall(method, path string) int {
 // "/v1/calls/{call_id}") match any single non-empty segment in that position, so
 // REST path params don't each need enumerating. An empty allow-list permits
 // nothing (safe default; prod must declare).
-func (a *AppEntry) allowed(path string) bool {
+// Entries are METHOD-SCOPED: "POST /v1/messages" allows only that verb, while a
+// bare "/v1/messages" allows any verb on that path (backwards compatible). Method
+// scoping is what lets a registry express "reads yes, writes no" — without it,
+// allowing GET /v1/numbers necessarily also allows DELETE /v1/numbers/{id}.
+func (a *AppEntry) allowed(method, path string) bool {
 	if len(a.allowSet) == 0 && len(a.allowPatterns) == 0 {
 		return false
 	}
-	if a.allowSet[path] {
+	method = strings.ToUpper(method)
+	if a.allowSet[costKey(method, path)] || a.allowSet[costKey("", path)] {
 		return true
 	}
 	segs := strings.Split(path, "/")
 	for _, pat := range a.allowPatterns {
-		if segmentsMatch(pat, segs) {
+		if pat.method != "" && pat.method != method {
+			continue
+		}
+		if segmentsMatch(pat.segs, segs) {
 			return true
 		}
 	}
 	return false
 }
 
+// allowPattern is a templated allow entry; method "" matches any verb.
+type allowPattern struct {
+	method string
+	segs   []string
+}
+
 // segmentsMatch reports whether request segments satisfy a templated pattern. A
 // "{name}" pattern segment matches any single non-empty segment; every other
 // segment must match literally. Lengths must be equal (no implicit wildcards).
@@ -307,13 +352,28 @@ func ParseRegistry(raw []byte, getenv func(string) string) (*Registry, error) {
 			return nil, fmt.Errorf("registry: app %s: env %s (master key) is empty", a.ID, a.KeyEnv)
 		}
 		a.injector = injectorFor(a.AuthStyle, a.AuthHeader, a.AuthScheme, a.AuthParam, a.AuthUser)
+		// A classic managed app that forwards with header-style auth but no header
+		// name builds an injector that sets an EMPTY header, which fails only when
+		// the first call is dialed (a 502). Catch it at load so a config typo is a
+		// boot failure, not a silent runtime outage on live traffic. Skipped for
+		// provisioned apps (they mint per-user keys, not a master-key header) and
+		// for empty-allow apps (which forward nothing).
+		if a.Provision == nil && len(a.Allow) > 0 &&
+			(a.AuthStyle == "" || a.AuthStyle == "header") && strings.TrimSpace(a.AuthHeader) == "" {
+			return nil, fmt.Errorf("registry: app %s: header auth needs a non-empty auth_header (set auth_style to query/basic if intended)", a.ID)
+		}
 		a.allowSet = map[string]bool{}
 		a.allowPatterns = nil
-		for _, p := range a.Allow {
+		a.allowSegs = nil
+		for _, entry := range a.Allow {
+			method, p := parseCostKey(entry) // "POST /v1/x" → ("POST","/v1/x"); "/v1/x" → ("","/v1/x")
+			method = strings.ToUpper(method)
 			if strings.Contains(p, "{") {
-				a.allowPatterns = append(a.allowPatterns, strings.Split(p, "/"))
+				segs := strings.Split(p, "/")
+				a.allowPatterns = append(a.allowPatterns, allowPattern{method: method, segs: segs})
+				a.allowSegs = append(a.allowSegs, segs)
 			} else {
-				a.allowSet[p] = true
+				a.allowSet[costKey(method, p)] = true
 			}
 		}
 		if a.CostField == "" {
@@ -330,6 +390,12 @@ func ParseRegistry(raw []byte, getenv func(string) string) (*Registry, error) {
 				return nil, err
 			}
 		}
+		if a.Tenancy != nil {
+			if err := validateTenancy(a); err != nil {
+				return nil, err
+			}
+			a.Tenancy.compile()
+		}
 		if a.Credit != nil {
 			if a.Provision != nil {
 				return nil, fmt.Errorf("registry: app %s: `credit` (HTTP budget) and `provision` (cloud) are mutually exclusive", a.ID)
@@ -342,10 +408,21 @@ func ParseRegistry(raw []byte, getenv func(string) string) (*Registry, error) {
 			if a.creditDefault < 0 {
 				a.creditDefault = 0
 			}
+			// Per-IP identity cap. Defaults CLOSED: omitted means the safe default,
+			// and only an explicit negative disables the guard, so switching it off is
+			// always a deliberate and visible choice rather than an oversight.
 			a.creditMaxPerIP = a.Credit.MaxIdentitiesPerIP
-			if a.creditMaxPerIP < 0 {
-				a.creditMaxPerIP = 0
+			switch {
+			case a.creditMaxPerIP == 0:
+				a.creditMaxPerIP = defaultMaxIdentitiesPerIP
+			case a.creditMaxPerIP < 0:
+				a.creditMaxPerIP = 0 // explicit opt-out: unlimited
 			}
+			// NOTE: mint_cooldown deliberately has NO default. Provision is touched on
+			// EVERY call (not just on re-seed), and its cooldown check rejects any
+			// touch inside the window — so a non-zero default would 429 every caller's
+			// second call. It stays opt-in for apps that really want re-mint churn
+			// control. The Sybil guard here is max_identities_per_ip, above.
 			a.creditMintCooldown = time.Duration(a.Credit.MintCooldownMs) * time.Millisecond
 			a.creditRespCost = a.Credit.CostSource == "response"
 			a.creditCostScale = a.Credit.CostScale
@@ -427,3 +504,17 @@ func resolveProvision(a *AppEntry, getenv func(string) string) error {
 	}
 	return nil
 }
+
+// AppsRequiringAccessKey lists apps whose registry entry demands an access key.
+// main uses it to refuse to boot when no keys are configured, so a config
+// mistake surfaces as a loud startup failure rather than a silent 401 wall.
+func (r *Registry) AppsRequiringAccessKey() []string {
+	var out []string
+	for id, a := range r.apps {
+		if a.RequireAccessKey {
+			out = append(out, id)
+		}
+	}
+	sort.Strings(out)
+	return out
+}
diff --git a/internal/broker/store.go b/internal/broker/store.go
index 803cefb..e43b6c9 100644
--- a/internal/broker/store.go
+++ b/internal/broker/store.go
@@ -2,6 +2,7 @@ package broker
 
 import (
 	"errors"
+	"strings"
 	"sync"
 	"time"
 )
@@ -97,13 +98,64 @@ type cell struct {
 // MemStore is the default in-memory Store + ProvisionStore (single instance,
 // non-durable). Used in dev and tests; prod uses SQLiteStore for durability.
 type MemStore struct {
-	mu   sync.Mutex
-	m    map[string]*cell
-	prov map[string]*ProvisionRecord // keyed "app|caller"
+	mu    sync.Mutex
+	m     map[string]*cell
+	prov  map[string]*ProvisionRecord // keyed "app|caller"
+	owner map[string]string           // keyed "app|rtype|rid" -> caller (tenancy ledger)
 }
 
 func NewMemStore() *MemStore {
-	return &MemStore{m: map[string]*cell{}, prov: map[string]*ProvisionRecord{}}
+	return &MemStore{m: map[string]*cell{}, prov: map[string]*ProvisionRecord{}, owner: map[string]string{}}
+}
+
+// ownKey keys the ownership ledger. rtype is part of the key so ids are never
+// confused across resource types (an agent id must not authorize a number).
+func ownKey(app, rtype, rid string) string { return app + "|" + rtype + "|" + rid }
+
+// Claim is first-writer-wins under the lock: a second caller claiming the same
+// resource gets ErrOwned rather than taking it over.
+func (s *MemStore) Claim(app, rtype, rid, caller string, _ time.Time) error {
+	if rid == "" || caller == "" {
+		return errors.New("broker: claim requires a resource id and caller")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	k := ownKey(app, rtype, rid)
+	if cur, ok := s.owner[k]; ok {
+		if cur == caller {
+			return nil // idempotent re-claim by the same owner
+		}
+		return ErrOwned
+	}
+	s.owner[k] = caller
+	return nil
+}
+
+func (s *MemStore) OwnerOf(app, rtype, rid string) (string, bool, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	c, ok := s.owner[ownKey(app, rtype, rid)]
+	return c, ok, nil
+}
+
+func (s *MemStore) OwnedSet(app, rtype, caller string) (map[string]bool, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := map[string]bool{}
+	prefix := app + "|" + rtype + "|"
+	for k, c := range s.owner {
+		if c == caller && strings.HasPrefix(k, prefix) {
+			out[strings.TrimPrefix(k, prefix)] = true
+		}
+	}
+	return out, nil
+}
+
+func (s *MemStore) Release(app, rtype, rid string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.owner, ownKey(app, rtype, rid))
+	return nil
 }
 
 func (s *MemStore) Provision(app, caller, ip string, seed, maxPerIP int, cooldown time.Duration, now time.Time) (ProvisionRecord, error) {
diff --git a/internal/broker/store_sqlite.go b/internal/broker/store_sqlite.go
index 4145a69..1583b75 100644
--- a/internal/broker/store_sqlite.go
+++ b/internal/broker/store_sqlite.go
@@ -2,6 +2,7 @@ package broker
 
 import (
 	"database/sql"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -70,6 +71,26 @@ func OpenSQLiteStore(path string) (*SQLiteStore, error) {
 		_ = db.Close()
 		return nil, fmt.Errorf("sqlite migrate provision.rot: %w", err)
 	}
+	// ownership: the tenancy ledger — which caller owns which upstream resource,
+	// for apps on a shared partner account. The PRIMARY KEY is what enforces
+	// first-writer-wins: a concurrent takeover attempt loses on the unique
+	// constraint rather than on an application-level check-then-write race.
+	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS ownership (
+		app     TEXT NOT NULL,
+		rtype   TEXT NOT NULL,
+		rid     TEXT NOT NULL,
+		caller  TEXT NOT NULL,
+		created INTEGER NOT NULL,
+		PRIMARY KEY (app, rtype, rid)
+	)`); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("sqlite migrate ownership: %w", err)
+	}
+	// Listing a caller's own resources is the hot path for response filtering.
+	if _, err := db.Exec(`CREATE INDEX IF NOT EXISTS ownership_owner ON ownership(app, rtype, caller)`); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("sqlite migrate ownership index: %w", err)
+	}
 	if _, err := db.Exec(`CREATE INDEX IF NOT EXISTS provision_app_ip ON provision(app, ip)`); err != nil {
 		_ = db.Close()
 		return nil, fmt.Errorf("sqlite migrate provision index: %w", err)
@@ -309,3 +330,68 @@ func (s *SQLiteStore) Snapshot() map[string]struct {
 	}
 	return out
 }
+
+// Claim records ownership of a resource, first-writer-wins.
+//
+// The INSERT relies on the (app, rtype, rid) PRIMARY KEY: two concurrent callers
+// racing to claim the same id cannot both succeed, so ownership can never be
+// silently transferred. A repeat claim by the SAME caller is idempotent (no
+// error); a claim by a DIFFERENT caller returns ErrOwned.
+func (s *SQLiteStore) Claim(app, rtype, rid, caller string, now time.Time) error {
+	if rid == "" || caller == "" {
+		return errors.New("broker: claim requires a resource id and caller")
+	}
+	res, err := s.db.Exec(
+		`INSERT INTO ownership (app, rtype, rid, caller, created) VALUES (?,?,?,?,?)
+		 ON CONFLICT(app, rtype, rid) DO NOTHING`,
+		app, rtype, rid, caller, now.Unix())
+	if err != nil {
+		return err
+	}
+	if n, err := res.RowsAffected(); err == nil && n > 0 {
+		return nil // we inserted: we are the owner
+	}
+	// Nothing inserted → a row already exists. It is ours only if the owner matches.
+	var cur string
+	if err := s.db.QueryRow(`SELECT caller FROM ownership WHERE app=? AND rtype=? AND rid=?`, app, rtype, rid).Scan(&cur); err != nil {
+		return err
+	}
+	if cur == caller {
+		return nil
+	}
+	return ErrOwned
+}
+
+func (s *SQLiteStore) OwnerOf(app, rtype, rid string) (string, bool, error) {
+	var caller string
+	err := s.db.QueryRow(`SELECT caller FROM ownership WHERE app=? AND rtype=? AND rid=?`, app, rtype, rid).Scan(&caller)
+	if err == sql.ErrNoRows {
+		return "", false, nil
+	}
+	if err != nil {
+		return "", false, err
+	}
+	return caller, true, nil
+}
+
+func (s *SQLiteStore) OwnedSet(app, rtype, caller string) (map[string]bool, error) {
+	rows, err := s.db.Query(`SELECT rid FROM ownership WHERE app=? AND rtype=? AND caller=?`, app, rtype, caller)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := map[string]bool{}
+	for rows.Next() {
+		var rid string
+		if err := rows.Scan(&rid); err != nil {
+			return nil, err
+		}
+		out[rid] = true
+	}
+	return out, rows.Err()
+}
+
+func (s *SQLiteStore) Release(app, rtype, rid string) error {
+	_, err := s.db.Exec(`DELETE FROM ownership WHERE app=? AND rtype=? AND rid=?`, app, rtype, rid)
+	return err
+}
diff --git a/internal/broker/tenancy.go b/internal/broker/tenancy.go
new file mode 100644
index 0000000..6e5ad71
--- /dev/null
+++ b/internal/broker/tenancy.go
@@ -0,0 +1,731 @@
+package broker
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Tenancy makes ONE shared partner account behave like N isolated tenants.
+//
+// Some managed apps cannot give each pilot user their own partner account — e.g.
+// io.pilot.agentphone, whose account is bound to a provider campaign that allows
+// number generation. Upstream, every pilot user is literally the same customer.
+// The partner therefore cannot enforce isolation, so the broker must.
+//
+// The model has exactly one rule: a caller may only touch a resource it OWNS.
+//   - Creating a resource claims it (owner = creator) — see claimFrom.
+//   - Referencing a resource (in the path, the query, or the body) is checked
+//     against the ledger BEFORE the request is forwarded — see EnforceRequest.
+//   - Listing resources filters the partner's account-wide answer down to the
+//     caller's own rows — see FilterResponse.
+//
+// Deny-by-default is load-bearing. A resource with no ledger row belongs to
+// nobody and is refused to everybody. That is what neutralises resources created
+// before this shipped (the shared test number/agent): rather than being up for
+// grabs, they become unreachable to every tenant.
+type Tenancy struct {
+	// ParamTypes maps a path-template param name to a resource type, e.g.
+	// "agent_id" -> "agent". Any {param} in an allow pattern that names a
+	// resource is ownership-checked.
+	ParamTypes map[string]string `json:"param_types"`
+	// BodyRefs maps a JSON body/query field to a resource type, e.g.
+	// "agent_id" -> "agent". Path checks alone are not sufficient: an operation
+	// often names the resource it acts THROUGH in the body (the sending agent on
+	// a send), so an unchecked body field is an unchecked resource.
+	BodyRefs map[string]string `json:"body_refs"`
+	// Create declares routes that mint a resource; on 2xx the new id is claimed.
+	Create []CreateRoute `json:"create"`
+	// List declares routes whose response must be filtered to owned resources.
+	List []ListRoute `json:"list"`
+	// Delete declares routes that destroy a resource; on 2xx the claim is dropped
+	// so a recycled id can be re-claimed by its next owner.
+	Delete []DeleteRoute `json:"delete"`
+	// Object declares routes whose response is a SUMMARY of the shared account
+	// rather than a list of rows.
+	Object []ObjectRoute `json:"object"`
+
+	createIdx []compiledRoute
+	listIdx   []compiledList
+	deleteIdx []compiledRoute
+	objectIdx []compiledObject
+}
+
+// ObjectRoute rewrites a response that describes the shared ACCOUNT instead of
+// returning rows.
+//
+// Filtering arrays is not sufficient on a shared account: a usage/summary
+// endpoint reports totals computed across every tenant. Those totals are a
+// side-channel — a caller who can see none of the resources can still read how
+// many exist and watch the numbers move as other tenants work.
+//
+// OwnedCounts replaces a count with the caller's OWN count from the ledger.
+// Redact removes a field outright, for aggregates that cannot be attributed to
+// one tenant at all (and so can only be wrong or leaky).
+type ObjectRoute struct {
+	Method      string            `json:"method"`
+	Path        string            `json:"path"`
+	OwnedCounts map[string]string `json:"owned_counts"` // dotted field -> resource type
+	Redact      []string          `json:"redact"`       // dotted fields to drop
+}
+
+type compiledObject struct {
+	method string
+	segs   []string
+	route  ObjectRoute
+}
+
+// CreateRoute: a 2xx on Method+Path means the caller created a resource of Type
+// whose id is at IDField in the response body.
+type CreateRoute struct {
+	Method  string `json:"method"`
+	Path    string `json:"path"`
+	Type    string `json:"type"`
+	IDField string `json:"id_field"` // dotted path into the response, e.g. "id" or "data.id"
+}
+
+// ListRoute: the response of Method+Path carries an array at Array whose
+// elements must be filtered to the caller's own.
+//
+// OwnerBy is why this is not simply "filter by element id": an INBOUND call or
+// message is a resource the tenant never created, so it has no ledger row of its
+// own — but it belongs to them because it is attached to a number they own. Each
+// OwnerBy is a (field, type) link; an element is kept if ANY link resolves to a
+// resource the caller owns. Elements with no resolvable link are DROPPED.
+type ListRoute struct {
+	Method  string      `json:"method"`
+	Path    string      `json:"path"`
+	Array   string      `json:"array"` // dotted path to the array, e.g. "data"
+	OwnerBy []OwnerLink `json:"owner_by"`
+	// ClaimAs, when set, claims each KEPT element's own id as this resource type.
+	//
+	// This is what makes derived resources reachable by id. A resource the partner
+	// creates (an inbound call) has no ledger row of its own; it is attributable
+	// only via its link to a resource the caller owns. Claiming it as it is listed
+	// is what lets a later fetch-by-id be ownership-checked at all — and every
+	// type named in param_types MUST be claimable somewhere, because a param with
+	// no resolvable type is a param that is never checked.
+	ClaimAs string `json:"claim_as"`
+	// ClaimIDField is where the element's own id lives (default "id").
+	ClaimIDField string `json:"claim_id_field"`
+	// CountFields are sibling fields that describe the SIZE of the array (e.g.
+	// "total"). They must be recomputed after filtering.
+	//
+	// Filtering the array alone is not enough: a count left at the partner's value
+	// still reports the whole shared account, so a caller who can see none of the
+	// rows can still read how many exist and watch that number move. Any field
+	// derived from the unfiltered set is a leak of the unfiltered set.
+	CountFields []string `json:"count_fields"`
+}
+
+// OwnerLink names a field on a list element and the resource type it points at.
+type OwnerLink struct {
+	Field string `json:"field"`
+	Type  string `json:"type"`
+}
+
+// DeleteRoute: a 2xx on Method+Path destroyed the resource named by Param.
+type DeleteRoute struct {
+	Method string `json:"method"`
+	Path   string `json:"path"`
+	Type   string `json:"type"`
+	Param  string `json:"param"` // which path param holds the id
+}
+
+type compiledRoute struct {
+	method string
+	segs   []string
+	route  any // CreateRoute | DeleteRoute
+}
+
+type compiledList struct {
+	method string
+	segs   []string
+	route  ListRoute
+}
+
+// compile pre-splits path templates so matching is allocation-light per request.
+func (t *Tenancy) compile() {
+	t.createIdx = nil
+	for _, c := range t.Create {
+		t.createIdx = append(t.createIdx, compiledRoute{strings.ToUpper(c.Method), strings.Split(c.Path, "/"), c})
+	}
+	t.listIdx = nil
+	for _, l := range t.List {
+		t.listIdx = append(t.listIdx, compiledList{strings.ToUpper(l.Method), strings.Split(l.Path, "/"), l})
+	}
+	t.deleteIdx = nil
+	for _, d := range t.Delete {
+		t.deleteIdx = append(t.deleteIdx, compiledRoute{strings.ToUpper(d.Method), strings.Split(d.Path, "/"), d})
+	}
+	t.objectIdx = nil
+	for _, o := range t.Object {
+		t.objectIdx = append(t.objectIdx, compiledObject{strings.ToUpper(o.Method), strings.Split(o.Path, "/"), o})
+	}
+}
+
+// FilterObject rewrites an account-summary response so it describes only the
+// caller. Returns (body, true) when it handled the route.
+//
+// Like FilterResponse it fails closed: if the body cannot be parsed or rewritten
+// it returns an error document rather than passing the partner's account-wide
+// answer through untouched.
+func (t *Tenancy) FilterObject(s OwnerStore, app, method, path string, respBody []byte, caller string) ([]byte, bool) {
+	if t == nil || s == nil {
+		return respBody, false
+	}
+	segs := strings.Split(path, "/")
+	for _, oi := range t.objectIdx {
+		o := oi.route
+		if oi.method != strings.ToUpper(method) || !segmentsMatch(oi.segs, segs) {
+			continue
+		}
+		var v any
+		dec := json.NewDecoder(strings.NewReader(string(respBody)))
+		dec.UseNumber()
+		if dec.Decode(&v) != nil {
+			return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+		}
+		for field, rtype := range o.OwnedCounts {
+			if _, ok := dig(v, field); !ok {
+				continue
+			}
+			owned, err := s.OwnedSet(app, rtype, caller)
+			if err != nil {
+				return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+			}
+			if !setDug(v, field, json.Number(strconv.Itoa(len(owned)))) {
+				return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+			}
+		}
+		for _, field := range o.Redact {
+			delDug(v, field)
+		}
+		out, err := json.Marshal(v)
+		if err != nil {
+			return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+		}
+		return out, true
+	}
+	return respBody, false
+}
+
+// delDug removes a dotted field from decoded JSON.
+func delDug(v any, dotted string) {
+	parts := strings.Split(dotted, ".")
+	cur := v
+	for i, p := range parts {
+		m, ok := cur.(map[string]any)
+		if !ok {
+			return
+		}
+		if i == len(parts)-1 {
+			delete(m, p)
+			return
+		}
+		cur = m[p]
+	}
+}
+
+// pathParams extracts {name} -> value by matching path against the app's allow
+// patterns. Reusing the allow patterns means a route can never be
+// ownership-checked under a template that does not also admit it.
+func pathParams(patterns [][]string, path string) map[string]string {
+	segs := strings.Split(path, "/")
+	out := map[string]string{}
+	for _, pat := range patterns {
+		if !segmentsMatch(pat, segs) {
+			continue
+		}
+		for i, p := range pat {
+			if len(p) > 2 && p[0] == '{' && p[len(p)-1] == '}' {
+				out[p[1:len(p)-1]] = segs[i]
+			}
+		}
+		// Keep scanning: several templates may match, and every {param} any of
+		// them binds must be checked. Being permissive here would let a caller
+		// pick the template with the fewest checks.
+	}
+	return out
+}
+
+// dig walks a dotted path ("data.id") through decoded JSON.
+func dig(v any, dotted string) (any, bool) {
+	cur := v
+	for _, part := range strings.Split(dotted, ".") {
+		m, ok := cur.(map[string]any)
+		if !ok {
+			return nil, false
+		}
+		cur, ok = m[part]
+		if !ok {
+			return nil, false
+		}
+	}
+	return cur, true
+}
+
+// asID renders a JSON scalar as a resource id. Only strings and numbers are
+// ids; anything else (object, array, bool, null) is not, and returns false.
+func asID(v any) (string, bool) {
+	switch x := v.(type) {
+	case string:
+		return x, x != ""
+	case json.Number:
+		return x.String(), true
+	case float64:
+		// Non-integral or huge floats are not ids; avoid a lossy render.
+		if x != float64(int64(x)) {
+			return "", false
+		}
+		return json.Number(strings.TrimSuffix(strings.TrimRight(jsonNum(x), "0"), ".")).String(), true
+	}
+	return "", false
+}
+
+func jsonNum(f float64) string {
+	b, _ := json.Marshal(f)
+	return string(b)
+}
+
+// refusal is why a request was denied. It is deliberately opaque to the caller:
+// EnforceRequest's 404 must not distinguish "not yours" from "does not exist",
+// or the broker becomes an oracle for enumerating other tenants' resource ids.
+type refusal struct {
+	Type string
+	ID   string
+}
+
+// EnforceRequest is the authorization gate: it runs BEFORE the request is
+// forwarded, so a rejected write never reaches the partner and never has a side
+// effect. It collects every resource reference in the path, the query string,
+// and the JSON body, and requires the caller to own each one.
+//
+// It FAILS CLOSED: an unparseable body, an unknown id, or a store error all deny.
+func (t *Tenancy) EnforceRequest(s OwnerStore, app string, allowPatterns [][]string, method, path, rawQuery string, body []byte, caller string) (*refusal, bool) {
+	if t == nil {
+		return nil, true
+	}
+	if s == nil || caller == "" {
+		return &refusal{}, false // no ledger or no identity → deny
+	}
+
+	// 1. Path params: /v1/agents/{agent_id}/... — the id is in the URL.
+	for name, val := range pathParams(allowPatterns, path) {
+		rtype, ok := t.ParamTypes[name]
+		if !ok {
+			continue // param names no resource (e.g. a filter) → nothing to own
+		}
+		if !Owns(s, app, rtype, val, caller) {
+			return &refusal{Type: rtype, ID: val}, false
+		}
+	}
+
+	// 2. Query params: ?agent_id=... — a filter can also be a reference.
+	if rawQuery != "" {
+		q, err := url.ParseQuery(rawQuery)
+		if err != nil {
+			return &refusal{}, false // unparseable query → deny rather than skip
+		}
+		for field, vals := range q {
+			rtype, ok := t.BodyRefs[field]
+			if !ok {
+				continue
+			}
+			for _, v := range vals {
+				if v == "" {
+					continue
+				}
+				if !Owns(s, app, rtype, v, caller) {
+					return &refusal{Type: rtype, ID: v}, false
+				}
+			}
+		}
+	}
+
+	// 3. Body refs: {"agent_id": "..."}. An operation that names the resource it
+	//    acts through in the body must be checked here, or path-level isolation
+	//    is decorative.
+	if len(body) > 0 && len(t.BodyRefs) > 0 {
+		// PARSER DIFFERENTIAL. The broker validates the body with Go's decoder but
+		// forwards the RAW bytes, so the partner re-parses them with a different
+		// parser. Go keeps the LAST duplicate key; a parser that keeps the FIRST
+		// would act on a value we never checked:
+		//
+		//   {"agent_id":"<victim's>", "agent_id":"<mine>"}
+		//
+		// We would validate "<mine>" and approve; the partner would send from
+		// "<victim's>". Duplicate keys have no legitimate use here, so a body
+		// containing any is refused outright rather than reasoned about.
+		if hasDuplicateKeys(body) {
+			return &refusal{}, false
+		}
+		var v any
+		dec := json.NewDecoder(strings.NewReader(string(body)))
+		dec.UseNumber() // keep big ids exact; float64 would corrupt them
+		if err := dec.Decode(&v); err != nil {
+			// A body we cannot parse is a body we cannot check. If the app declares
+			// body refs at all, refuse rather than forward it unchecked.
+			return &refusal{}, false
+		}
+		// Trailing content after the first JSON value ("{...}{...}") is another way
+		// two parsers can disagree about what the body says. Refuse it.
+		if dec.More() {
+			return &refusal{}, false
+		}
+		if ref, ok := t.checkRefs(s, app, v, caller, 0); !ok {
+			return ref, false
+		}
+	}
+	return nil, true
+}
+
+// maxRefDepth bounds recursion into an attacker-supplied body. A deeply nested
+// body must not become a stack-exhaustion lever.
+const maxRefDepth = 24
+
+// checkRefs walks the decoded body and ownership-checks every field named in
+// BodyRefs, at ANY depth. Depth matters: a ref nested inside an object must be
+// checked too, or wrapping it becomes the bypass.
+func (t *Tenancy) checkRefs(s OwnerStore, app string, v any, caller string, depth int) (*refusal, bool) {
+	if depth > maxRefDepth {
+		return &refusal{}, false // too deep to verify → deny
+	}
+	switch x := v.(type) {
+	case map[string]any:
+		for k, val := range x {
+			if rtype, ok := t.BodyRefs[k]; ok {
+				if id, isID := asID(val); isID {
+					if !Owns(s, app, rtype, id, caller) {
+						return &refusal{Type: rtype, ID: id}, false
+					}
+				} else if val != nil {
+					// The field names a resource but is not a scalar id (e.g. an
+					// object or array). We cannot verify it → deny.
+					return &refusal{Type: rtype}, false
+				}
+			}
+			if ref, ok := t.checkRefs(s, app, val, caller, depth+1); !ok {
+				return ref, false
+			}
+		}
+	case []any:
+		for _, val := range x {
+			if ref, ok := t.checkRefs(s, app, val, caller, depth+1); !ok {
+				return ref, false
+			}
+		}
+	}
+	return nil, true
+}
+
+// ClaimFrom records ownership after a successful create. It runs only on 2xx, so
+// a failed create never claims an id.
+func (t *Tenancy) ClaimFrom(s OwnerStore, app, method, path string, respBody []byte, caller string, now time.Time) {
+	if t == nil || s == nil {
+		return
+	}
+	segs := strings.Split(path, "/")
+	for _, cr := range t.createIdx {
+		c := cr.route.(CreateRoute)
+		if cr.method != strings.ToUpper(method) || !segmentsMatch(cr.segs, segs) {
+			continue
+		}
+		var v any
+		dec := json.NewDecoder(strings.NewReader(string(respBody)))
+		dec.UseNumber()
+		if dec.Decode(&v) != nil {
+			continue
+		}
+		raw, ok := dig(v, c.IDField)
+		if !ok {
+			continue
+		}
+		if id, isID := asID(raw); isID {
+			_ = s.Claim(app, c.Type, id, caller, now)
+		}
+	}
+}
+
+// ReleaseFrom drops a claim after a successful delete, so a partner-recycled id
+// can be claimed by whoever gets it next instead of staying bound to its old
+// owner forever.
+func (t *Tenancy) ReleaseFrom(s OwnerStore, app string, allowPatterns [][]string, method, path string) {
+	if t == nil || s == nil {
+		return
+	}
+	segs := strings.Split(path, "/")
+	params := pathParams(allowPatterns, path)
+	for _, dr := range t.deleteIdx {
+		d := dr.route.(DeleteRoute)
+		if dr.method != strings.ToUpper(method) || !segmentsMatch(dr.segs, segs) {
+			continue
+		}
+		if id := params[d.Param]; id != "" {
+			_ = s.Release(app, d.Type, id)
+		}
+	}
+}
+
+// FilterResponse rewrites a list response to contain only the caller's own
+// resources. This is what makes "pilot users must not see numbers other than
+// their own" true: the partner answers for the whole shared account, and the
+// broker strips it to the caller's rows before it ever reaches them.
+//
+// It also CLAIMS kept elements (lazy claim). Inbound calls/messages are created
+// by the partner, not the tenant, so they have no row yet; claiming them when
+// they are provably linked to an owned number is what lets the tenant then fetch
+// them by id.
+//
+// On any doubt it returns an EMPTY array rather than the unfiltered body:
+// leaking another tenant's rows is worse than showing none.
+func (t *Tenancy) FilterResponse(s OwnerStore, app, method, path string, respBody []byte, caller string, now time.Time) ([]byte, bool) {
+	if t == nil || s == nil {
+		return respBody, false
+	}
+	segs := strings.Split(path, "/")
+	for _, li := range t.listIdx {
+		l := li.route
+		if li.method != strings.ToUpper(method) || !segmentsMatch(li.segs, segs) {
+			continue
+		}
+		var v any
+		dec := json.NewDecoder(strings.NewReader(string(respBody)))
+		dec.UseNumber()
+		if dec.Decode(&v) != nil {
+			// Cannot parse → cannot filter → do not pass the partner's raw
+			// account-wide answer through.
+			return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+		}
+		arrRaw, ok := dig(v, l.Array)
+		if !ok {
+			return respBody, false // no array here; nothing to filter
+		}
+		arr, ok := arrRaw.([]any)
+		if !ok {
+			return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+		}
+		kept := make([]any, 0, len(arr))
+		for _, el := range arr {
+			if t.keepElement(s, app, el, caller, l.OwnerBy, l.ClaimAs, l.ClaimIDField, now) {
+				kept = append(kept, el)
+			}
+		}
+		if !setDug(v, l.Array, kept) {
+			return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+		}
+		// Recompute any count that described the unfiltered set. Note that
+		// pagination flags (hasMore) are deliberately left alone: they describe the
+		// partner's underlying paging, and forcing them false would stop a client
+		// paging before it reached its OWN rows on a later page.
+		for _, cf := range l.CountFields {
+			if _, ok := dig(v, cf); ok {
+				_ = setDug(v, cf, json.Number(strconv.Itoa(len(kept))))
+			}
+		}
+		out, err := json.Marshal(v)
+		if err != nil {
+			return []byte(`{"error":"tenancy: unfilterable upstream response"}`), true
+		}
+		return out, true
+	}
+	return respBody, false
+}
+
+// keepElement decides if a list element belongs to caller, and lazily claims it.
+func (t *Tenancy) keepElement(s OwnerStore, app string, el any, caller string, links []OwnerLink, claimAs, claimIDField string, now time.Time) bool {
+	m, ok := el.(map[string]any)
+	if !ok {
+		return false // not an object → cannot attribute → drop
+	}
+	for _, link := range links {
+		raw, ok := m[link.Field]
+		if !ok {
+			continue
+		}
+		id, isID := asID(raw)
+		if !isID {
+			continue
+		}
+		if Owns(s, app, link.Type, id, caller) {
+			// Provably the caller's. Claim the element itself so a later
+			// fetch-by-id can be ownership-checked at all.
+			if claimAs != "" {
+				field := claimIDField
+				if field == "" {
+					field = "id"
+				}
+				if selfID, ok := asID(m[field]); ok && selfID != "" {
+					_ = s.Claim(app, claimAs, selfID, caller, now)
+				}
+			}
+			return true
+		}
+	}
+	return false
+}
+
+// setDug writes a value at a dotted path in decoded JSON.
+func setDug(v any, dotted string, val any) bool {
+	parts := strings.Split(dotted, ".")
+	cur := v
+	for i, p := range parts {
+		m, ok := cur.(map[string]any)
+		if !ok {
+			return false
+		}
+		if i == len(parts)-1 {
+			m[p] = val
+			return true
+		}
+		cur, ok = m[p]
+		if !ok {
+			return false
+		}
+	}
+	return false
+}
+
+// validateTenancy rejects a tenancy block that would not actually isolate.
+//
+// This runs at registry load and FAILS THE BOOT rather than warning. A tenancy
+// spec is a security control: a typo that silently disables a check is the whole
+// bug class we are fixing, so a broken spec must never start serving traffic.
+func validateTenancy(a *AppEntry) error {
+	t := a.Tenancy
+	if len(t.ParamTypes) == 0 && len(t.BodyRefs) == 0 {
+		return fmt.Errorf("registry: app %s: tenancy declares no param_types or body_refs — it would enforce nothing", a.ID)
+	}
+	types := map[string]bool{}
+	for _, ty := range t.ParamTypes {
+		types[ty] = true
+	}
+	for _, ty := range t.BodyRefs {
+		types[ty] = true
+	}
+	for _, c := range t.Create {
+		if c.Type == "" || c.Path == "" || c.Method == "" || c.IDField == "" {
+			return fmt.Errorf("registry: app %s: tenancy.create needs method, path, type and id_field", a.ID)
+		}
+		types[c.Type] = true
+	}
+	for _, d := range t.Delete {
+		if d.Type == "" || d.Path == "" || d.Method == "" || d.Param == "" {
+			return fmt.Errorf("registry: app %s: tenancy.delete needs method, path, type and param", a.ID)
+		}
+	}
+	for _, l := range t.List {
+		if l.Path == "" || l.Array == "" || len(l.OwnerBy) == 0 {
+			return fmt.Errorf("registry: app %s: tenancy.list %q needs array and at least one owner_by link (an unfiltered list leaks every tenant)", a.ID, l.Path)
+		}
+		for _, link := range l.OwnerBy {
+			if link.Field == "" || link.Type == "" {
+				return fmt.Errorf("registry: app %s: tenancy.list %q has an owner_by with an empty field/type", a.ID, l.Path)
+			}
+			if !types[link.Type] {
+				return fmt.Errorf("registry: app %s: tenancy.list %q links to resource type %q that nothing ever claims — every element would be dropped", a.ID, l.Path, link.Type)
+			}
+		}
+	}
+	// Every claimable type should be reachable: a type referenced by params/body
+	// but never created can never be owned, which would deny the app entirely.
+	created := map[string]bool{}
+	for _, c := range t.Create {
+		created[c.Type] = true
+	}
+	for _, ty := range t.ParamTypes {
+		if !created[ty] && !t.lazyClaimed(ty) {
+			return fmt.Errorf("registry: app %s: tenancy resource %q is referenced but never created or list-claimed — nobody could ever own it", a.ID, ty)
+		}
+	}
+	return nil
+}
+
+// lazyClaimed reports whether a type can be claimed from a list response (the
+// inbound-resource path), so it need not have an explicit create route.
+func (t *Tenancy) lazyClaimed(rtype string) bool {
+	for _, l := range t.List {
+		if l.ClaimAs == rtype {
+			return true
+		}
+	}
+	return false
+}
+
+// hasDuplicateKeys reports whether any JSON object in body repeats a key.
+//
+// This exists purely to kill parser-differential bypasses (see EnforceRequest).
+// It walks the TOKEN stream, not the decoded value: by the time a value is
+// decoded the duplicate is already gone, because Go silently kept the last one.
+//
+// It fails closed — a body that is malformed or unbalanced counts as duplicated.
+// If we cannot prove the body says exactly one thing, we do not forward it.
+func hasDuplicateKeys(body []byte) bool {
+	type frame struct {
+		keys      map[string]bool // nil for arrays
+		expectVal bool            // an object key was read; the next token is its value
+	}
+	dec := json.NewDecoder(strings.NewReader(string(body)))
+	dec.UseNumber()
+	var stack []*frame
+
+	// valueConsumed marks that the enclosing object's pending key now has its
+	// value. It must run for EVERY kind of value — including a nested object or
+	// array — or the key that follows the nested value is mistaken for a value
+	// and its duplicate goes unnoticed.
+	valueConsumed := func() {
+		if n := len(stack); n > 0 && stack[n-1].keys != nil {
+			stack[n-1].expectVal = false
+		}
+	}
+
+	for {
+		tok, err := dec.Token()
+		if err != nil {
+			// Clean end only counts if every container closed.
+			if errors.Is(err, io.EOF) {
+				return len(stack) != 0
+			}
+			return true // malformed → ambiguous → refuse
+		}
+		switch v := tok.(type) {
+		case json.Delim:
+			switch v {
+			case '{':
+				valueConsumed()
+				stack = append(stack, &frame{keys: map[string]bool{}})
+			case '[':
+				valueConsumed()
+				stack = append(stack, &frame{})
+			case '}', ']':
+				if len(stack) == 0 {
+					return true // unbalanced
+				}
+				stack = stack[:len(stack)-1]
+			}
+		case string:
+			n := len(stack)
+			if n == 0 {
+				continue // top-level scalar
+			}
+			top := stack[n-1]
+			if top.keys != nil && !top.expectVal {
+				// Directly inside an object and not awaiting a value → this is a KEY.
+				if top.keys[v] {
+					return true
+				}
+				top.keys[v] = true
+				top.expectVal = true
+				continue
+			}
+			valueConsumed()
+		default:
+			valueConsumed()
+		}
+	}
+}
diff --git a/internal/broker/zz_allow_pattern_test.go b/internal/broker/zz_allow_pattern_test.go
index 6d075ee..1562953 100644
--- a/internal/broker/zz_allow_pattern_test.go
+++ b/internal/broker/zz_allow_pattern_test.go
@@ -31,7 +31,7 @@ func TestAllowPatternMatching(t *testing.T) {
 		{"/v1/usage/daily", false},           // longer than the exact entry
 	}
 	for _, c := range cases {
-		if got := app.allowed(c.path); got != c.want {
+		if got := app.allowed("GET", c.path); got != c.want {
 			t.Errorf("allowed(%q) = %v, want %v", c.path, got, c.want)
 		}
 	}
@@ -44,7 +44,7 @@ func TestAllowEmptyDeniesAll(t *testing.T) {
 	if err != nil {
 		t.Fatalf("ParseRegistry: %v", err)
 	}
-	if reg.Get("io.pilot.y").allowed("/anything") {
+	if reg.Get("io.pilot.y").allowed("GET", "/anything") {
 		t.Error("empty allow-list must deny all")
 	}
 }
diff --git a/internal/broker/zz_coverage_test.go b/internal/broker/zz_coverage_test.go
new file mode 100644
index 0000000..8ee3c8a
--- /dev/null
+++ b/internal/broker/zz_coverage_test.go
@@ -0,0 +1,236 @@
+package broker
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func okUpstream() http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(200)
+		_, _ = w.Write([]byte(`{"ok":true}`))
+	})
+}
+
+// --- access keys ---------------------------------------------------------
+
+func TestAccessKeys_ParseAndCheck(t *testing.T) {
+	ak := NewAccessKeys([]string{"alpha:key-one", "  ", "bare-two", "beta: key-three "})
+	if ak.Len() != 3 {
+		t.Fatalf("Len = %d, want 3 (blank entry ignored)", ak.Len())
+	}
+	// header form
+	if lbl, ok := ak.Check(hdr(map[string]string{AccessKeyHeader: "key-one"})); !ok || lbl != "alpha" {
+		t.Errorf("valid header key: ok=%v label=%q", ok, lbl)
+	}
+	// bearer form
+	if lbl, ok := ak.Check(hdr(map[string]string{"Authorization": "Bearer bare-two"})); !ok || lbl != "" {
+		t.Errorf("valid bearer key: ok=%v label=%q", ok, lbl)
+	}
+	// label:key with spaces trimmed
+	if _, ok := ak.Check(hdr(map[string]string{AccessKeyHeader: "key-three"})); !ok {
+		t.Error("trimmed label:key should verify")
+	}
+	// wrong + missing → deny
+	if _, ok := ak.Check(hdr(map[string]string{AccessKeyHeader: "nope"})); ok {
+		t.Error("wrong key must not verify")
+	}
+	if _, ok := ak.Check(hdr(map[string]string{})); ok {
+		t.Error("missing key must not verify")
+	}
+}
+
+func TestAccessKeys_NoKeysFailsClosed(t *testing.T) {
+	var nilAK *AccessKeys
+	if _, ok := nilAK.Check(hdr(map[string]string{AccessKeyHeader: "x"})); ok {
+		t.Error("nil AccessKeys must deny")
+	}
+	empty := NewAccessKeys([]string{"", "  "})
+	if empty.Len() != 0 {
+		t.Fatalf("empty Len = %d", empty.Len())
+	}
+	if _, ok := empty.Check(hdr(map[string]string{AccessKeyHeader: "x"})); ok {
+		t.Error("no keys configured must authorize nothing")
+	}
+}
+
+// TestRequireAccessKey_GatesApp: an app with require_access_key returns 401
+// without a valid key and forwards with one. Also covers AppsRequiringAccessKey.
+func TestRequireAccessKey_GatesApp(t *testing.T) {
+	up := httptest.NewServer(okUpstream())
+	t.Cleanup(up.Close)
+	reg, err := ParseRegistry([]byte(`[{
+		"id":"io.pilot.gated","upstream":"`+up.URL+`","key_env":"G_KEY",
+		"auth_header":"Authorization","auth_scheme":"Bearer","allow":["/v1/x"],
+		"require_access_key":true
+	}]`), func(string) string { return "master" })
+	if err != nil {
+		t.Fatalf("ParseRegistry: %v", err)
+	}
+	if got := reg.AppsRequiringAccessKey(); len(got) != 1 || got[0] != "io.pilot.gated" {
+		t.Fatalf("AppsRequiringAccessKey = %v", got)
+	}
+	b := New(reg, NewMemStore())
+	b.Verify = VerifyConfig{Window: time.Hour}
+	b.AccessKeys = NewAccessKeys([]string{"secret-key"})
+	_, priv := newKey(t)
+
+	// no key → 401
+	rec := httptest.NewRecorder()
+	b.ServeHTTP(rec, signedReq(t, priv, "POST", "/io.pilot.gated/v1/x", []byte(`{}`), time.Now()))
+	if rec.Code != 401 {
+		t.Errorf("no access key: status %d, want 401", rec.Code)
+	}
+	// with key → forwarded (200)
+	rec = httptest.NewRecorder()
+	req := signedReq(t, priv, "POST", "/io.pilot.gated/v1/x", []byte(`{}`), time.Now())
+	req.Header.Set(AccessKeyHeader, "secret-key")
+	b.ServeHTTP(rec, req)
+	if rec.Code != 200 {
+		t.Errorf("with access key: status %d, want 200", rec.Code)
+	}
+}
+
+// --- SQLite ownership ledger ---------------------------------------------
+
+func TestSQLiteOwnership_ClaimFirstWriterWins(t *testing.T) {
+	s, err := OpenSQLiteStore(":memory:")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	now := time.Unix(1_800_000_000, 0)
+
+	if err := s.Claim("app", "number", "n1", "alice", now); err != nil {
+		t.Fatalf("alice claim: %v", err)
+	}
+	if err := s.Claim("app", "number", "n1", "alice", now); err != nil {
+		t.Errorf("idempotent re-claim: %v", err)
+	}
+	if err := s.Claim("app", "number", "n1", "mallory", now); err != ErrOwned {
+		t.Errorf("takeover: %v, want ErrOwned", err)
+	}
+	owner, found, err := s.OwnerOf("app", "number", "n1")
+	if err != nil || !found || owner != "alice" {
+		t.Errorf("OwnerOf = %q,%v,%v", owner, found, err)
+	}
+	if _, found, _ := s.OwnerOf("app", "number", "ghost"); found {
+		t.Error("unknown resource must not be found")
+	}
+	// OwnedSet + Release
+	_ = s.Claim("app", "number", "n2", "alice", now)
+	_ = s.Claim("app", "agent", "a1", "alice", now)
+	set, err := s.OwnedSet("app", "number", "alice")
+	if err != nil || len(set) != 2 || !set["n1"] || !set["n2"] {
+		t.Errorf("OwnedSet(number) = %v (err %v)", set, err)
+	}
+	if err := s.Release("app", "number", "n1"); err != nil {
+		t.Fatal(err)
+	}
+	if _, found, _ := s.OwnerOf("app", "number", "n1"); found {
+		t.Error("released resource still owned")
+	}
+	// Owns() helper against SQLite
+	if !Owns(s, "app", "agent", "a1", "alice") || Owns(s, "app", "agent", "a1", "mallory") {
+		t.Error("Owns disagrees with ledger")
+	}
+}
+
+// TestSQLiteCredit_DebitRefundSettle exercises the SQLite credit ledger paths.
+func TestSQLiteCredit_DebitRefundSettle(t *testing.T) {
+	s, err := OpenSQLiteStore(":memory:")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	now := time.Unix(1_800_000_000, 0)
+
+	if _, err := s.Provision("app", "c", "1.2.3.4", 100, 0, 0, now); err != nil {
+		t.Fatal(err)
+	}
+	if bal, err := s.Credit("app", "c"); err != nil || bal != 100 {
+		t.Fatalf("seed Credit = %d,%v", bal, err)
+	}
+	ok, rem, err := s.Debit("app", "c", 40)
+	if err != nil || !ok || rem != 60 {
+		t.Fatalf("Debit 40 = %v,%d,%v", ok, rem, err)
+	}
+	s.Refund("app", "c", 10) // back to 70
+	if bal, _ := s.Credit("app", "c"); bal != 70 {
+		t.Fatalf("after refund = %d, want 70", bal)
+	}
+	if rem, err := s.Settle("app", "c", 1000); err != nil || rem != 0 {
+		t.Fatalf("Settle overshoot = %d,%v, want clamp to 0", rem, err)
+	}
+	// over-debit refused
+	if ok, _, _ := s.Debit("app", "c", 5); ok {
+		t.Error("debit past zero must fail")
+	}
+	rec, found, err := s.Get("app", "c")
+	if err != nil || !found || rec.Credits != 0 {
+		t.Errorf("Get = %+v,%v,%v", rec, found, err)
+	}
+}
+
+// --- tenancy account-summary redaction -----------------------------------
+
+// TestFilterObject_RecomputesAndRedacts: an account-summary response has its
+// count replaced by the caller's own count and its unattributable fields dropped.
+func TestFilterObject_RecomputesAndRedacts(t *testing.T) {
+	s := NewMemStore()
+	now := time.Unix(1_800_000_000, 0)
+	_ = s.Claim("app", "number", "n1", "alice", now)
+	_ = s.Claim("app", "number", "n2", "alice", now)
+
+	tn := &Tenancy{
+		ParamTypes: map[string]string{"number_id": "number"},
+		Create:     []CreateRoute{{Method: "POST", Path: "/v1/numbers", Type: "number", IDField: "id"}},
+		Object: []ObjectRoute{{
+			Method: "GET", Path: "/v1/usage",
+			OwnedCounts: map[string]string{"numbers.used": "number"},
+			Redact:      []string{"stats", "numbers.remaining"},
+		}},
+	}
+	if err := validateTenancy(&AppEntry{ID: "app", Tenancy: tn}); err != nil {
+		t.Fatalf("validateTenancy: %v", err)
+	}
+	tn.compile()
+
+	// partner reports the whole account: 9 numbers used, remaining 991, stats blob
+	raw := []byte(`{"numbers":{"used":9,"remaining":991},"stats":{"total":123}}`)
+	out, did := tn.FilterObject(s, "app", "GET", "/v1/usage", raw, "alice")
+	if !did {
+		t.Fatal("FilterObject did not handle the route")
+	}
+	var got map[string]any
+	if err := json.Unmarshal(out, &got); err != nil {
+		t.Fatalf("unmarshal %s: %v", out, err)
+	}
+	nums := got["numbers"].(map[string]any)
+	if fmtNum(nums["used"]) != "2" {
+		t.Errorf("numbers.used = %v, want 2 (alice's own)", nums["used"])
+	}
+	if _, ok := nums["remaining"]; ok {
+		t.Error("numbers.remaining should be redacted (derived from account-wide used)")
+	}
+	if _, ok := got["stats"]; ok {
+		t.Error("stats should be redacted")
+	}
+	// a route it doesn't cover passes through untouched
+	if _, did := tn.FilterObject(s, "app", "GET", "/v1/other", raw, "alice"); did {
+		t.Error("uncovered route should not be handled")
+	}
+}
+
+func fmtNum(v any) string {
+	switch x := v.(type) {
+	case json.Number:
+		return x.String()
+	case float64:
+		return json.Number(jsonNum(x)).String()
+	}
+	return ""
+}
diff --git a/internal/broker/zz_credit_respcost_test.go b/internal/broker/zz_credit_respcost_test.go
index 118f2a5..558a8ca 100644
--- a/internal/broker/zz_credit_respcost_test.go
+++ b/internal/broker/zz_credit_respcost_test.go
@@ -1,6 +1,7 @@
 package broker
 
 import (
+	"crypto/ed25519"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
@@ -220,6 +221,10 @@ func TestCreditBalance_PerUserNotAccount(t *testing.T) {
 // pilot identity from the same source IP cannot claim a fresh budget — this is the
 // anti-Sybil guard that stops farming new $5 grants after depletion. Both callers
 // share httptest's default RemoteAddr (192.0.2.1).
+// TestCreditPath_PerIPIdentityCap pins the per-IP GRANT cap contract: one source
+// IP may mint at most N funded identities. The (N+1)th is not refused — it is
+// recorded with a ZERO grant, so it can still read but cannot spend. This is the
+// control that stops budget-farming while keeping a shared-NAT newcomer usable.
 func TestCreditPath_PerIPIdentityCap(t *testing.T) {
 	now := time.Unix(1_800_000_000, 0)
 	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -227,10 +232,12 @@ func TestCreditPath_PerIPIdentityCap(t *testing.T) {
 		_, _ = w.Write([]byte(`{"ok":true}`))
 	}))
 	t.Cleanup(up.Close)
+	// "/v1/read" is free; "/v1/act" costs $1. Cap = 1 funded identity per IP.
 	reg, err := ParseRegistry([]byte(`[{
 		"id":"io.pilot.orthogonal","upstream":"`+up.URL+`","key_env":"ORTH_KEY",
-		"auth_header":"Authorization","auth_scheme":"Bearer","allow":["/v1/search"],
-		"credit":{"seed_credits":5000000,"default_cost":0,"max_identities_per_ip":1}
+		"auth_header":"Authorization","auth_scheme":"Bearer","allow":["/v1/read","/v1/act"],
+		"credit":{"seed_credits":5000000,"default_cost":0,
+		          "cost_credits":{"POST /v1/act":1000000},"max_identities_per_ip":1}
 	}]`), func(string) string { return "MASTERKEY" })
 	if err != nil {
 		t.Fatalf("ParseRegistry: %v", err)
@@ -238,17 +245,26 @@ func TestCreditPath_PerIPIdentityCap(t *testing.T) {
 	b := New(reg, NewMemStore())
 	b.Verify = VerifyConfig{Now: fixedClock(now)}
 
+	call := func(priv ed25519.PrivateKey, path string) int {
+		rec := httptest.NewRecorder()
+		b.ServeHTTP(rec, signedReq(t, priv, "POST", "/io.pilot.orthogonal"+path, []byte(`{}`), now))
+		return rec.Code
+	}
+
+	// Identity 1 is funded: it can spend.
 	_, priv1 := newKey(t)
-	rec1 := httptest.NewRecorder()
-	b.ServeHTTP(rec1, signedReq(t, priv1, "POST", "/io.pilot.orthogonal/v1/search", []byte(`{}`), now))
-	if rec1.Code != 200 {
-		t.Fatalf("first identity: status %d, want 200", rec1.Code)
+	if got := call(priv1, "/v1/act"); got != 200 {
+		t.Fatalf("funded identity spend: status %d, want 200", got)
 	}
-	// A different key = a different pilot identity from the same IP → capped.
+
+	// Identity 2 on the SAME IP is over the cap → zero grant.
 	_, priv2 := newKey(t)
-	rec2 := httptest.NewRecorder()
-	b.ServeHTTP(rec2, signedReq(t, priv2, "POST", "/io.pilot.orthogonal/v1/search", []byte(`{}`), now))
-	if rec2.Code != http.StatusTooManyRequests {
-		t.Fatalf("second identity same IP: status %d, want 429 (IP cap)", rec2.Code)
+	// It is NOT locked out: a free read still works.
+	if got := call(priv2, "/v1/read"); got != 200 {
+		t.Errorf("capped identity read: status %d, want 200 (must not be hard-blocked)", got)
+	}
+	// But it has no budget, so a priced call is refused with 402 — no free money.
+	if got := call(priv2, "/v1/act"); got != http.StatusPaymentRequired {
+		t.Errorf("capped identity spend: status %d, want 402 (no grant)", got)
 	}
 }
diff --git a/internal/broker/zz_dupkey_test.go b/internal/broker/zz_dupkey_test.go
new file mode 100644
index 0000000..d93a03a
--- /dev/null
+++ b/internal/broker/zz_dupkey_test.go
@@ -0,0 +1,35 @@
+package broker
+
+import "testing"
+
+// TestHasDuplicateKeys guards the parser-differential defense. Go's decoder keeps
+// the LAST duplicate key; a partner parser keeping the FIRST would act on a value
+// the broker never authorized. These cases pin that door shut.
+func TestHasDuplicateKeys(t *testing.T) {
+	cases := []struct {
+		name string
+		body string
+		want bool
+	}{
+		{"clean object", `{"agent_id":"a1","to":"+1555"}`, false},
+		{"repeated key, victim first", `{"agent_id":"victim","agent_id":"mine"}`, true},
+		{"dup key reversed", `{"agent_id":"mine","agent_id":"victim"}`, true},
+		{"dup nested", `{"m":{"agent_id":"victim","agent_id":"mine"}}`, true},
+		{"dup in array element", `{"b":[{"agent_id":"v","agent_id":"m"}]}`, true},
+		{"same key different objects is fine", `{"a":{"id":"1"},"b":{"id":"2"}}`, false},
+		{"repeated key across array elements is fine", `[{"id":"1"},{"id":"2"}]`, false},
+		{"value equal to a key name is fine", `{"a":"b","b":"c"}`, false},
+		{"string value matching sibling key", `{"agent_id":"agent_id","x":"y"}`, false},
+		{"numbers and nulls", `{"a":1,"b":null,"c":true,"d":[1,2]}`, false},
+		{"deep nesting clean", `{"a":{"b":{"c":{"d":{"id":"x"}}}}}`, false},
+		{"malformed → treated as ambiguous", `{"a":`, true},
+		{"empty object", `{}`, false},
+		{"array of scalars", `["a","a"]`, false},
+		{"dup after nested object closes", `{"n":{"x":1},"k":1,"k":2}`, true},
+	}
+	for _, c := range cases {
+		if got := hasDuplicateKeys([]byte(c.body)); got != c.want {
+			t.Errorf("%s: hasDuplicateKeys(%s) = %v, want %v", c.name, c.body, got, c.want)
+		}
+	}
+}
diff --git a/internal/broker/zz_tenancy_test.go b/internal/broker/zz_tenancy_test.go
new file mode 100644
index 0000000..8a20b16
--- /dev/null
+++ b/internal/broker/zz_tenancy_test.go
@@ -0,0 +1,414 @@
+package broker
+
+import (
+	"crypto/ed25519"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+// These tests are written as ATTACKS, not as feature checks: each asserts that a
+// hostile caller CANNOT reach another tenant's resources. Isolation bugs do not
+// show up on the happy path — a caller acting on someone else's resource looks
+// exactly like a caller acting on their own unless something checks. So every
+// case here is "Mallory tries X against Alice's resource and must fail", plus a
+// paired assertion that Alice's own use still works.
+
+// tenancyRegistry mirrors the shape of the real io.pilot.agentphone entry: a
+// shared partner account, ownership tracked broker-side.
+const tenancyRegistryJSON = `[{
+  "id": "io.pilot.phone",
+  "upstream": "%s",
+  "key_env": "PHONE_KEY",
+  "auth_header": "Authorization",
+  "auth_scheme": "Bearer",
+  "quota": 0,
+  "allow": [
+    "GET /v1/agents", "POST /v1/agents", "GET /v1/agents/{agent_id}",
+    "DELETE /v1/agents/{agent_id}",
+    "GET /v1/numbers", "POST /v1/numbers", "DELETE /v1/numbers/{number_id}",
+    "GET /v1/numbers/{number_id}/messages",
+    "POST /v1/messages", "GET /v1/calls"
+  ],
+  "credit": {"seed_credits": 5000000, "default_cost": 0,
+             "cost_credits": {"POST /v1/numbers": 3000000, "POST /v1/messages": 20000}},
+  "tenancy": {
+    "param_types": {"agent_id": "agent", "number_id": "number"},
+    "body_refs":   {"agent_id": "agent", "agentId": "agent", "number_id": "number", "phoneNumberId": "number"},
+    "create": [
+      {"method": "POST", "path": "/v1/agents",  "type": "agent",  "id_field": "id"},
+      {"method": "POST", "path": "/v1/numbers", "type": "number", "id_field": "id"}
+    ],
+    "delete": [
+      {"method": "DELETE", "path": "/v1/numbers/{number_id}", "type": "number", "param": "number_id"}
+    ],
+    "list": [
+      {"method": "GET", "path": "/v1/agents",  "array": "data", "owner_by": [{"field": "id", "type": "agent"}]},
+      {"method": "GET", "path": "/v1/numbers", "array": "data", "owner_by": [{"field": "id", "type": "number"}]},
+      {"method": "GET", "path": "/v1/calls",   "array": "data",
+       "owner_by": [{"field": "phoneNumberId", "type": "number"}, {"field": "agentId", "type": "agent"}]}
+    ]
+  }
+}]`
+
+// phoneUpstream fakes the partner: it answers for the WHOLE shared account,
+// exactly as the real one does — which is why the broker must filter.
+func phoneUpstream(t *testing.T, sent *[]string) *httptest.Server {
+	t.Helper()
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		switch {
+		case r.Method == "POST" && r.URL.Path == "/v1/agents":
+			fmt.Fprintf(w, `{"id":"agent_%d","name":"a"}`, time.Now().UnixNano())
+		case r.Method == "POST" && r.URL.Path == "/v1/numbers":
+			fmt.Fprintf(w, `{"id":"num_%d","phoneNumber":"+1555%04d"}`, time.Now().UnixNano(), time.Now().UnixNano()%10000)
+		case r.Method == "POST" && r.URL.Path == "/v1/messages":
+			b := make([]byte, r.ContentLength)
+			_, _ = r.Body.Read(b)
+			if sent != nil {
+				*sent = append(*sent, string(b))
+			}
+			fmt.Fprint(w, `{"id":"msg_1","status":"sent"}`)
+		case r.URL.Path == "/v1/agents":
+			// the partner sees one shared account: EVERY tenant's agents
+			fmt.Fprint(w, `{"data":[{"id":"agent_alice"},{"id":"agent_mallory"},{"id":"agent_legacy"}]}`)
+		case r.URL.Path == "/v1/numbers":
+			fmt.Fprint(w, `{"data":[{"id":"num_alice","phoneNumber":"+15550001"},{"id":"num_legacy","phoneNumber":"+15550002"}]}`)
+		case r.URL.Path == "/v1/calls":
+			fmt.Fprint(w, `{"data":[{"id":"call_1","phoneNumberId":"num_alice"},{"id":"call_2","phoneNumberId":"num_legacy"}]}`)
+		default:
+			fmt.Fprint(w, `{"ok":true}`)
+		}
+	}))
+}
+
+func tenancyBroker(t *testing.T, sent *[]string) (*Broker, func()) {
+	t.Helper()
+	up := phoneUpstream(t, sent)
+	reg, err := ParseRegistry([]byte(fmt.Sprintf(tenancyRegistryJSON, up.URL)),
+		func(string) string { return "master-key" })
+	if err != nil {
+		t.Fatalf("ParseRegistry: %v", err)
+	}
+	b := New(reg, NewMemStore())
+	b.Verify = VerifyConfig{Window: time.Hour}
+	return b, up.Close
+}
+
+func do(t *testing.T, b *Broker, priv ed25519.PrivateKey, method, path string, body []byte) *httptest.ResponseRecorder {
+	t.Helper()
+	rec := httptest.NewRecorder()
+	b.ServeHTTP(rec, signedReq(t, priv, method, path, body, time.Now()))
+	return rec
+}
+
+// TestTenancy_CannotSendFromAnotherTenantsAgent: Mallory names Alice's agent in
+// the BODY of a send. The path is entirely her own, so only a body-level
+// ownership check can stop this. It must never reach the partner.
+func TestTenancy_CannotSendFromAnotherTenantsAgent(t *testing.T) {
+	var sent []string
+	b, done := tenancyBroker(t, &sent)
+	defer done()
+
+	_, alice := newKey(t)
+	_, mallory := newKey(t)
+
+	// Alice creates an agent → she owns it.
+	rec := do(t, b, alice, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"alice"}`))
+	if rec.Code != 200 {
+		t.Fatalf("alice create agent: %d %s", rec.Code, rec.Body)
+	}
+	var created struct{ ID string }
+	if err := json.Unmarshal(rec.Body.Bytes(), &created); err != nil {
+		t.Fatal(err)
+	}
+
+	// Mallory names Alice's agent in her own request body.
+	body := []byte(fmt.Sprintf(`{"agent_id":%q,"to_number":"+15551234","body":"spam"}`, created.ID))
+	rec = do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", body)
+	if rec.Code != http.StatusNotFound {
+		t.Errorf("cross-tenant send: status %d, want 404 — mallory acted through alice's agent", rec.Code)
+	}
+	if len(sent) != 0 {
+		t.Errorf("cross-tenant send REACHED THE PARTNER: %v", sent)
+	}
+
+	// Alice sending from her own agent still works — isolation must not break the product.
+	rec = do(t, b, alice, "POST", "/io.pilot.phone/v1/messages", body)
+	if rec.Code != 200 {
+		t.Errorf("alice's own send: %d %s, want 200", rec.Code, rec.Body)
+	}
+	if len(sent) != 1 {
+		t.Errorf("alice's send should have reached the partner exactly once, got %d", len(sent))
+	}
+}
+
+// TestTenancy_LegacyResourcesUnreachable: resources that predate the ledger are
+// owned by nobody. Deny-by-default must make them unreachable to EVERYONE rather
+// than unclaimed and therefore available to the first caller who names one.
+func TestTenancy_LegacyResourcesUnreachable(t *testing.T) {
+	var sent []string
+	b, done := tenancyBroker(t, &sent)
+	defer done()
+	_, mallory := newKey(t)
+
+	body := []byte(`{"agent_id":"agent_legacy","to_number":"+15551234","body":"spam"}`)
+	if rec := do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", body); rec.Code != 404 {
+		t.Errorf("legacy agent send: %d, want 404", rec.Code)
+	}
+	if rec := do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers/num_legacy/messages", nil); rec.Code != 404 {
+		t.Errorf("legacy number read: %d, want 404", rec.Code)
+	}
+	// And it must not be claimable by simply asserting it.
+	if rec := do(t, b, mallory, "DELETE", "/io.pilot.phone/v1/numbers/num_legacy", nil); rec.Code != 404 {
+		t.Errorf("legacy number delete: %d, want 404", rec.Code)
+	}
+	if len(sent) != 0 {
+		t.Errorf("legacy resource reached partner: %v", sent)
+	}
+}
+
+// TestTenancy_ListsAreFilteredToOwner: a pilot user must not see any number but
+// their own. The partner answers for the whole shared account, so the broker has
+// to strip the answer down to the caller's rows before returning it.
+func TestTenancy_ListsAreFilteredToOwner(t *testing.T) {
+	b, done := tenancyBroker(t, nil)
+	defer done()
+	_, alice := newKey(t)
+	_, mallory := newKey(t)
+
+	// Nobody owns anything yet → every list is empty, NOT the account-wide answer.
+	rec := do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers", nil)
+	if n := countData(t, rec.Body.Bytes()); n != 0 {
+		t.Errorf("unowned caller saw %d numbers, want 0 — body: %s", n, rec.Body)
+	}
+	if strings.Contains(rec.Body.String(), "+15550002") {
+		t.Errorf("LEAK: another tenant's number was disclosed: %s", rec.Body)
+	}
+
+	// Alice owns num_alice → she sees exactly one, and never num_legacy.
+	if err := b.ownerStore().Claim("io.pilot.phone", "number", "num_alice", callerOf(t, alice), time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	rec = do(t, b, alice, "GET", "/io.pilot.phone/v1/numbers", nil)
+	if n := countData(t, rec.Body.Bytes()); n != 1 {
+		t.Errorf("alice saw %d numbers, want 1 — body: %s", n, rec.Body)
+	}
+	if strings.Contains(rec.Body.String(), "num_legacy") {
+		t.Errorf("LEAK: alice saw another tenant's number: %s", rec.Body)
+	}
+	// Mallory still sees nothing.
+	rec = do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers", nil)
+	if n := countData(t, rec.Body.Bytes()); n != 0 {
+		t.Errorf("mallory saw %d numbers, want 0", n)
+	}
+}
+
+// TestTenancy_InboundResourcesAttributedByLink: an inbound call is created by
+// the partner, so the tenant never "created" it. It must still be visible to the
+// number's owner (via owner_by link) and invisible to everyone else.
+func TestTenancy_InboundCallsFollowNumberOwnership(t *testing.T) {
+	b, done := tenancyBroker(t, nil)
+	defer done()
+	_, alice := newKey(t)
+	_, mallory := newKey(t)
+	if err := b.ownerStore().Claim("io.pilot.phone", "number", "num_alice", callerOf(t, alice), time.Now()); err != nil {
+		t.Fatal(err)
+	}
+
+	rec := do(t, b, alice, "GET", "/io.pilot.phone/v1/calls", nil)
+	if n := countData(t, rec.Body.Bytes()); n != 1 {
+		t.Errorf("alice saw %d calls, want 1 (her inbound call) — %s", n, rec.Body)
+	}
+	if strings.Contains(rec.Body.String(), "call_2") {
+		t.Errorf("LEAK: alice saw a call on another tenant's number: %s", rec.Body)
+	}
+	rec = do(t, b, mallory, "GET", "/io.pilot.phone/v1/calls", nil)
+	if n := countData(t, rec.Body.Bytes()); n != 0 {
+		t.Errorf("mallory saw %d calls, want 0", n)
+	}
+}
+
+// TestTenancy_NestedBodyRefIsChecked: wrapping the stolen id one level deep must
+// not bypass the check. A top-level-only scan would be trivially defeated.
+func TestTenancy_NestedBodyRefIsChecked(t *testing.T) {
+	var sent []string
+	b, done := tenancyBroker(t, &sent)
+	defer done()
+	_, alice := newKey(t)
+	_, mallory := newKey(t)
+
+	rec := do(t, b, alice, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"alice"}`))
+	var created struct{ ID string }
+	_ = json.Unmarshal(rec.Body.Bytes(), &created)
+
+	for _, body := range []string{
+		fmt.Sprintf(`{"message":{"agent_id":%q},"to_number":"+1555"}`, created.ID),      // nested object
+		fmt.Sprintf(`{"batch":[{"agent_id":%q}],"to_number":"+1555"}`, created.ID),      // inside an array
+		fmt.Sprintf(`{"a":{"b":{"c":{"agentId":%q}}},"to_number":"+1555"}`, created.ID), // deep + camelCase alias
+	} {
+		rec := do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", []byte(body))
+		if rec.Code != 404 {
+			t.Errorf("nested ref %s: status %d, want 404", body, rec.Code)
+		}
+	}
+	if len(sent) != 0 {
+		t.Errorf("a nested cross-tenant ref reached the partner: %v", sent)
+	}
+}
+
+// TestTenancy_OwnershipIsNotTransferable: a second caller claiming a resource
+// that already has an owner must lose. Otherwise "claim" is a takeover.
+func TestTenancy_OwnershipIsNotTransferable(t *testing.T) {
+	s := NewMemStore()
+	now := time.Now()
+	if err := s.Claim("app", "number", "num_1", "alice", now); err != nil {
+		t.Fatalf("alice claim: %v", err)
+	}
+	if err := s.Claim("app", "number", "num_1", "mallory", now); err != ErrOwned {
+		t.Errorf("mallory claim: %v, want ErrOwned", err)
+	}
+	if !Owns(s, "app", "number", "num_1", "alice") {
+		t.Error("alice must still own num_1")
+	}
+	if Owns(s, "app", "number", "num_1", "mallory") {
+		t.Error("mallory must NOT own num_1")
+	}
+	// Re-claim by the true owner stays idempotent.
+	if err := s.Claim("app", "number", "num_1", "alice", now); err != nil {
+		t.Errorf("idempotent re-claim: %v", err)
+	}
+}
+
+// TestTenancy_TypeConfusion: an id owned as one type must not authorize the same
+// id used as another type.
+func TestTenancy_TypeConfusion(t *testing.T) {
+	s := NewMemStore()
+	_ = s.Claim("app", "agent", "x_1", "alice", time.Now())
+	if Owns(s, "app", "number", "x_1", "alice") {
+		t.Error("owning agent x_1 must not confer ownership of number x_1")
+	}
+}
+
+// TestTenancy_FailsClosedWithoutLedger: tenancy declared but the store cannot
+// record ownership → deny, never serve unisolated.
+func TestTenancy_FailsClosedWithoutLedger(t *testing.T) {
+	tn := &Tenancy{ParamTypes: map[string]string{"agent_id": "agent"}}
+	tn.compile()
+	if _, ok := tn.EnforceRequest(nil, "app", nil, "GET", "/v1/agents/a1", "", nil, "alice"); ok {
+		t.Error("nil ledger must deny")
+	}
+	if _, ok := tn.EnforceRequest(NewMemStore(), "app", nil, "GET", "/v1/agents/a1", "", nil, ""); ok {
+		t.Error("empty caller must deny")
+	}
+}
+
+// TestTenancy_MalformedBodyDenied: a body that cannot be parsed cannot be
+// checked, so it must not be forwarded.
+func TestTenancy_MalformedBodyDenied(t *testing.T) {
+	var sent []string
+	b, done := tenancyBroker(t, &sent)
+	defer done()
+	_, mallory := newKey(t)
+	rec := do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", []byte(`{"agent_id": broken`))
+	if rec.Code != 404 {
+		t.Errorf("malformed body: %d, want 404", rec.Code)
+	}
+	if len(sent) != 0 {
+		t.Errorf("malformed body reached partner: %v", sent)
+	}
+}
+
+// countData counts elements of the "data" array in a JSON body.
+func countData(t *testing.T, b []byte) int {
+	t.Helper()
+	var v struct {
+		Data []any `json:"data"`
+	}
+	if err := json.Unmarshal(b, &v); err != nil {
+		t.Fatalf("unmarshal %s: %v", b, err)
+	}
+	return len(v.Data)
+}
+
+// callerOf renders a private key's identity the way the broker records it.
+func callerOf(t *testing.T, priv ed25519.PrivateKey) string {
+	t.Helper()
+	return base64.RawStdEncoding.EncodeToString(priv.Public().(ed25519.PublicKey))
+}
+
+// TestTenancy_DuplicateKeyBypassDenied: the broker validates the body with one
+// parser and forwards raw bytes for the partner to re-parse with another. When a
+// key repeats, the two parsers can disagree about its value, and the broker would
+// authorize a value the partner never acts on. Such a body must be refused.
+func TestTenancy_DuplicateKeyBypassDenied(t *testing.T) {
+	var sent []string
+	b, done := tenancyBroker(t, &sent)
+	defer done()
+	_, alice := newKey(t)
+	_, mallory := newKey(t)
+
+	rec := do(t, b, alice, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"alice"}`))
+	var created struct{ ID string }
+	_ = json.Unmarshal(rec.Body.Bytes(), &created)
+
+	// Mallory owns an agent of her own, and names Alice's FIRST.
+	rec = do(t, b, mallory, "POST", "/io.pilot.phone/v1/agents", []byte(`{"name":"mallory"}`))
+	var mine struct{ ID string }
+	_ = json.Unmarshal(rec.Body.Bytes(), &mine)
+
+	body := []byte(fmt.Sprintf(`{"agent_id":%q,"agent_id":%q,"to_number":"+1555","body":"spam"}`, created.ID, mine.ID))
+	rec = do(t, b, mallory, "POST", "/io.pilot.phone/v1/messages", body)
+	if rec.Code != 404 {
+		t.Errorf("duplicate-key bypass: status %d, want 404", rec.Code)
+	}
+	if len(sent) != 0 {
+		t.Errorf("duplicate-key body REACHED THE PARTNER: %v", sent)
+	}
+}
+
+// TestTenancy_CountsDoNotLeakUnfilteredSet: filtering the array is not enough.
+// A sibling count left at the partner's value still reports the size of the
+// whole shared account, so a caller who can see none of the rows can still read
+// how many exist — and watch that number move as other tenants work.
+func TestTenancy_CountsDoNotLeakUnfilteredSet(t *testing.T) {
+	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, `{"data":[{"id":"num_alice"},{"id":"num_other"},{"id":"num_third"}],"total":3,"hasMore":true}`)
+	}))
+	defer up.Close()
+	reg, err := ParseRegistry([]byte(`[{
+	  "id":"io.pilot.phone","upstream":"`+up.URL+`","key_env":"K","auth_header":"Authorization","auth_scheme":"Bearer",
+	  "allow":["GET /v1/numbers"],
+	  "tenancy":{"param_types":{"number_id":"number"},
+	    "create":[{"method":"POST","path":"/v1/numbers","type":"number","id_field":"id"}],
+	    "list":[{"method":"GET","path":"/v1/numbers","array":"data",
+	             "owner_by":[{"field":"id","type":"number"}],"claim_as":"number",
+	             "count_fields":["total"]}]}}]`), func(string) string { return "k" })
+	if err != nil {
+		t.Fatalf("ParseRegistry: %v", err)
+	}
+	b := New(reg, NewMemStore())
+	b.Verify = VerifyConfig{Window: time.Hour}
+	_, mallory := newKey(t)
+
+	rec := do(t, b, mallory, "GET", "/io.pilot.phone/v1/numbers", nil)
+	var got struct {
+		Data  []any       `json:"data"`
+		Total json.Number `json:"total"`
+	}
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatalf("unmarshal %s: %v", rec.Body, err)
+	}
+	if len(got.Data) != 0 {
+		t.Errorf("data = %d rows, want 0", len(got.Data))
+	}
+	if got.Total.String() != "0" {
+		t.Errorf("total = %s, want 0 — the account-wide count leaked", got.Total)
+	}
+}