CB-5xx: Stage 5 hardening — auth, authz+audit, metrics, CI, supervision

Closes out single-host before the cross-host work. Sequenced BEFORE CB-308
deliberately: federation's own gating concern is the trust model, and it
inherits whatever identity shape lands here.

The finding this stage is built around: bridged had exactly ONE security
control, the loopback bind. ConnectionIdentity resolves a worker from its
connection (unforgeable), but every caller that was not a recognised worker
pane fell through to being treated as the PRIMARY -- the most privileged role
on the bus. Latent today; load-bearing the moment a bind widens.

CB-501 auth:
- Role/Principal/CallerResolver: connection identity first, bearer token
  second, ANONYMOUS third. Inverts the old default so absence of identity
  means nothing, not everything.
- Worker identity is never token-gated, so enabling auth cannot lock the
  fleet out of bridge_reply.
- Constant-time token compare (MessageDigest.isEqual).
- validateAuthExposure(): a non-loopback bind under loopback-trust now
  REFUSES TO START. Makes the dangerous config unrepresentable rather than
  merely documented.
- TLS terminates at a reverse proxy by design (D3), not in the JVM.

CB-505 authz + audit, enforced on BOTH entry paths:
- The docs describe MCP as "a thin adapter over the REST core"; at code level
  it is not. BridgeMcp calls MessageService directly, and /mcp is a raw
  servlet on Jetty's context handler that never traverses Javalin's before
  filter. Enforcing only at REST would have left /mcp open.
- Load-bearing rule is own-session-only: a worker may reply/ask only as
  itself. Structurally true over MCP already; over REST the session id in the
  URL path had simply been trusted.
- Audit: JSON lines to a dedicated appender, additivity=false. Never records
  message content -- this bus carries source and prompts.

CB-502 metrics: zero new dependencies. A ~150-line Prometheus text renderer
instead of the specced Micrometer, because this pom already hand-pins
jackson-annotations to reconcile Jackson 2/3, imports a Jetty BOM against
skew, and carries four accepted-CVE advisories -- and CLAUDE.md's mandated
dependency CVE gate could not be run (no JetBrains MCP server connected).
Instrumented at MessageService, the single funnel both surfaces share.

CB-503 CI: .gitea/workflows/ci.yml against the already-running Gitea runner.
Needs no contract-exclusion flag -- the pom's default-excludes profile
already sets excludedGroups=contract, so plain `mvn clean install` IS the
mock-socket surface. Provisions JDK 25 explicitly (runner default-jdk is older).

CB-504 supervision: launchd agent (the real target -- this host is macOS,
there is no systemd) plus a systemd unit for the Linux gateways CB-308 adds.
Ordering directives are advisory, so the actual fix is that startup now waits
up to 30s for the herdr socket and then serves degraded, instead of crashing
into a restart loop on a boot-order race.

Also fixes drift found while surveying:
- bridged.example.yaml documented spawn_ready_timeout_ms in snake_case; config
  binds via plain Jackson with ignoreUnknown, so uncommenting it would have
  been silently dropped and the default kept. Now camelCase, with a test that
  loads the shipped example and one that pins every documented knob's
  spelling -- no test had ever loaded that file.
- Added the 6 shipped-but-undocumented knobs (worktreeRoot, parityOverlay,
  gitTokenEnv, gitHostEnv, configDir, primary:).
- README "Next" listed bridge_ask and session lifecycle as upcoming; both
  shipped long ago.
- docs/CB-301-ext and docs/CB-402 status headers said "design"/"pre-
  implementation" for work already merged.

307 unit/acceptance tests green (was 266), mvn clean install BUILD SUCCESS.
Note: CLAUDE.md's per-file ide_diagnostics gate and the pom Mend.io CVE check
could not be run -- no JetBrains/intellij-index MCP server is connected this
session. mvn clean install is the only gate that ran.

.gitea/workflows/ci.yml

@@ -0,0 +1,40 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      # The wiki submodule is docs only and is not needed to build — leave it unfetched so CI
+      # does not depend on the wiki repo being reachable.
+      - uses: actions/checkout@v4
+
+      # The runner image ships an older default-jdk; bridged sets maven.compiler.release=25, so
+      # provision the JDK explicitly rather than apt-installing whatever "default" means today.
+      - name: Set up JDK 25
+        uses: actions/setup-java@v4
+        with:
+          distribution: temurin
+          java-version: '25'
+          cache: maven
+
+      - name: Build and test
+        working-directory: bridged
+        # This IS the mock-socket surface CB-503 asks for: the pom's `default-excludes` profile
+        # already sets excludedGroups=contract, so the @Tag("contract") tests — which need a live
+        # herdr socket and a RabbitMQ container — are excluded without any flag here. Everything
+        # that runs does so against the fake UDS herdr and fake ccs/claude stubs.
+        run: mvn -B clean install
+
+      - name: Publish test report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: surefire-reports
+          path: bridged/target/surefire-reports/
+          if-no-files-found: warn

README.md

@@ -92,7 +92,8 @@ bridge (code reviews delegated this way have produced committed bug fixes). Sele
 primary approach 2026-07-11, superseding the AgentAPI plan (2026-07-08); AgentAPI retained as a
 fallback injector.
 
-**Shipped** (Java 25 · Maven · 105 tests green — unit/acceptance + live-herdr contract tests):
+**Shipped** (Java 25 · Maven · 266 unit/acceptance tests green; the live-herdr and broker contract
+tests run separately via `mvn test -Pcontract`):
 
 - **Core gateway** — herdr socket client (contract-tested vs live 0.7.0); guard-checked worker
   spawn with `ANTHROPIC_BASE_URL` injected only into the worker's env; status-gated injector;
@@ -106,7 +107,17 @@ fallback injector.
 - **Fleet** — multiple worker profiles, each with an independent base_url guard check; workers
   inherit the primary's working directory (never `$HOME`); a readiness gate holds delivery until
   a worker's Claude has connected the bridge MCP (no paste lost into its boot window).
+- **Blocked-worker path** — `bridge_ask` reverse rendezvous: a worker pauses its delegated turn to
+  ask the primary and resumes the *same* turn with the answer (CB-205).
+- **Session lifecycle** — session manager with spawn/reuse/recycle, `idle_ttl` reaper, `context_cap`,
+  and graceful drain on shutdown (CB-301/CB-303); per-worker git worktrees on their own branch with
+  a config-parity overlay, so parallel implementers never stomp each other (CB-301-ext).
+- **Reliable worker→primary delivery** — a durable `ReplyInbox` (in-memory by default, AMQP/LavinMQ
+  for cross-restart durability) holds a reply that arrives with no open send, and an active
+  status-gated push loop nudges the primary to drain it (CB-307).
+- **Pluggable peers** — a `PeerLauncher` SPI with two in-tree adapters, `claude-code` and `opencode`,
+  routed by a `kind:` discriminator (CB-401/CB-402).
 
-**Next** (see the [roadmap](wiki/8-Roadmap.md)) — structured envelope schema, `bridge_ask`
+**Next** (see the [roadmap](wiki/8-Roadmap.md)) — Stage 5 hardening (auth/TLS, `/metrics`, CI,
-(blocked-worker path), session lifecycle / recycle / `idle_ttl`, split-host, and hardening
+service supervision, per-session authz + audit), then cross-host: CB-308 multi-host federation and
-(auth/TLS, `/metrics`, CI, systemd).
+CB-500 multi-tier coordination.

bridged/.gitignore

@@ -5,6 +5,9 @@ dependency-reduced-pom.xml
 # Local runtime config (copy from bridged.example.yaml)
 bridged.yaml
 
+# CB-505 audit trail + daemon stdout/stderr — runtime records, never source
+logs/
+
 # Editor / OS
 *.iml
 .idea/

bridged/bridged.example.yaml

@@ -3,11 +3,34 @@
 # bridged is the sole gateway between primary/worker Claude sessions and herdr.
 # It is NOT a Claude process and must never carry ANTHROPIC_BASE_URL.
 
-# REST + MCP listen address. Keep it on loopback — bridged is same-host in Stage-1.
+# REST + MCP listen address. Keep it on loopback unless you also switch auth.mode to `token`
+# below — bridged REFUSES TO START on a non-loopback bind under loopback-trust (see auth).
 bind:
   host: 127.0.0.1
   port: 8765
 
+# API authentication (CB-501). Governs how a caller that is NOT an on-host worker pane proves it
+# is the primary. Worker identity never depends on this: a loopback peer PID that maps to a herdr
+# pane is unforgeable and is always honoured, so turning auth on cannot lock the fleet out.
+#
+#   mode: loopback-trust  → DEFAULT, and the historical behaviour: any loopback caller that is not
+#                           a worker is the primary, no credential needed. Sound ONLY because the
+#                           OS refuses remote connections to a loopback socket.
+#   mode: token           → such a caller must send `Authorization: Bearer <token>`; without it it
+#                           is anonymous and authorized for nothing. REQUIRED for a non-loopback
+#                           bind — the daemon fails fast otherwise, because "unauthenticated ⇒
+#                           primary" on a reachable port would hand spawn/stop/send to anyone.
+#   tokenEnv              → host env var holding the token (never the literal value). Default
+#                           BRIDGED_API_TOKEN. Read only in token mode; empty ⇒ startup fails.
+#
+# TLS is deliberately NOT terminated in the daemon (CB-501 D3): run a reverse proxy in front and
+# let it own certificate lifecycle, e.g.
+#     location / { proxy_pass http://127.0.0.1:8765; proxy_set_header Authorization $http_authorization; }
+# The broker link gets TLS from its own URI (amqps://…) — see `broker` below.
+# auth:
+#   mode: token
+#   tokenEnv: BRIDGED_API_TOKEN
+
 # herdr Unix socket. Omit to use the client default
 # (${HERDR_SOCKET_PATH:-~/.config/herdr/herdr.sock}).
 herdrSocket: ~/.config/herdr/herdr.sock
@@ -26,6 +49,17 @@ herdrSocket: ~/.config/herdr/herdr.sock
 #   cwd              → pin this profile's working directory (CB-112). Omit to inherit the primary's
 #                      cwd on an MCP spawn, else the daemon's cwd — never $HOME. See
 #                      docs/Worker-Startup-and-Trust.md.
+#   configDir        → CLAUDE_CONFIG_DIR for the worker, so it inherits that profile's
+#                      skills/MCP/hooks. Omit to leave the worker on the host default.
+#   parityOverlay    → repo-relative paths copied primary→worktree so a worker in a provisioned
+#                      worktree sees the same local config (CB-301-ext). Omit for the default set:
+#                      [.mcp.json, .claude/settings.local.json, .env, .envrc].
+#   gitTokenEnv      → host env var holding the git-forge API token. When set, its value is injected
+#                      as GITEA_TOKEN so the worker can open its OWN PR at checkpoint (CB-302).
+#                      Opt-in by design — omit and the worker gets no PR-create grant (push over
+#                      SSH is unaffected). The token value itself is never stored in this file.
+#   gitHostEnv       → host env var holding the forge host (default GITEA_HOST). Injected as
+#                      GITEA_HOST *only* alongside a resolved gitTokenEnv.
 # Put `defaultMode: "auto"` in each ccs profile so the worker runs autonomously.
 workers:
   gx10:                              # ccs profile name (NOT a hostname)
@@ -38,6 +72,11 @@ workers:
     mcpUrl: http://127.0.0.1:8765/mcp
     tokenEnv: BRIDGED_WORKER_TOKEN
     argv: ["ccs", "gx10"]
+    # gitTokenEnv: GITEA_TOKEN        # opt-in: let this profile's workers open their own PR (CB-302)
+    # gitHostEnv: GITEA_HOST          # defaults to GITEA_HOST; injected only with gitTokenEnv
+    # configDir: /Users/me/.ccs/instances/gx10   # CLAUDE_CONFIG_DIR — inherit that profile's skills/MCP
+    # cwd: /Users/me/src/myrepo       # pin the working dir; omit to inherit the primary's
+    # parityOverlay: [".mcp.json", ".claude/settings.local.json", ".env", ".envrc"]
   ollama:
     baseUrl: http://ollama.ltms.dev  # local/self-hosted; usually no token
     placement: tab
@@ -70,8 +109,15 @@ guard:
 
 # Spawn-readiness gate (CB-306). The launcher blocks until the worker's herdr status is
 # injectable (IDLE/BLOCKED/DONE) or the timeout elapses. 0 disables the gate.
-# spawn_ready_timeout_ms: 20000
+# NOTE: keys are camelCase — config is bound by plain Jackson with no naming strategy and
-# spawn_ready_poll_ms: 300
+# unknown keys are ignored, so a snake_case key would be silently dropped (default kept).
+# spawnReadyTimeoutMs: 20000
+# spawnReadyPollMs: 300
+
+# Worktree provisioning root (CB-301-ext). Where per-worker git worktrees are checked out so
+# each worker owns an isolated branch instead of sharing the primary's tree. Omit to default
+# to a sibling directory of the repo root.
+# worktreeRoot: /Users/me/src/.bridged-worktrees
 
 # Session lifecycle limits (CB-303). All knobs are opt-in; omit or set to null to keep
 # the feature disabled. By default the daemon never reaps, caps, or drains sessions.
@@ -93,3 +139,18 @@ guard:
 #         is vhost "" and will NOT connect. Encode a named vhost as .../%2Fmyvhost.
 # broker:
 #   uri: amqp://guest:guest@127.0.0.1:5672
+
+# Active push-to-primary (CB-307 Stage 3). When a worker reply lands with no open bridge_send,
+# the ReplyPushLoop injects a *drain nudge* (never the payload) into the primary's own herdr
+# pane — status-gated (only when injectable, never mid-turn) and bounded. Ack = drain: the loop
+# stops as soon as the primary's inbox is empty.
+#   terminal       → pin the primary's herdr terminal id. Omit to learn it from the connection on
+#                    the first orchestration-side MCP call (the normal case). An off-host or
+#                    non-herdr primary leaves this unresolved → the loop is a no-op and delivery
+#                    degrades to pull; the reply is still never lost.
+#   pushReminders  → max nudges before giving up (default 5)
+#   pushBackoffMs  → delay between nudges in ms (default 15000)
+# primary:
+#   terminal: term_65619bd6174568
+#   pushReminders: 5
+#   pushBackoffMs: 15000

bridged/src/main/java/dev/ltms/bridged/Bridged.java

@@ -3,6 +3,8 @@ package dev.ltms.bridged;
 import dev.ltms.bridged.config.BridgedConfig;
 import dev.ltms.bridged.guard.SubscriptionGuard;
 import dev.ltms.bridged.herdr.AgentControl;
+import dev.ltms.bridged.herdr.HerdrClient;
+import dev.ltms.bridged.herdr.HerdrException;
 import dev.ltms.bridged.herdr.PaneLocator;
 import dev.ltms.bridged.herdr.UnixSocketHerdrClient;
 import dev.ltms.bridged.herdr.WorkspaceControl;
@@ -11,8 +13,11 @@ import dev.ltms.bridged.inject.Injector;
 import dev.ltms.bridged.inject.StatusPoller;
 import dev.ltms.bridged.inject.TurnListener;
 import dev.ltms.bridged.inject.WorkerPresence;
+import dev.ltms.bridged.auth.CallerResolver;
 import dev.ltms.bridged.mcp.BridgeMcp;
 import dev.ltms.bridged.mcp.ConnectionIdentity;
+import dev.ltms.bridged.metrics.BridgedMetrics;
+import dev.ltms.bridged.metrics.Metrics;
 import dev.ltms.bridged.mcp.PrimaryRegistry;
 import dev.ltms.bridged.mcp.LsofPeerPidLookup;
 import dev.ltms.bridged.mcp.LsofProcessCwdLookup;
@@ -54,6 +59,10 @@ public final class Bridged {
     /** How often the injector samples a busy worker's status while it has queued work. */
     private static final long INJECT_POLL_MILLIS = 250;
 
+    /** CB-504: how long to wait at startup for herdr's socket before serving degraded. */
+    private static final long HERDR_WAIT_SECONDS = 30;
+    private static final long HERDR_WAIT_POLL_MILLIS = 500;
+
     static void main(String[] args) {
         Path configPath = Path.of(args.length > 0 ? args[0] : "bridged.yaml");
         BridgedConfig cfg = BridgedConfig.load(configPath);
@@ -62,6 +71,12 @@ public final class Bridged {
         SubscriptionGuard guard = new SubscriptionGuard(cfg.guard().hostSet());
         guard.assertPrimaryClean(System.getenv());
 
+        // CB-501: refuse to start if the bind is wider than the auth mode can defend. Under
+        // loopback-trust, "not a known worker" means "the primary" — sound only because the OS
+        // refuses remote connections to a loopback socket. This throws rather than warns so the
+        // dangerous configuration cannot be reached by ignoring a log line.
+        cfg.validateAuthExposure();
+
         Path socket = cfg.herdrSocket() != null && !cfg.herdrSocket().isBlank()
                 ? Path.of(cfg.herdrSocket())
                 : UnixSocketHerdrClient.defaultSocketPath();
@@ -97,9 +112,20 @@ public final class Bridged {
                     cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs()));
         }
         PeerLauncher workers = new CompositePeerLauncher(adapters, cfg.defaultProfile());
-        // CB-117: herdr keeps worker panes alive across a daemon restart, and their ids died with
+        // CB-504: under supervision (launchd/systemd) bridged can start before herdr's socket
-        // the previous process — reap those leaked orphans now, before we start serving.
+        // exists. The client itself is lazy — it connects per call — but the orphan reap below is
-        workers.reapOrphanWorkers();
+        // the first thing that actually talks to herdr, so without this wait a boot-order race
+        // would crash the daemon into a restart loop. Wait, then degrade rather than die: serving
+        // with /healthz reporting "degraded" is strictly more useful than exiting.
+        if (awaitHerdr(herdr)) {
+            // CB-117: herdr keeps worker panes alive across a daemon restart, and their ids died
+            // with the previous process — reap those leaked orphans now, before we start serving.
+            workers.reapOrphanWorkers();
+        } else {
+            log.warn("herdr did not answer within {}s — starting anyway; /healthz will report "
+                    + "degraded until it comes up. Orphaned worker panes (if any) were NOT reaped.",
+                    HERDR_WAIT_SECONDS);
+        }
 
         // CB-301: authoritative session registry + lifecycle FSM on top of ClaudeCodeLauncher.
         // CB-301-ext: worktree provisioning seam, optionally rooted at a configured directory.
@@ -176,14 +202,36 @@ public final class Bridged {
                 Thread.ofVirtual().name("bridge-push-").unstarted(r));
         var pushLoop = new ReplyPushLoop(primaryRegistry, agents, replyInbox,
                 pushScheduler, maxReminders, backoffMs);
-        MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox, pushLoop);
+        // CB-502: the registry is built before the service so send/reply outcomes are counted at
+        // their single funnel rather than at each of the two caller-facing surfaces.
+        Metrics metrics = BridgedMetrics.create(sessions, replyInbox);
+        MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox,
+                pushLoop, metrics);
 
         // MCP server face (CB-105): bridge_send/bridge_reply/bridge_status, mounted at /mcp.
         // Caller identity is resolved from the connection (peer PID → herdr pane), not arguments.
         ConnectionIdentity identity = new ConnectionIdentity(
                 new PaneLocator(herdr), new LsofPeerPidLookup(), new LsofProcessCwdLookup());
+
+        // CB-501: one resolver behind both entry paths. Worker identity still comes from the
+        // connection and is never token-gated, so enabling token mode cannot lock the fleet out.
+        final CallerResolver callers;
+        if (cfg.auth().tokenMode()) {
+            String token = System.getenv(cfg.auth().tokenEnv());
+            if (token == null || token.isBlank()) {
+                throw new IllegalStateException("auth.mode=token but env var " + cfg.auth().tokenEnv()
+                        + " is unset or empty — export it before starting bridged");
+            }
+            callers = new CallerResolver(identity, true, token);
+            log.info("auth: token mode (bearer required for non-worker callers, env {})",
+                    cfg.auth().tokenEnv());
+        } else {
+            callers = new CallerResolver(identity);
+            log.info("auth: loopback-trust (any loopback non-worker caller is the primary)");
+        }
+
         BridgeMcp mcp = new BridgeMcp(messages, workers, sessions, identity, presence,
-                primaryRegistry);
+                primaryRegistry, callers, metrics);
 
         // CB-303 part 3: single ordered shutdown hook. Drain sessions first while herdr is still
         // open (so releases reach the daemon), then stop poller/message/mcp/reaper, and close herdr
@@ -206,12 +254,46 @@ public final class Bridged {
             herdr.close();
         }));
 
-        Javalin app = new BridgedApp(herdr, workers, sessions, messages, presence, mcp.servlet()).build();
+        Javalin app = new BridgedApp(herdr, workers, sessions, messages, presence, mcp.servlet(),
+                callers, metrics).build();
         app.start(cfg.bind().host(), cfg.bind().port());
         log.info("bridged listening on {}:{}, herdr socket {}",
                 cfg.bind().host(), cfg.bind().port(), socket);
     }
 
+    /**
+     * Poll herdr's {@code ping} until it answers or {@link #HERDR_WAIT_SECONDS} elapses (CB-504).
+     *
+     * @return true if herdr answered, false if it never did
+     */
+    private static boolean awaitHerdr(HerdrClient herdr) {
+        long deadline = System.nanoTime() + HERDR_WAIT_SECONDS * 1_000_000_000L;
+        boolean waited = false;
+        while (true) {
+            try {
+                herdr.call("ping");
+                if (waited) {
+                    log.info("herdr is up");
+                }
+                return true;
+            } catch (HerdrException e) {
+                if (System.nanoTime() >= deadline) {
+                    return false;
+                }
+                if (!waited) {
+                    log.info("waiting up to {}s for the herdr socket…", HERDR_WAIT_SECONDS);
+                    waited = true;
+                }
+                try {
+                    Thread.sleep(HERDR_WAIT_POLL_MILLIS);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    return false;
+                }
+            }
+        }
+    }
+
     private Bridged() {
     }
 }

bridged/src/main/java/dev/ltms/bridged/auth/AuditLog.java

@@ -0,0 +1,63 @@
+package dev.ltms.bridged.auth;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Append-only record of privileged actions (CB-505).
+ *
+ * <p>Writes JSON lines to a dedicated {@code audit} logger — its own appender, separate from the
+ * chatty app log — so the trail stays greppable and can later be shipped without dragging debug
+ * noise along.
+ *
+ * <p><strong>Message content is never recorded.</strong> This bridge carries the user's source
+ * code, diffs, and prompts; an audit trail that quietly accumulated them would be a transcript
+ * archive wearing a security control's clothing. Records carry <em>who / what / against what /
+ * outcome</em> and correlation ids only.
+ */
+public final class AuditLog {
+
+    private static final Logger AUDIT = LoggerFactory.getLogger("audit");
+
+    private AuditLog() {
+    }
+
+    /** Record an allowed action. */
+    public static void allowed(Principal caller, Authz.Action action, String target) {
+        write(caller, action, target, "allowed", null);
+    }
+
+    /** Record a refused action and why. */
+    public static void denied(Principal caller, Authz.Action action, String target, String reason) {
+        write(caller, action, target, "denied", reason);
+    }
+
+    /** Record an action that was authorized but then failed downstream (guard, timeout, herdr). */
+    public static void failed(Principal caller, Authz.Action action, String target, String reason) {
+        write(caller, action, target, "failed", reason);
+    }
+
+    private static void write(Principal caller, Authz.Action action, String target,
+                              String outcome, String reason) {
+        Principal c = caller != null ? caller : Principal.anonymous();
+        StringBuilder sb = new StringBuilder(160);
+        sb.append("{\"role\":\"").append(c.role()).append('"')
+                .append(",\"actor\":\"").append(esc(c.describe())).append('"')
+                .append(",\"pid\":").append(c.pid())
+                .append(",\"action\":\"").append(action).append('"')
+                .append(",\"target\":").append(target == null ? "null" : '"' + esc(target) + '"')
+                .append(",\"outcome\":\"").append(outcome).append('"');
+        if (reason != null) {
+            sb.append(",\"reason\":\"").append(esc(reason)).append('"');
+        }
+        sb.append('}');
+        // The appender supplies the timestamp, so it cannot disagree with the app log's clock.
+        AUDIT.info(sb.toString());
+    }
+
+    /** Minimal JSON string escaping — these values are ids and short reasons, never free text. */
+    private static String esc(String s) {
+        return s.replace("\\", "\\\\").replace("\"", "\\\"")
+                .replace("\n", "\\n").replace("\r", "\\r").replace("\t", "\\t");
+    }
+}

bridged/src/main/java/dev/ltms/bridged/auth/Authz.java

@@ -0,0 +1,72 @@
+package dev.ltms.bridged.auth;
+
+/**
+ * The authorization table (CB-505), stated once and enforced on both entry paths.
+ *
+ * <p>Most of these rules are already true de facto — {@code BridgeMcp} derives a worker's identity
+ * from the connection rather than reading it from an argument, so a worker has never been able to
+ * reply <em>as</em> another worker over MCP. What was missing is that the REST surface trusted the
+ * session id in the URL path, and neither surface checked role at all. This class makes the
+ * invariant explicit and testable rather than emergent.
+ */
+public final class Authz {
+
+    private Authz() {
+    }
+
+    /** A privileged operation, named for the audit trail. */
+    public enum Action {
+        /** Spawn a worker peer. */
+        SPAWN,
+        /** Tear a worker peer down. */
+        STOP,
+        /** Deliver a turn to a session (or answer a worker's question). */
+        SEND,
+        /** A worker's terminal reply for its own turn. */
+        REPLY,
+        /** A worker's mid-turn question to the primary. */
+        ASK,
+        /** Collect held replies from a session's inbox. */
+        DRAIN,
+        /** Read-only observation: status, roster, profiles, task polling. */
+        READ,
+        /** Scrape the metrics endpoint. */
+        METRICS
+    }
+
+    /**
+     * Whether {@code caller} may perform {@code action} against {@code targetSession}.
+     *
+     * @param targetSession the session id in the request path; only consulted for the worker-scoped
+     *                      actions ({@code REPLY}, {@code ASK}), ignored otherwise, may be
+     *                      {@code null}
+     */
+    public static boolean permits(Principal caller, Action action, String targetSession) {
+        if (caller == null || caller.isAnonymous()) {
+            return false; // authenticated as nothing ⇒ authorized for nothing
+        }
+        return switch (action) {
+            // Orchestration is the primary's alone. A worker driving spawn/stop/send would be a
+            // worker escalating into the orchestrator role.
+            case SPAWN, STOP, SEND, DRAIN -> caller.isPrimary();
+
+            // The load-bearing rule: a worker acts only as itself. The primary is deliberately
+            // excluded — a reply/ask is a worker's own turn output, and letting the primary forge
+            // one would corrupt the rendezvous correlation it is itself waiting on.
+            case REPLY, ASK -> caller.ownsSession(targetSession);
+
+            // Observation is open to both authenticated roles: a worker legitimately polls its own
+            // status, and the roster carries no secrets.
+            case READ, METRICS -> caller.isPrimary() || caller.isWorker();
+        };
+    }
+
+    /**
+     * Why a request was refused, for the error body. Distinguishes "you are nobody" from "you are
+     * somebody, but not the right somebody" — the first is a credential problem (401), the second
+     * an authorization one (403).
+     */
+    public static boolean isUnauthenticated(Principal caller) {
+        return caller == null || caller.isAnonymous();
+    }
+}

bridged/src/main/java/dev/ltms/bridged/auth/CallerResolver.java

@@ -0,0 +1,119 @@
+package dev.ltms.bridged.auth;
+
+import dev.ltms.bridged.mcp.ConnectionIdentity;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+
+/**
+ * Resolves every caller to a {@link Principal}, for both entry paths into the core (CB-501).
+ *
+ * <p>There are two of them and they are not layered the way the docs suggest: {@code BridgeMcp}
+ * calls the service layer directly and is mounted as a raw servlet (so it never passes through a
+ * Javalin filter), while the REST routes historically resolved no identity at all. Both now
+ * delegate here, so the authorization rules are stated once instead of drifting apart.
+ *
+ * <p><strong>Resolution order</strong> — connection identity first, token second, nothing third:
+ * <ol>
+ *   <li>A loopback peer PID that maps to a herdr worker pane ⇒ {@link Role#WORKER}. This is
+ *       unforgeable (the OS reports the PID, herdr owns the PID→pane map) and is honoured
+ *       regardless of auth mode, so enabling auth never breaks the fleet.</li>
+ *   <li>Otherwise, under {@code token} mode, a valid bearer token ⇒ {@link Role#PRIMARY}.</li>
+ *   <li>Otherwise, under {@code loopback-trust}, a loopback caller ⇒ {@link Role#PRIMARY}
+ *       (the historical behaviour, now an explicit configured choice).</li>
+ *   <li>Otherwise {@link Role#ANONYMOUS}.</li>
+ * </ol>
+ */
+public final class CallerResolver {
+
+    private final ConnectionIdentity identity;
+    private final boolean tokenMode;
+    private final byte[] expectedToken; // null unless tokenMode
+
+    /** Loopback-trust resolver: no token required, historical behaviour. */
+    public CallerResolver(ConnectionIdentity identity) {
+        this(identity, false, null);
+    }
+
+    /**
+     * @param identity  connection-based worker identification
+     * @param tokenMode when true, a non-worker caller must present a valid bearer token
+     * @param token     the expected bearer token; required (non-blank) when {@code tokenMode}
+     */
+    public CallerResolver(ConnectionIdentity identity, boolean tokenMode, String token) {
+        if (tokenMode && (token == null || token.isBlank())) {
+            throw new IllegalArgumentException(
+                    "auth.mode=token requires a non-empty token; check that the env var named by "
+                            + "auth.tokenEnv is exported to the daemon's environment");
+        }
+        this.identity = identity;
+        this.tokenMode = tokenMode;
+        this.expectedToken = tokenMode ? token.getBytes(StandardCharsets.UTF_8) : null;
+    }
+
+    /**
+     * Resolve the caller of a request.
+     *
+     * @param remoteAddr           the connection's remote address
+     * @param remotePort           the connection's remote port (used for the peer-PID lookup)
+     * @param authorizationHeader  the raw {@code Authorization} header, or {@code null}
+     */
+    public Principal resolve(String remoteAddr, int remotePort, String authorizationHeader) {
+        ConnectionIdentity.Caller c = identity.resolve(remoteAddr, remotePort);
+        if (c.terminal() != null) {
+            return Principal.worker(c.terminal(), c.pid()); // unforgeable; never token-gated
+        }
+
+        if (tokenMode) {
+            return presentedTokenMatches(authorizationHeader)
+                    ? Principal.primary(c.pid())
+                    : Principal.anonymous();
+        }
+
+        // loopback-trust: same-host callers that are not workers are the primary. A non-loopback
+        // caller is anonymous even here — and startup refuses that combination anyway
+        // (BridgedConfig.validateAuthExposure), so this is defence in depth, not the control.
+        return isLoopback(remoteAddr) ? Principal.primary(c.pid()) : Principal.anonymous();
+    }
+
+    /** The working directory of the calling process (CB-112 spawn cwd inheritance), or {@code null}. */
+    public String cwdForPid(long pid) {
+        return identity.cwdForPid(pid);
+    }
+
+    /** True when auth requires a bearer token of non-worker callers. */
+    public boolean tokenMode() {
+        return tokenMode;
+    }
+
+    private boolean presentedTokenMatches(String authorizationHeader) {
+        String presented = bearerValue(authorizationHeader);
+        if (presented == null) {
+            return false;
+        }
+        // Constant-time: MessageDigest.isEqual does not short-circuit on the first differing byte,
+        // so a token cannot be recovered a byte at a time by timing the response.
+        return MessageDigest.isEqual(presented.getBytes(StandardCharsets.UTF_8), expectedToken);
+    }
+
+    /** Extract the credential from {@code Authorization: Bearer <token>}, or {@code null}. */
+    private static String bearerValue(String header) {
+        if (header == null) {
+            return null;
+        }
+        String h = header.trim();
+        if (h.length() < 7 || !h.regionMatches(true, 0, "Bearer ", 0, 7)) {
+            return null;
+        }
+        String token = h.substring(7).trim();
+        return token.isEmpty() ? null : token;
+    }
+
+    private static boolean isLoopback(String remoteAddr) {
+        if (remoteAddr == null) {
+            return false;
+        }
+        return remoteAddr.equals("127.0.0.1") || remoteAddr.equals("::1")
+                || remoteAddr.equals("0:0:0:0:0:0:0:1") || remoteAddr.startsWith("127.");
+    }
+}

bridged/src/main/java/dev/ltms/bridged/auth/Principal.java

@@ -0,0 +1,57 @@
+package dev.ltms.bridged.auth;
+
+/**
+ * A resolved caller: its {@link Role}, and — for a worker — the herdr {@code terminal_id} that
+ * identifies which worker it is (CB-501).
+ *
+ * @param role     what this caller is authorized to act as
+ * @param terminal the worker's herdr terminal id; {@code null} for {@code PRIMARY}/{@code ANONYMOUS}
+ * @param pid      the connecting process id, or {@code -1} when not resolvable (audit context)
+ */
+public record Principal(Role role, String terminal, long pid) {
+
+    /** A caller authenticated as nothing — the default when no check establishes anything else. */
+    public static Principal anonymous() {
+        return new Principal(Role.ANONYMOUS, null, -1);
+    }
+
+    /** The orchestrating session. */
+    public static Principal primary(long pid) {
+        return new Principal(Role.PRIMARY, null, pid);
+    }
+
+    /** A worker peer, identified by its herdr pane. */
+    public static Principal worker(String terminal, long pid) {
+        return new Principal(Role.WORKER, terminal, pid);
+    }
+
+    public boolean isPrimary() {
+        return role == Role.PRIMARY;
+    }
+
+    public boolean isWorker() {
+        return role == Role.WORKER;
+    }
+
+    public boolean isAnonymous() {
+        return role == Role.ANONYMOUS;
+    }
+
+    /**
+     * Whether this caller may act <em>as</em> {@code sessionId} — the "own session only" rule that
+     * keeps one worker from replying or asking on another's behalf. Only a worker can own a
+     * session, and only its own.
+     */
+    public boolean ownsSession(String sessionId) {
+        return isWorker() && terminal != null && terminal.equals(sessionId);
+    }
+
+    /** Short, non-sensitive description for audit lines and error details. */
+    public String describe() {
+        return switch (role) {
+            case WORKER -> "worker:" + terminal;
+            case PRIMARY -> "primary";
+            case ANONYMOUS -> "anonymous";
+        };
+    }
+}

bridged/src/main/java/dev/ltms/bridged/auth/Role.java

@@ -0,0 +1,29 @@
+package dev.ltms.bridged.auth;
+
+/**
+ * What a caller is allowed to be on the bus (CB-501).
+ *
+ * <p>The ordering matters conceptually: {@link #PRIMARY} is the <em>most</em> privileged role
+ * (it spawns, stops, sends to any session, and drains any inbox), not the least. Before CB-501
+ * the daemon reached {@code PRIMARY} by <em>failing</em> every other check — any caller that did
+ * not resolve to a known worker pane was treated as the primary. That is inverted here:
+ * {@link #ANONYMOUS} is the fallback, and {@code PRIMARY} must be established.
+ */
+public enum Role {
+
+    /**
+     * The orchestrating session. Established either by being a loopback caller that is not a
+     * worker pane (under {@code loopback-trust}) or by presenting a valid bearer token (under
+     * {@code token} mode).
+     */
+    PRIMARY,
+
+    /**
+     * A worker peer, identified by its herdr pane. Unforgeable: derived from the connection's
+     * loopback peer PID via herdr's PID→pane map, never from a request argument.
+     */
+    WORKER,
+
+    /** Authenticated as nothing. Authorized for nothing but {@code /healthz}. */
+    ANONYMOUS
+}

bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java

@@ -36,6 +36,8 @@ import java.util.Set;
  * @param primary       optional pinned primary terminal config ({@code null} → derived from connection);
  *                      a non-blank {@code terminal} seeds {@code PrimaryRegistry} and prevents
  *                      connection-derived overrides, CB-307
+ * @param auth          API authentication mode ({@code null} → {@code loopback-trust}, the
+ *                      historical behaviour), CB-501
  */
 @JsonIgnoreProperties(ignoreUnknown = true)
 public record BridgedConfig(
@@ -50,7 +52,8 @@ public record BridgedConfig(
         Integer spawnReadyTimeoutMs,
         Integer spawnReadyPollMs,
         Broker broker,
-        Primary primary) {
+        Primary primary,
+        Auth auth) {
 
     @JsonIgnoreProperties(ignoreUnknown = true)
     public record Bind(String host, int port) {
@@ -256,6 +259,43 @@ public record BridgedConfig(
         }
     }
 
+    /**
+     * API authentication (CB-501). Governs how a caller that is <em>not</em> an on-host worker
+     * pane proves it is the primary.
+     *
+     * <p>Worker identity never depends on this block: a loopback peer PID that maps to a herdr
+     * pane is unforgeable and is always honoured (see
+     * {@link dev.ltms.bridged.mcp.ConnectionIdentity}). This only decides what happens for
+     * <em>everyone else</em>.
+     *
+     * @param mode     {@code "loopback-trust"} (default) — any loopback caller that is not a known
+     *                 worker is the primary, no credential needed; this is the historical
+     *                 behaviour, now chosen explicitly rather than implied. {@code "token"} — such
+     *                 a caller must present {@code Authorization: Bearer <token>} or it is
+     *                 {@code ANONYMOUS} and authorized for nothing.
+     * @param tokenEnv name of the host env var holding the bearer token; the literal value is
+     *                 never stored in config. Defaults to {@code BRIDGED_API_TOKEN}. Only read
+     *                 when {@code mode} is {@code token}.
+     */
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public record Auth(String mode, String tokenEnv) {
+
+        /** Historical behaviour: loopback non-worker ⇒ primary, no credential. */
+        public static final String MODE_LOOPBACK_TRUST = "loopback-trust";
+        /** A non-worker caller must present a valid bearer token to be the primary. */
+        public static final String MODE_TOKEN = "token";
+
+        public Auth {
+            mode = (mode == null || mode.isBlank()) ? MODE_LOOPBACK_TRUST : mode.toLowerCase();
+            tokenEnv = (tokenEnv == null || tokenEnv.isBlank()) ? "BRIDGED_API_TOKEN" : tokenEnv;
+        }
+
+        /** True when a bearer token is required of every non-worker caller. */
+        public boolean tokenMode() {
+            return MODE_TOKEN.equals(mode);
+        }
+    }
+
     /**
      * Subscription boundary. Only these hosts may back a worker's
      * {@code ANTHROPIC_BASE_URL}; the primary must carry none.
@@ -326,8 +366,43 @@ public record BridgedConfig(
         Lifecycle l = lifecycle != null ? lifecycle : new Lifecycle(null, null, null);
         Integer timeout = (spawnReadyTimeoutMs != null) ? spawnReadyTimeoutMs : 20000;
         Integer pollMs = (spawnReadyPollMs != null) ? spawnReadyPollMs : 300;
+        Auth a = auth != null ? auth : new Auth(null, null);
         // broker is left as-is: null (or an empty/blank uri) keeps the in-memory soft-state inbox.
         // primary is left as-is: null defaults to connection-derived identity.
-        return new BridgedConfig(b, herdrSocket, worker, workers, defaultWorker, g, worktreeRoot, l, timeout, pollMs, broker, primary);
+        return new BridgedConfig(b, herdrSocket, worker, workers, defaultWorker, g, worktreeRoot, l, timeout, pollMs, broker, primary, a);
+    }
+
+    /**
+     * Reject a configuration whose network exposure outruns its authentication (CB-501).
+     *
+     * <p>{@code loopback-trust} means "any caller that is not a known worker pane is the primary" —
+     * safe only because the OS refuses non-local connections to a loopback bind. Widen
+     * {@code bind.host} without switching to {@code token} mode and that sentence becomes "any
+     * client that can reach this port is the primary", which is the most privileged role on the
+     * bus. Rather than document the hazard, make it unrepresentable: fail fast at startup.
+     *
+     * @throws IllegalStateException when a non-loopback bind is paired with {@code loopback-trust}
+     */
+    public void validateAuthExposure() {
+        String host = bind().host();
+        if (isLoopbackBind(host) || auth().tokenMode()) {
+            return;
+        }
+        throw new IllegalStateException(
+                "refusing to start: bind.host=" + host + " is not loopback, but auth.mode="
+                        + auth().mode() + ". A non-loopback bind treats every unauthenticated "
+                        + "caller as the primary (spawn/stop/send/drain on any session). Set "
+                        + "auth.mode: token (with auth.tokenEnv) before exposing this port, or "
+                        + "bind to 127.0.0.1 and put a reverse proxy in front.");
+    }
+
+    /** True for the loopback addresses and the unspecified-but-local forms we treat as same-host. */
+    private static boolean isLoopbackBind(String host) {
+        if (host == null || host.isBlank()) {
+            return true; // Bind's own default is 127.0.0.1
+        }
+        String h = host.trim().toLowerCase();
+        return h.equals("127.0.0.1") || h.equals("::1") || h.equals("localhost")
+                || h.startsWith("127.");
     }
 }

bridged/src/main/java/dev/ltms/bridged/mcp/BridgeMcp.java

@@ -1,7 +1,14 @@
 package dev.ltms.bridged.mcp;
 
+import dev.ltms.bridged.auth.AuditLog;
+import dev.ltms.bridged.auth.Authz;
+import dev.ltms.bridged.auth.CallerResolver;
+import dev.ltms.bridged.auth.Principal;
+import dev.ltms.bridged.auth.Role;
 import dev.ltms.bridged.guard.GuardException;
 import dev.ltms.bridged.herdr.Agent;
+import dev.ltms.bridged.metrics.BridgedMetrics;
+import dev.ltms.bridged.metrics.Metrics;
 import dev.ltms.bridged.inject.WorkerPresence;
 import dev.ltms.bridged.herdr.HerdrException;
 import dev.ltms.bridged.msg.MessageService;
@@ -56,13 +63,34 @@ public final class BridgeMcp {
     static final String CALLER_TERMINAL = "callerTerminal";
     /** Transport-context key under which the extractor stashes the caller's PID (for cwd inherit). */
     static final String CALLER_PID = "callerPid";
+    /** Transport-context key under which the extractor stashes the resolved {@link Role} (CB-501). */
+    static final String CALLER_ROLE = "callerRole";
 
     private final HttpServletStreamableServerTransportProvider transport;
     private final McpSyncServer server;
+    private final CallerResolver authz; // CB-501: null → authorization not enforced (legacy)
+    private final Metrics metrics;      // CB-502: null → auth failures not counted
 
+    /**
+     * Legacy constructor — no authorization. Retained so existing tests exercise tool behaviour
+     * without an auth fixture.
+     */
     public BridgeMcp(MessageService messages, PeerLauncher workers,
                      SessionManager sessions, ConnectionIdentity identity, WorkerPresence presence,
                      PrimaryRegistry primaryRegistry) {
+        this(messages, workers, sessions, identity, presence, primaryRegistry, null, null);
+    }
+
+    /**
+     * @param callers resolves each call's {@link Principal}; {@code null} disables authorization.
+     *                This surface needs its own enforcement: {@code /mcp} is a raw servlet on
+     *                Jetty's context handler and never passes through Javalin's {@code before}
+     *                filter, so the REST guard does not cover it.
+     * @param metrics registry for auth-failure counting; may be {@code null}
+     */
+    public BridgeMcp(MessageService messages, PeerLauncher workers,
+                     SessionManager sessions, ConnectionIdentity identity, WorkerPresence presence,
+                     PrimaryRegistry primaryRegistry, CallerResolver callers, Metrics metrics) {
         McpJsonMapper json = new JacksonMcpJsonMapperSupplier().get();
         this.transport = HttpServletStreamableServerTransportProvider.builder()
                 .jsonMapper(json)
@@ -72,17 +100,26 @@ public final class BridgeMcp {
                 // inherit the primary's cwd (CB-112). Any contact from a worker marks it available
                 // (CB-113) — its MCP initialize is the reliable "the agent is up" signal.
                 .contextExtractor(req -> {
-                    ConnectionIdentity.Caller c = identity.resolve(req.getRemoteAddr(), req.getRemotePort());
+                    // One resolution per call, shared with the REST surface via CallerResolver so
-                    presence.markPresent(c.terminal()); // no-op for the primary (null terminal)
+                    // the two paths cannot drift on who a caller is.
+                    Principal p = callers != null
+                            ? callers.resolve(req.getRemoteAddr(), req.getRemotePort(),
+                                    req.getHeader("Authorization"))
+                            : legacyPrincipal(identity, req.getRemoteAddr(), req.getRemotePort());
+                    presence.markPresent(p.terminal()); // no-op for the primary (null terminal)
                     return McpTransportContext.create(Map.of(
-                            CALLER_TERMINAL, orEmpty(c.terminal()),
+                            CALLER_TERMINAL, orEmpty(p.terminal()),
-                            CALLER_PID, Long.toString(c.pid())));
+                            CALLER_PID, Long.toString(p.pid()),
+                            CALLER_ROLE, p.role().name()));
                 })
                 .build();
         this.server = McpServer.sync(transport)
                 .serverInfo("bridge", "0.1.0")
                 .capabilities(McpSchema.ServerCapabilities.builder().tools(true).build())
                 .toolCall(sendTool(), (exchange, req) -> {
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.SEND,
+                            str(req.arguments(), "sessionId"));
+                    if (denied != null) return denied;
                     String caller = callerTerminal(exchange);
                     if (caller != null) primaryRegistry.record(caller);
                     Map<String, Object> a = req.arguments();
@@ -97,25 +134,44 @@ public final class BridgeMcp {
                             ? sendAsync(messages, str(a, "sessionId"), str(a, "content"))
                             : send(messages, str(a, "sessionId"), str(a, "content"), timeoutMs(a));
                 })
-                // bridge_reply's identity is the CONNECTION, never an argument.
+                // bridge_reply's identity is the CONNECTION, never an argument — so the authz check
-                .toolCall(replyTool(), (exchange, req) ->
+                // is "is this caller a worker at all", and it can only ever reply as itself.
-                        reply(messages, callerTerminal(exchange), str(req.arguments(), "content")))
+                .toolCall(replyTool(), (exchange, req) -> {
+                    String self = callerTerminal(exchange);
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.REPLY, self);
+                    if (denied != null) return denied;
+                    return reply(messages, self, str(req.arguments(), "content"));
+                })
                 // bridge_ask (CB-205): a worker's mid-turn question — identity from the CONNECTION.
-                .toolCall(askTool(), (exchange, req) ->
+                .toolCall(askTool(), (exchange, req) -> {
-                        ask(messages, callerTerminal(exchange), str(req.arguments(), "question"), timeoutMs(req.arguments())))
+                    String self = callerTerminal(exchange);
-                .toolCall(statusTool(), (_, req) ->
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.ASK, self);
-                        status(messages, str(req.arguments(), "sessionId")))
+                    if (denied != null) return denied;
-                .toolCall(pollTool(), (_, req) -> {
+                    return ask(messages, self, str(req.arguments(), "question"), timeoutMs(req.arguments()));
+                })
+                .toolCall(statusTool(), (exchange, req) -> {
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
+                    if (denied != null) return denied;
+                    return status(messages, str(req.arguments(), "sessionId"));
+                })
+                .toolCall(pollTool(), (exchange, req) -> {
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
+                    if (denied != null) return denied;
                     Map<String, Object> a = req.arguments();
                     return poll(messages, str(a, "ticket"), str(a, "target"));
                 })
                 // CB-307 Increment 3: per-msgId ack (not needed in v1 but supported by the inbox).
-                .toolCall(ackTool(), (_, req) -> {
+                // Acking removes a reply from the inbox, so it is a drain, not a read.
+                .toolCall(ackTool(), (exchange, req) -> {
                     Map<String, Object> a = req.arguments();
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.DRAIN, str(a, "target"));
+                    if (denied != null) return denied;
                     return ack(messages, str(a, "target"), str(a, "msgId"));
                 })
                 // Fleet management (CB-108): spawn/list/stop over ClaudeCodeLauncher.
                 .toolCall(spawnTool(), (exchange, req) -> {
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.SPAWN, null);
+                    if (denied != null) return denied;
                     String caller = callerTerminal(exchange);
                     if (caller != null) primaryRegistry.record(caller);
                     Map<String, Object> a = req.arguments();
@@ -126,10 +182,72 @@ public final class BridgeMcp {
                     return spawn(sessions, str(a, "profile"), str(a, "cwd"), callerCwd,
                             callerTerminal(exchange), worktreeRequest(a));
                 })
-                .toolCall(listTool(), (_, _) -> listWorkers(workers, sessions))
+                .toolCall(listTool(), (exchange, _) -> {
-                .toolCall(stopTool(), (_, req) -> stop(sessions, str(req.arguments(), "paneId")))
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
-                .toolCall(profilesTool(), (_, _) -> profiles(workers))
+                    if (denied != null) return denied;
+                    return listWorkers(workers, sessions);
+                })
+                .toolCall(stopTool(), (exchange, req) -> {
+                    String paneId = str(req.arguments(), "paneId");
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.STOP, paneId);
+                    if (denied != null) return denied;
+                    return stop(sessions, paneId);
+                })
+                .toolCall(profilesTool(), (exchange, _) -> {
+                    McpSchema.CallToolResult denied = deny(exchange, Authz.Action.READ, null);
+                    if (denied != null) return denied;
+                    return profiles(workers);
+                })
                 .build();
+        this.authz = callers;
+        this.metrics = metrics;
+    }
+
+    /**
+     * Pre-CB-501 identity: worker if the connection maps to a pane, otherwise the primary. Used
+     * only by the legacy constructor, where authorization is not enforced anyway.
+     */
+    private static Principal legacyPrincipal(ConnectionIdentity identity, String addr, int port) {
+        ConnectionIdentity.Caller c = identity.resolve(addr, port);
+        return c.terminal() != null
+                ? Principal.worker(c.terminal(), c.pid())
+                : Principal.primary(c.pid());
+    }
+
+    /** The caller reconstructed from the transport context. */
+    private static Principal principal(McpSyncServerExchange exchange) {
+        Object r = exchange.transportContext().get(CALLER_ROLE);
+        String terminal = callerTerminal(exchange);
+        long pid = callerPid(exchange);
+        if (r == null) {
+            // No role stashed (legacy path): fall back to the historical interpretation.
+            return terminal != null ? Principal.worker(terminal, pid) : Principal.primary(pid);
+        }
+        return new Principal(Role.valueOf(r.toString()), terminal, pid);
+    }
+
+    /**
+     * Gate a tool call on the CB-505 table. Returns {@code null} when the call may proceed, or the
+     * error result to return when it may not.
+     */
+    private McpSchema.CallToolResult deny(McpSyncServerExchange exchange, Authz.Action action,
+                                          String target) {
+        if (authz == null) {
+            return null; // legacy: authorization not enforced
+        }
+        Principal caller = principal(exchange);
+        if (Authz.permits(caller, action, target)) {
+            if (action != Authz.Action.READ) {
+                AuditLog.allowed(caller, action, target);
+            }
+            return null;
+        }
+        String reason = Authz.isUnauthenticated(caller) ? "unauthenticated" : "forbidden";
+        AuditLog.denied(caller, action, target, reason);
+        if (metrics != null) {
+            metrics.inc(BridgedMetrics.AUTH_FAILURES, "reason", reason);
+        }
+        return error(reason + ": " + caller.describe() + " may not " + action);
     }
 
     /** The worker identity resolved from this call's connection, or {@code null} if the primary. */

bridged/src/main/java/dev/ltms/bridged/metrics/BridgedMetrics.java

@@ -0,0 +1,96 @@
+package dev.ltms.bridged.metrics;
+
+import dev.ltms.bridged.msg.ReplyInbox;
+import dev.ltms.bridged.session.SessionManager;
+import dev.ltms.bridged.session.WorkerSession;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * The daemon's metric definitions (CB-502) — one place where every series is named, described, and
+ * (for gauges) bound to live state.
+ *
+ * <p>The set is deliberately small: each series maps to a failure mode this project has actually
+ * hit, not to whatever was easy to count. The two worth watching in practice are
+ * {@code bridged_sends_total{outcome="completion_fallback"}} — a rising share means turn detection
+ * is degrading, the CB-115/116/118 failure family — and
+ * {@code bridged_push_nudges_total{outcome="exhausted"}}, which means the primary stopped draining
+ * its inbox and CB-307's active push gave up.
+ */
+public final class BridgedMetrics {
+
+    /** Counter: delegated sends by terminal outcome. */
+    public static final String SENDS = "bridged_sends_total";
+    /** Counter: worker replies by the path that carried them (rendezvous vs stranded-to-inbox). */
+    public static final String REPLIES = "bridged_replies_total";
+    /** Counter: push-loop nudges to the primary, by outcome. */
+    public static final String PUSH_NUDGES = "bridged_push_nudges_total";
+    /** Counter: spawn attempts by peer kind and outcome. */
+    public static final String SPAWNS = "bridged_spawns_total";
+    /** Counter: herdr socket calls by method and outcome. */
+    public static final String HERDR_CALLS = "bridged_herdr_calls_total";
+    /** Counter: rejected requests by reason (CB-501). */
+    public static final String AUTH_FAILURES = "bridged_auth_failures_total";
+    /** Gauge: session census by lifecycle state. */
+    public static final String SESSIONS = "bridged_sessions";
+    /** Gauge: undrained replies held per target. */
+    public static final String INBOX_DEPTH = "bridged_inbox_depth";
+
+    private BridgedMetrics() {
+    }
+
+    /**
+     * Build the registry with its help text and live gauges bound.
+     *
+     * @param sessions the authoritative session registry (census gauge)
+     * @param inbox    the reply inbox; only used for a depth gauge when it can be inspected
+     */
+    public static Metrics create(SessionManager sessions, ReplyInbox inbox) {
+        Metrics m = new Metrics();
+
+        m.describe(SENDS, "counter",
+                "Delegated sends by terminal outcome (replied|completion_fallback|timeout|failed).");
+        m.describe(REPLIES, "counter",
+                "Worker replies by delivery path (rendezvous=resolved an open send, inbox=stranded and held).");
+        m.describe(PUSH_NUDGES, "counter",
+                "CB-307 push-loop nudges to the primary (delivered|exhausted).");
+        m.describe(SPAWNS, "counter",
+                "Worker spawn attempts by peer kind and outcome (ready|timeout|guard_rejected).");
+        m.describe(HERDR_CALLS, "counter",
+                "herdr socket calls by method and outcome — the dependency everything else rests on.");
+        m.describe(AUTH_FAILURES, "counter",
+                "Requests refused by CB-501/505 (unauthenticated|forbidden).");
+        m.describe(SESSIONS, "gauge",
+                "Registered worker sessions by lifecycle state.");
+        m.describe(INBOX_DEPTH, "gauge",
+                "Replies held for a target that the primary has not drained. Steady state is 0; "
+                        + "a target stuck above 0 means CB-307 delivery is not completing.");
+
+        // One gauge per state so a scrape shows the whole census even when a state is empty —
+        // an absent series and a zero series read very differently on a dashboard.
+        for (WorkerSession.State state : WorkerSession.State.values()) {
+            String label = state.name().toLowerCase();
+            m.gauge(SESSIONS, () -> countIn(sessions, state), "state", label);
+        }
+
+        // Depth is per live session, so the label set is only known at scrape time. peek() is the
+        // port's non-destructive read — scraping metrics must never ack a reply out of the inbox.
+        m.collector(INBOX_DEPTH, "target", () -> {
+            Map<String, Number> depths = new LinkedHashMap<>();
+            for (WorkerSession s : sessions.roster()) {
+                String target = s.terminalId();
+                if (target == null) {
+                    continue;
+                }
+                depths.put(target, inbox.peek(target).size());
+            }
+            return depths;
+        });
+        return m;
+    }
+
+    private static long countIn(SessionManager sessions, WorkerSession.State state) {
+        return sessions.roster().stream().filter(s -> s.state() == state).count();
+    }
+}

bridged/src/main/java/dev/ltms/bridged/metrics/Metrics.java

@@ -0,0 +1,181 @@
+package dev.ltms.bridged.metrics;
+
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Supplier;
+
+/**
+ * The daemon's metric registry and Prometheus text renderer (CB-502).
+ *
+ * <p>Deliberately dependency-free. The roadmap's tech-stack table specified Micrometer, but this
+ * pom already carries an unusual reconciliation burden (a hand-pinned {@code jackson-annotations}
+ * to make the MCP SDK's Jackson 3 coexist with our Jackson 2, a Jetty BOM import to stop version
+ * skew, and four documented accepted-CVE advisories), and the dependency CVE gate this project
+ * mandates could not be run when this landed. The metric set is small and fully known, and
+ * Prometheus text exposition is a stable, well-specified format — so the registry is ~100 lines
+ * here instead of a new transitive tree. {@code GET /metrics} is the swap seam if Micrometer's
+ * ecosystem is ever wanted.
+ *
+ * <p>Thread-safe: counters are {@link LongAdder} (built for contended increment), gauges are
+ * supplier-backed so they read live state at scrape time rather than needing to be pushed.
+ */
+public final class Metrics {
+
+    /** Counter series, keyed by the fully-rendered {@code name{labels}} sample id. */
+    private final NavigableMap<String, LongAdder> counters = new ConcurrentSkipListMap<>();
+    /** Gauge series, evaluated at scrape time. */
+    private final NavigableMap<String, Supplier<Number>> gauges = new ConcurrentSkipListMap<>();
+    /** Gauge families whose label set is only known at scrape time, keyed by metric name. */
+    private final NavigableMap<String, Collector> collectors = new ConcurrentSkipListMap<>();
+    /** HELP/TYPE metadata, keyed by bare metric name. */
+    private final Map<String, String[]> meta = new ConcurrentHashMap<>();
+
+    /** A gauge family whose series are discovered per scrape (one label, many values). */
+    private record Collector(String labelName, Supplier<Map<String, Number>> samples) {
+    }
+
+    /** Declare a metric's help text and type once, so the exposition carries HELP/TYPE lines. */
+    public Metrics describe(String name, String type, String help) {
+        meta.put(name, new String[]{type, help});
+        return this;
+    }
+
+    /** Increment a counter by one. */
+    public void inc(String name, String... labelPairs) {
+        add(name, 1, labelPairs);
+    }
+
+    /** Increment a counter by {@code delta}. */
+    public void add(String name, long delta, String... labelPairs) {
+        counters.computeIfAbsent(sample(name, labelPairs), _ -> new LongAdder()).add(delta);
+    }
+
+    /**
+     * Register a live gauge. The supplier is called at scrape time, so it reflects current state
+     * (session census, inbox depth) without anything having to remember to update it.
+     */
+    public void gauge(String name, Supplier<Number> value, String... labelPairs) {
+        gauges.put(sample(name, labelPairs), value);
+    }
+
+    /**
+     * Register a gauge family whose label values are not known up front — inbox depth per target,
+     * for instance, where the set of targets changes as workers come and go. The supplier returns
+     * {@code labelValue → value} and is evaluated once per scrape.
+     */
+    public void collector(String name, String labelName, Supplier<Map<String, Number>> samples) {
+        collectors.put(name, new Collector(labelName, samples));
+    }
+
+    /** Current value of a counter series — for assertions in tests. */
+    public long count(String name, String... labelPairs) {
+        LongAdder a = counters.get(sample(name, labelPairs));
+        return a == null ? 0 : a.sum();
+    }
+
+    /**
+     * Render the Prometheus text exposition format (version 0.0.4): optional {@code # HELP} and
+     * {@code # TYPE} lines per metric family, then one line per sample.
+     */
+    public String render() {
+        StringBuilder out = new StringBuilder(1024);
+        String lastFamily = null;
+        for (Map.Entry<String, LongAdder> e : counters.entrySet()) {
+            lastFamily = emitHeader(out, e.getKey(), lastFamily);
+            out.append(e.getKey()).append(' ').append(e.getValue().sum()).append('\n');
+        }
+        for (Map.Entry<String, Supplier<Number>> e : gauges.entrySet()) {
+            lastFamily = emitHeader(out, e.getKey(), lastFamily);
+            Number v;
+            try {
+                v = e.getValue().get();
+            } catch (RuntimeException ex) {
+                continue; // a broken gauge must never break the whole scrape
+            }
+            if (v == null) {
+                continue;
+            }
+            out.append(e.getKey()).append(' ').append(format(v)).append('\n');
+        }
+        for (Map.Entry<String, Collector> e : collectors.entrySet()) {
+            Map<String, Number> samples;
+            try {
+                samples = e.getValue().samples().get();
+            } catch (RuntimeException ex) {
+                continue; // a broken collector must never break the whole scrape
+            }
+            if (samples == null || samples.isEmpty()) {
+                continue;
+            }
+            lastFamily = emitHeader(out, e.getKey(), lastFamily);
+            // Sort so repeated scrapes are byte-stable and diffable.
+            new java.util.TreeMap<>(samples).forEach((label, v) -> {
+                if (v != null) {
+                    out.append(sample(e.getKey(), e.getValue().labelName(), label))
+                            .append(' ').append(format(v)).append('\n');
+                }
+            });
+        }
+        return out.toString();
+    }
+
+    /** Emit HELP/TYPE when the sample starts a new metric family; returns the current family. */
+    private String emitHeader(StringBuilder out, String sampleId, String lastFamily) {
+        String family = familyOf(sampleId);
+        if (family.equals(lastFamily)) {
+            return lastFamily;
+        }
+        String[] m = meta.get(family);
+        if (m != null) {
+            out.append("# HELP ").append(family).append(' ').append(m[1]).append('\n');
+            out.append("# TYPE ").append(family).append(' ').append(m[0]).append('\n');
+        }
+        return family;
+    }
+
+    private static String familyOf(String sampleId) {
+        int brace = sampleId.indexOf('{');
+        return brace < 0 ? sampleId : sampleId.substring(0, brace);
+    }
+
+    /** Whole numbers render without a decimal point; everything else as-is. */
+    private static String format(Number v) {
+        double d = v.doubleValue();
+        return (d == Math.rint(d) && !Double.isInfinite(d))
+                ? Long.toString((long) d)
+                : Double.toString(d);
+    }
+
+    /** Build the {@code name{k="v",k2="v2"}} sample id; labels are sorted for stable output. */
+    private static String sample(String name, String... labelPairs) {
+        if (labelPairs == null || labelPairs.length == 0) {
+            return name;
+        }
+        if (labelPairs.length % 2 != 0) {
+            throw new IllegalArgumentException("labels must be key/value pairs, got " + labelPairs.length);
+        }
+        NavigableMap<String, String> sorted = new java.util.TreeMap<>();
+        for (int i = 0; i < labelPairs.length; i += 2) {
+            sorted.put(labelPairs[i], labelPairs[i + 1] == null ? "" : labelPairs[i + 1]);
+        }
+        StringBuilder sb = new StringBuilder(name.length() + 16 * sorted.size());
+        sb.append(name).append('{');
+        boolean first = true;
+        for (Map.Entry<String, String> e : sorted.entrySet()) {
+            if (!first) {
+                sb.append(',');
+            }
+            first = false;
+            sb.append(e.getKey()).append("=\"").append(escapeLabel(e.getValue())).append('"');
+        }
+        return sb.append('}').toString();
+    }
+
+    /** Label values are escaped per the exposition format: backslash, quote, newline. */
+    private static String escapeLabel(String v) {
+        return v.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n");
+    }
+}

bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java

@@ -3,6 +3,8 @@ package dev.ltms.bridged.msg;
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.AgentStatus;
 import dev.ltms.bridged.inject.Injector;
+import dev.ltms.bridged.metrics.BridgedMetrics;
+import dev.ltms.bridged.metrics.Metrics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -151,6 +153,7 @@ public final class MessageService {
     private final Rendezvous rendezvous;
     private final ReplyInbox inbox;
     private final ReplyPushLoop pushLoop;
+    private final Metrics metrics; // CB-502: nullable — no registry in unit tests
     private final ConcurrentHashMap<String, ReentrantLock> sessionLocks = new ConcurrentHashMap<>();
     private final ConcurrentHashMap<String, Task> tasks = new ConcurrentHashMap<>();
     private final AtomicLong ticketSeq = new AtomicLong();
@@ -165,11 +168,23 @@ public final class MessageService {
      */
     public MessageService(AgentControl agents, Injector injector, Rendezvous rendezvous,
                           ReplyInbox inbox, ReplyPushLoop pushLoop) {
+        this(agents, injector, rendezvous, inbox, pushLoop, null);
+    }
+
+    /**
+     * As above, with a metric registry (CB-502). Instrumenting here rather than at the REST and MCP
+     * edges means both surfaces are counted by one piece of code and cannot drift.
+     *
+     * @param metrics nullable — when null, nothing is recorded
+     */
+    public MessageService(AgentControl agents, Injector injector, Rendezvous rendezvous,
+                          ReplyInbox inbox, ReplyPushLoop pushLoop, Metrics metrics) {
         this.agents = agents;
         this.injector = injector;
         this.rendezvous = rendezvous;
         this.inbox = inbox;
         this.pushLoop = pushLoop;
+        this.metrics = metrics;
     }
 
     /** Create with an explicit {@link ReplyInbox} and no push loop. */
@@ -200,15 +215,46 @@ public final class MessageService {
      */
     public boolean reply(String session, String content) {
         if (rendezvous.resolve(session, content)) {
+            count(BridgedMetrics.REPLIES, "path", "rendezvous");
             return true;                       // a live send took it — unchanged fast path
         }
         inbox.publish(session, UUID.randomUUID().toString(), content);
+        // A rising inbox share is the signal CB-307 exists to make visible: the worker finished but
+        // nobody was waiting, so delivery now depends on the push loop and a drain.
+        count(BridgedMetrics.REPLIES, "path", "inbox");
         if (pushLoop != null) {
             pushLoop.onReplyQueued(session);
         }
         return true;                           // held, not lost
     }
 
+    /** Record a counter sample when a registry is wired; a no-op in unit tests. */
+    private void count(String name, String... labels) {
+        if (metrics != null) {
+            metrics.inc(name, labels);
+        }
+    }
+
+    /** Count a send's terminal outcome and pass the reply through unchanged. */
+    private Reply recorded(Reply r) {
+        String label = sendOutcomeLabel(r.outcome());
+        if (label != null) {
+            count(BridgedMetrics.SENDS, "outcome", label);
+        }
+        return r;
+    }
+
+    /** Map a terminal send outcome to its metric label, or {@code null} for non-terminal ones. */
+    private static String sendOutcomeLabel(Outcome o) {
+        return switch (o) {
+            case REPLIED -> "replied";
+            case COMPLETED_UNREPLIED -> "completion_fallback";
+            case TIMED_OUT_WORKING, TIMED_OUT_QUEUED, BUSY -> "timeout";
+            case WORKER_FAILED -> "failed";
+            case STALE_TURN, QUESTION -> null; // not a completed delegation
+        };
+    }
+
     /**
      * Acknowledge a specific reply by {@code msgId} for {@code target}. Removes it from the inbox
      * so that a subsequent drain or peek no longer returns it.
@@ -248,11 +294,12 @@ public final class MessageService {
             CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
             try {
                 Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
-                return new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
+                return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
             } catch (TimeoutException e) {
                 boolean wasDelivered = delivered.isDone() && !delivered.isCompletedExceptionally();
                 log.debug("send to {} timed out (delivered={})", target, wasDelivered);
-                return new Reply(wasDelivered ? Outcome.TIMED_OUT_WORKING : Outcome.TIMED_OUT_QUEUED, null);
+                return recorded(new Reply(
+                        wasDelivered ? Outcome.TIMED_OUT_WORKING : Outcome.TIMED_OUT_QUEUED, null));
             } catch (ExecutionException e) {
                 Throwable cause = e.getCause();
                 throw cause instanceof RuntimeException re ? re : new IllegalStateException(cause);

bridged/src/main/java/dev/ltms/bridged/rest/BridgedApp.java

@@ -2,7 +2,12 @@ package dev.ltms.bridged.rest;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import dev.ltms.bridged.auth.AuditLog;
+import dev.ltms.bridged.auth.Authz;
+import dev.ltms.bridged.auth.CallerResolver;
+import dev.ltms.bridged.auth.Principal;
 import dev.ltms.bridged.guard.GuardException;
+import dev.ltms.bridged.metrics.Metrics;
 import dev.ltms.bridged.herdr.Agent;
 import dev.ltms.bridged.herdr.HerdrClient;
 import dev.ltms.bridged.herdr.HerdrException;
@@ -43,23 +48,47 @@ public final class BridgedApp {
     private static final long DEFAULT_ASK_TIMEOUT_MS = 55_000;
     private static final long MAX_ASK_TIMEOUT_MS = 115_000;
 
+    /** Context attribute under which the resolved caller is stashed by the auth filter. */
+    private static final String CALLER = "bridged.caller";
+
     private final HerdrClient herdr;
     private final PeerLauncher workers;
     private final SessionManager sessions; // CB-301: authoritative session registry
     private final MessageService messages;
     private final WorkerPresence presence; // CB-113: which workers are MCP-connected (available)
     private final HttpServlet mcpServlet; // MCP Streamable-HTTP endpoint, mounted at /mcp (nullable)
+    private final CallerResolver auth;    // CB-501: null → authz not enforced (legacy behaviour)
+    private final Metrics metrics;        // CB-502: null → /metrics not exposed
     private final ObjectMapper mapper = new ObjectMapper();
 
+    /**
+     * Legacy constructor — no identity resolution and no authorization, exactly as the REST surface
+     * behaved before CB-501. Retained so existing acceptance tests keep exercising handler
+     * behaviour without each needing an auth fixture.
+     */
     public BridgedApp(HerdrClient herdr, PeerLauncher workers, SessionManager sessions,
                       MessageService messages, WorkerPresence presence,
                       HttpServlet mcpServlet) {
+        this(herdr, workers, sessions, messages, presence, mcpServlet, null, null);
+    }
+
+    /**
+     * @param auth    resolves each request's {@link Principal}; {@code null} disables authorization
+     *                entirely (legacy). {@code main} always supplies one.
+     * @param metrics registry to instrument and expose at {@code GET /metrics}; {@code null} omits
+     *                the endpoint
+     */
+    public BridgedApp(HerdrClient herdr, PeerLauncher workers, SessionManager sessions,
+                      MessageService messages, WorkerPresence presence,
+                      HttpServlet mcpServlet, CallerResolver auth, Metrics metrics) {
         this.herdr = herdr;
         this.workers = workers;
         this.sessions = sessions;
         this.messages = messages;
         this.presence = presence;
         this.mcpServlet = mcpServlet;
+        this.auth = auth;
+        this.metrics = metrics;
     }
 
     /** Wire routes onto a fresh, unstarted Javalin instance. Caller starts it. */
@@ -72,7 +101,18 @@ public final class BridgedApp {
                         h.addServlet(new ServletHolder(mcpServlet), "/mcp"));
             }
         });
+        // CB-501: resolve identity once per request, before any handler. /mcp does NOT pass through
+        // here — it is a raw servlet on Jetty's context handler — so BridgeMcp enforces separately
+        // against the same CallerResolver. Any check that lives in only one place is not a control.
+        if (auth != null) {
+            app.before(ctx -> ctx.attribute(CALLER,
+                    auth.resolve(ctx.req().getRemoteAddr(), ctx.req().getRemotePort(),
+                            ctx.header("Authorization"))));
+        }
         app.get("/healthz", this::healthz);
+        if (metrics != null) {
+            app.get("/metrics", this::metrics);
+        }
         app.get("/sessions", this::sessions);
         app.get("/agents", this::agents);
         app.get("/workers", this::listWorkers);                 // CB-304: registry roster + live herdr status
@@ -88,6 +128,54 @@ public final class BridgedApp {
         return app;
     }
 
+    /**
+     * Gate a handler on the CB-505 authorization table. Returns {@code true} when the request may
+     * proceed; otherwise writes the error response and returns {@code false}.
+     *
+     * <p>401 vs 403 is a real distinction here: 401 means "you presented no usable identity" (a
+     * credential problem the caller can fix), 403 means "you are authenticated, but this is not
+     * yours" (a worker reaching for another worker's session, or for orchestration).
+     */
+    private boolean allow(Context ctx, Authz.Action action, String target) {
+        if (auth == null) {
+            return true; // legacy: authorization not enforced
+        }
+        Principal caller = ctx.attribute(CALLER);
+        if (Authz.permits(caller, action, target)) {
+            if (action != Authz.Action.READ && action != Authz.Action.METRICS) {
+                AuditLog.allowed(caller, action, target); // reads would drown the trail
+            }
+            return true;
+        }
+        if (Authz.isUnauthenticated(caller)) {
+            AuditLog.denied(caller, action, target, "unauthenticated");
+            countAuthFailure("unauthenticated");
+            ctx.status(401).json(Map.of("error", "unauthenticated",
+                    "detail", "present Authorization: Bearer <token>"));
+        } else {
+            AuditLog.denied(caller, action, target, "forbidden");
+            countAuthFailure("forbidden");
+            ctx.status(403).json(Map.of("error", "forbidden",
+                    "detail", caller.describe() + " may not " + action + " on "
+                            + (target == null ? "this resource" : target)));
+        }
+        return false;
+    }
+
+    private void countAuthFailure(String reason) {
+        if (metrics != null) {
+            metrics.inc("bridged_auth_failures_total", "reason", reason);
+        }
+    }
+
+    /** Prometheus scrape endpoint (CB-502). */
+    private void metrics(Context ctx) {
+        if (!allow(ctx, Authz.Action.METRICS, null)) {
+            return;
+        }
+        ctx.status(200).contentType("text/plain; version=0.0.4; charset=utf-8").result(metrics.render());
+    }
+
     /** Liveness + herdr reachability. 200 when herdr answers ping, 503 otherwise. */
     private void healthz(Context ctx) {
         try {
@@ -107,6 +195,9 @@ public final class BridgedApp {
 
     /** Sessions view derived from herdr {@code workspace.list} (one workspace → one row). */
     private void sessions(Context ctx) {
+        if (!allow(ctx, Authz.Action.READ, null)) {
+            return;
+        }
         JsonNode result = herdr.call("workspace.list");
         List<Map<String, Object>> out = new ArrayList<>();
         for (JsonNode w : result.path("workspaces")) {
@@ -122,12 +213,18 @@ public final class BridgedApp {
 
     /** Discovery: every agent herdr tracks, keyed by its Claude session UUID. */
     private void agents(Context ctx) {
+        if (!allow(ctx, Authz.Action.READ, null)) {
+            return;
+        }
         ctx.status(200).json(Map.of("agents",
                 workers.list().stream().map(Agent.class::cast).map(BridgedApp::view).toList()));
     }
 
     /** CB-304: bridge-owned roster merged with live herdr status by paneId. */
     private void listWorkers(Context ctx) {
+        if (!allow(ctx, Authz.Action.READ, null)) {
+            return;
+        }
         Map<String, Agent> live = workers.list().stream()
                 .map(Agent.class::cast)
                 .filter(a -> a.paneId() != null)
@@ -140,6 +237,9 @@ public final class BridgedApp {
 
     /** The configured worker profiles and which one a no-argument spawn uses. */
     private void profiles(Context ctx) {
+        if (!allow(ctx, Authz.Action.READ, null)) {
+            return;
+        }
         ctx.status(200).json(Map.of(
                 "profiles", workers.profiles(),
                 "default", workers.defaultProfile() == null ? "" : workers.defaultProfile()));
@@ -151,6 +251,9 @@ public final class BridgedApp {
      * the subscription boundary, 400 for an unknown profile.
      */
     private void spawnWorker(Context ctx) {
+        if (!allow(ctx, Authz.Action.SPAWN, null)) {
+            return;
+        }
         String profile = ctx.queryParam("profile");
         String cwd = ctx.queryParam("cwd");
         String worktree = ctx.queryParam("worktree");
@@ -203,7 +306,11 @@ public final class BridgedApp {
 
     /** Tear a worker down by pane id. */
     private void stopWorker(Context ctx) {
-        sessions.release(ctx.pathParam("paneId"));
+        String paneId = ctx.pathParam("paneId");
+        if (!allow(ctx, Authz.Action.STOP, paneId)) {
+            return;
+        }
+        sessions.release(paneId);
         ctx.status(204);
     }
 
@@ -215,6 +322,9 @@ public final class BridgedApp {
      */
     private void sendMessage(Context ctx) {
         String id = ctx.pathParam("id");
+        if (!allow(ctx, Authz.Action.SEND, id)) {
+            return;
+        }
         String content;
         String turnId;
         long timeout;
@@ -296,6 +406,9 @@ public final class BridgedApp {
      */
     private void askMessage(Context ctx) {
         String id = ctx.pathParam("id");
+        if (!allow(ctx, Authz.Action.ASK, id)) {
+            return;
+        }
         String question;
         long timeout;
         try {
@@ -329,6 +442,12 @@ public final class BridgedApp {
      */
     private void replyMessage(Context ctx) {
         String id = ctx.pathParam("id");
+        // The rule that matters: a worker may reply only as itself. Over MCP this was already true
+        // structurally (identity comes from the connection, never an argument); over REST the path
+        // id was simply trusted, so this is where the invariant actually gets enforced.
+        if (!allow(ctx, Authz.Action.REPLY, id)) {
+            return;
+        }
         String content;
         try {
             content = mapper.readTree(ctx.body()).path("content").asText("");
@@ -347,6 +466,9 @@ public final class BridgedApp {
      */
     private void drainReplies(Context ctx) {
         String id = ctx.pathParam("id");
+        if (!allow(ctx, Authz.Action.DRAIN, id)) {
+            return;
+        }
         var replies = messages.drainReplies(id);
         ctx.status(200).json(Map.of("sessionId", id, "replies",
                 replies.stream().map(m -> Map.of(
@@ -362,6 +484,9 @@ public final class BridgedApp {
      */
     private void sessionStatus(Context ctx) {
         String id = ctx.pathParam("id");
+        if (!allow(ctx, Authz.Action.READ, id)) {
+            return;
+        }
         try {
             ctx.status(200).json(Map.of(
                     "sessionId", id,
@@ -374,6 +499,9 @@ public final class BridgedApp {
 
     /** Poll an async (wait:false) delegation by ticket. 404 for an unknown/expired ticket. */
     private void taskStatus(Context ctx) {
+        if (!allow(ctx, Authz.Action.READ, null)) {
+            return;
+        }
         MessageService.TaskView v = messages.poll(ctx.pathParam("ticket"));
         if (v == null) {
             ctx.status(404).json(Map.of("error", "unknown_ticket", "detail", "no such task (or it has expired)"));

bridged/src/main/resources/logback.xml

@@ -5,10 +5,39 @@
         </encoder>
     </appender>
 
+    <!--
+      CB-505 audit trail. Its own file, deliberately not the app log: privileged actions
+      (spawn/stop/send/reply/drain) must stay greppable and shippable without dragging DEBUG noise
+      along. The message is already a JSON object (AuditLog builds it), so the pattern contributes
+      only an ISO-8601 timestamp — one appender owns the clock, so the trail cannot disagree with
+      itself. Rolls daily, 30 days retained, 100MB cap.
+
+      NOTE: records carry who/what/target/outcome only. Message CONTENT is never written here —
+      this bridge carries source code and prompts, and an audit log that accumulated them would be
+      a transcript archive rather than a control.
+    -->
+    <appender name="AUDIT" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>logs/audit.log</file>
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <fileNamePattern>logs/audit.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
+            <maxFileSize>10MB</maxFileSize>
+            <maxHistory>30</maxHistory>
+            <totalSizeCap>100MB</totalSizeCap>
+        </rollingPolicy>
+        <encoder>
+            <pattern>{"ts":"%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX}",%replace(%msg){'^\{',''}%n</pattern>
+        </encoder>
+    </appender>
+
     <logger name="dev.ltms.bridged" level="DEBUG"/>
     <logger name="io.javalin" level="INFO"/>
     <logger name="org.eclipse.jetty" level="WARN"/>
 
+    <!-- additivity=false keeps the audit stream out of stdout; it is its own record. -->
+    <logger name="audit" level="INFO" additivity="false">
+        <appender-ref ref="AUDIT"/>
+    </logger>
+
     <root level="INFO">
         <appender-ref ref="STDOUT"/>
     </root>

bridged/src/test/java/dev/ltms/bridged/auth/AuthzTest.java

@@ -0,0 +1,80 @@
+package dev.ltms.bridged.auth;
+
+import org.junit.jupiter.api.Test;
+
+import static dev.ltms.bridged.auth.Authz.Action.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+/** CB-505 — the authorization table, pinned so it cannot drift silently. */
+class AuthzTest {
+
+    private static final Principal PRIMARY = Principal.primary(100);
+    private static final Principal WORKER_A = Principal.worker("term_a", 200);
+    private static final Principal WORKER_B = Principal.worker("term_b", 300);
+    private static final Principal ANON = Principal.anonymous();
+
+    @Test
+    void anonymousIsAuthorizedForNothing() {
+        for (Authz.Action a : Authz.Action.values()) {
+            assertFalse(Authz.permits(ANON, a, "term_a"),
+                    a + " must be refused to an unauthenticated caller");
+        }
+    }
+
+    @Test
+    void aNullCallerIsTreatedAsAnonymous() {
+        assertFalse(Authz.permits(null, READ, null));
+        assertTrue(Authz.isUnauthenticated(null));
+    }
+
+    @Test
+    void orchestrationBelongsToThePrimaryAlone() {
+        for (Authz.Action a : new Authz.Action[]{SPAWN, STOP, SEND, DRAIN}) {
+            assertTrue(Authz.permits(PRIMARY, a, "term_a"), "the primary orchestrates: " + a);
+            assertFalse(Authz.permits(WORKER_A, a, "term_a"),
+                    "a worker performing " + a + " would be escalating into the orchestrator role");
+        }
+    }
+
+    @Test
+    void aWorkerMayReplyAndAskOnlyAsItself() {
+        assertTrue(Authz.permits(WORKER_A, REPLY, "term_a"));
+        assertTrue(Authz.permits(WORKER_A, ASK, "term_a"));
+
+        assertFalse(Authz.permits(WORKER_A, REPLY, "term_b"),
+                "worker A must not be able to reply on worker B's session");
+        assertFalse(Authz.permits(WORKER_B, ASK, "term_a"),
+                "worker B must not be able to ask as worker A");
+    }
+
+    @Test
+    void thePrimaryMayNotForgeAWorkersReply() {
+        // Not a hypothetical nicety: a forged reply would resolve the rendezvous the primary is
+        // itself blocked on, corrupting the correlation between a turn and its answer.
+        assertFalse(Authz.permits(PRIMARY, REPLY, "term_a"));
+        assertFalse(Authz.permits(PRIMARY, ASK, "term_a"));
+    }
+
+    @Test
+    void aWorkerWithNoTargetCannotReply() {
+        assertFalse(Authz.permits(WORKER_A, REPLY, null),
+                "an absent session id must not satisfy the own-session rule");
+    }
+
+    @Test
+    void observationIsOpenToBothAuthenticatedRoles() {
+        assertTrue(Authz.permits(PRIMARY, READ, null));
+        assertTrue(Authz.permits(WORKER_A, READ, null));
+        assertTrue(Authz.permits(PRIMARY, METRICS, null));
+        assertTrue(Authz.permits(WORKER_A, METRICS, null));
+    }
+
+    @Test
+    void unauthenticatedIsDistinguishedFromMerelyForbidden() {
+        // Drives the 401-vs-403 split: a missing credential is fixable by the caller, a wrong role
+        // is not.
+        assertTrue(Authz.isUnauthenticated(ANON));
+        assertFalse(Authz.isUnauthenticated(WORKER_A));
+        assertFalse(Authz.isUnauthenticated(PRIMARY));
+    }
+}

bridged/src/test/java/dev/ltms/bridged/auth/CallerResolverTest.java

@@ -0,0 +1,106 @@
+package dev.ltms.bridged.auth;
+
+import dev.ltms.bridged.herdr.FakeHerdr;
+import dev.ltms.bridged.herdr.PaneLocator;
+import dev.ltms.bridged.mcp.ConnectionIdentity;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * CB-501. The behaviour under test is the inversion of the pre-CB-501 default: failing every
+ * identity check must yield {@link Role#ANONYMOUS}, not {@code PRIMARY}.
+ */
+class CallerResolverTest {
+
+    private final FakeHerdr herdr = new FakeHerdr();
+
+    /** Identity resolving the canned worker pane, keyed off a faked peer-PID lookup. */
+    private ConnectionIdentity identity(long pid) {
+        return new ConnectionIdentity(new PaneLocator(herdr), _ -> pid);
+    }
+
+    /** A PID that owns a worker pane in the fake. */
+    private ConnectionIdentity workerIdentity() {
+        return identity(FakeHerdr.WORKER_PID);
+    }
+
+    /** A PID that owns no pane — i.e. the primary, or any other local process. */
+    private ConnectionIdentity nonWorkerIdentity() {
+        return identity(999_999);
+    }
+
+    @Test
+    void aLoopbackWorkerPaneResolvesToWorkerRegardlessOfAuthMode() {
+        Principal underTrust = new CallerResolver(workerIdentity()).resolve("127.0.0.1", 42, null);
+        Principal underToken = new CallerResolver(workerIdentity(), true, "s3cret")
+                .resolve("127.0.0.1", 42, null);
+
+        assertEquals(Role.WORKER, underTrust.role());
+        assertEquals("term_a", underTrust.terminal());
+        assertEquals(Role.WORKER, underToken.role(),
+                "worker identity is unforgeable and must never be token-gated — otherwise enabling "
+                        + "auth would lock the whole fleet out of bridge_reply");
+        assertEquals("term_a", underToken.terminal());
+    }
+
+    @Test
+    void loopbackTrustTreatsANonWorkerLoopbackCallerAsThePrimary() {
+        Principal p = new CallerResolver(nonWorkerIdentity()).resolve("127.0.0.1", 99, null);
+
+        assertEquals(Role.PRIMARY, p.role(), "the historical behaviour, now an explicit choice");
+    }
+
+    @Test
+    void tokenModeRefusesANonWorkerCallerThatPresentsNoToken() {
+        Principal p = new CallerResolver(nonWorkerIdentity(), true, "s3cret")
+                .resolve("127.0.0.1", 99, null);
+
+        assertEquals(Role.ANONYMOUS, p.role(),
+                "no credential must mean NOTHING, not the most privileged role on the bus");
+    }
+
+    @Test
+    void tokenModeAcceptsAValidBearerTokenAsThePrimary() {
+        Principal p = new CallerResolver(nonWorkerIdentity(), true, "s3cret")
+                .resolve("127.0.0.1", 99, "Bearer s3cret");
+
+        assertEquals(Role.PRIMARY, p.role());
+    }
+
+    @Test
+    void tokenModeRejectsAWrongOrMalformedCredential() {
+        CallerResolver r = new CallerResolver(nonWorkerIdentity(), true, "s3cret");
+
+        assertEquals(Role.ANONYMOUS, r.resolve("127.0.0.1", 99, "Bearer wrong").role());
+        assertEquals(Role.ANONYMOUS, r.resolve("127.0.0.1", 99, "s3cret").role(), "scheme required");
+        assertEquals(Role.ANONYMOUS, r.resolve("127.0.0.1", 99, "Bearer ").role(), "empty credential");
+        assertEquals(Role.ANONYMOUS, r.resolve("127.0.0.1", 99, "Basic s3cret").role(), "wrong scheme");
+    }
+
+    @Test
+    void theBearerSchemeIsCaseInsensitivePerRfc7235() {
+        CallerResolver r = new CallerResolver(nonWorkerIdentity(), true, "s3cret");
+
+        assertEquals(Role.PRIMARY, r.resolve("127.0.0.1", 99, "bearer s3cret").role());
+        assertEquals(Role.PRIMARY, r.resolve("127.0.0.1", 99, "BEARER s3cret").role());
+    }
+
+    @Test
+    void aNonLoopbackCallerIsNeverThePrimaryUnderLoopbackTrust() {
+        // Defence in depth: startup already refuses this pairing (validateAuthExposure), but if a
+        // proxy ever forwards a remote peer onto the loopback listener, the resolver must not
+        // hand it the primary role.
+        Principal p = new CallerResolver(nonWorkerIdentity()).resolve("10.0.0.7", 99, null);
+
+        assertEquals(Role.ANONYMOUS, p.role());
+    }
+
+    @Test
+    void tokenModeRequiresANonEmptyConfiguredToken() {
+        ConnectionIdentity id = nonWorkerIdentity();
+
+        assertThrows(IllegalArgumentException.class, () -> new CallerResolver(id, true, null));
+        assertThrows(IllegalArgumentException.class, () -> new CallerResolver(id, true, "  "));
+    }
+}

bridged/src/test/java/dev/ltms/bridged/config/BridgedConfigTest.java

@@ -233,4 +233,136 @@ class BridgedConfigTest {
         assertEquals(java.util.List.of("opencode"), cfg.workerProfiles().get("gemini").argv(),
                 "an opencode worker with no argv defaults to the opencode binary, never claude");
     }
+
+    @Test
+    void authDefaultsToLoopbackTrustSoExistingConfigsBehaveAsBefore(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("no-auth-block.yaml");
+        Files.writeString(f, "bind:\n  host: 127.0.0.1\n  port: 8765\n");
+
+        BridgedConfig cfg = BridgedConfig.load(f);
+        assertNotNull(cfg.auth(), "auth must default rather than be null");
+        assertFalse(cfg.auth().tokenMode());
+        assertEquals("BRIDGED_API_TOKEN", cfg.auth().tokenEnv(), "documented default env var");
+        assertDoesNotThrow(cfg::validateAuthExposure, "loopback + loopback-trust is the safe pairing");
+    }
+
+    /**
+     * CB-501's highest-value check. Under loopback-trust, "not a known worker" means "the primary" —
+     * sound only while the OS refuses remote connections. Widening the bind without token mode
+     * would silently promote every reachable client to the most privileged role on the bus.
+     */
+    @Test
+    void aNonLoopbackBindWithoutTokenModeIsRefusedAtStartup(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("exposed.yaml");
+        Files.writeString(f, "bind:\n  host: 0.0.0.0\n  port: 8765\n");
+
+        BridgedConfig cfg = BridgedConfig.load(f);
+        IllegalStateException e = assertThrows(IllegalStateException.class, cfg::validateAuthExposure);
+        assertTrue(e.getMessage().contains("auth.mode: token"),
+                "the error must say how to fix it, not just that it refused");
+    }
+
+    @Test
+    void aNonLoopbackBindIsAllowedOnceTokenModeIsOn(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("exposed-with-token.yaml");
+        Files.writeString(f, """
+                bind:
+                  host: 0.0.0.0
+                  port: 8765
+                auth:
+                  mode: token
+                  tokenEnv: MY_TOKEN
+                """);
+
+        BridgedConfig cfg = BridgedConfig.load(f);
+        assertTrue(cfg.auth().tokenMode());
+        assertEquals("MY_TOKEN", cfg.auth().tokenEnv());
+        assertDoesNotThrow(cfg::validateAuthExposure);
+    }
+
+    @Test
+    void loopbackFormsAreAllRecognised(@TempDir Path dir) throws Exception {
+        for (String host : new String[]{"127.0.0.1", "localhost", "::1", "127.0.0.53"}) {
+            Path f = dir.resolve("lb-" + host.replace(':', '_') + ".yaml");
+            Files.writeString(f, "bind:\n  host: \"" + host + "\"\n  port: 8765\n");
+            assertDoesNotThrow(() -> BridgedConfig.load(f).validateAuthExposure(),
+                    host + " is loopback and must not trip the exposure guard");
+        }
+    }
+
+    /**
+     * The shipped {@code bridged.example.yaml} must actually parse. Config binds through a plain
+     * Jackson mapper with {@code ignoreUnknown = true}, so a misspelled key in the example is
+     * silently dropped and the operator gets a default they did not ask for — exactly how a
+     * {@code spawn_ready_timeout_ms} typo survived in the example until the CB-5xx wrap-up.
+     */
+    @Test
+    void shippedExampleConfigParses() {
+        Path example = Path.of("bridged.example.yaml");
+        assertTrue(Files.exists(example), "bridged.example.yaml must ship next to the pom");
+
+        BridgedConfig cfg = BridgedConfig.load(example);
+        assertEquals(8765, cfg.bind().port(), "example binds the documented default port");
+        assertTrue(cfg.workerProfiles().containsKey("gx10"), "example documents the gx10 profile");
+        assertEquals("gx10", cfg.defaultProfile(), "example's defaultWorker resolves");
+        assertTrue(cfg.guard().hostSet().contains("gx01.gw"),
+                "every example profile's base_url host must be in the example allowlist");
+    }
+
+    /**
+     * Every optional knob the example documents must bind under the exact spelling used there.
+     * Keep this list in step with {@code bridged.example.yaml}: a rename that updates the record
+     * but not the example (or vice versa) fails here instead of silently no-op'ing in production.
+     */
+    @Test
+    void everyOptionalKnobDocumentedInTheExampleBinds(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("all-knobs.yaml");
+        Files.writeString(f, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                spawnReadyTimeoutMs: 25000
+                spawnReadyPollMs: 400
+                worktreeRoot: /tmp/bridged-worktrees
+                workers:
+                  gx10:
+                    kind: claude-code
+                    baseUrl: http://gx01.gw:8000
+                    configDir: /tmp/ccs/gx10
+                    cwd: /tmp/repo
+                    parityOverlay: [".mcp.json", ".env"]
+                    gitTokenEnv: GITEA_TOKEN
+                    gitHostEnv: GITEA_HOST
+                lifecycle:
+                  idleTtlSeconds: 300
+                  contextCap: 10
+                  drainTimeoutSeconds: 5
+                broker:
+                  uri: amqp://guest:guest@127.0.0.1:5672
+                primary:
+                  terminal: term_abc123
+                  pushReminders: 5
+                  pushBackoffMs: 15000
+                """);
+
+        BridgedConfig cfg = BridgedConfig.load(f);
+        assertEquals(25000, cfg.spawnReadyTimeoutMs(), "spawnReadyTimeoutMs is camelCase, not snake_case");
+        assertEquals(400, cfg.spawnReadyPollMs(), "spawnReadyPollMs is camelCase, not snake_case");
+        assertEquals("/tmp/bridged-worktrees", cfg.worktreeRoot());
+
+        BridgedConfig.Worker w = cfg.workerProfiles().get("gx10");
+        assertEquals("/tmp/ccs/gx10", w.configDir());
+        assertEquals("/tmp/repo", w.cwd());
+        assertEquals(java.util.List.of(".mcp.json", ".env"), w.parityOverlay());
+        assertTrue(w.hasGitToken(), "gitTokenEnv binds and enables the CB-302 PR grant");
+        assertEquals("GITEA_HOST", w.gitHostEnv());
+
+        assertEquals(300, cfg.lifecycle().idleTtlSeconds());
+        assertEquals(10, cfg.lifecycle().contextCap());
+        assertEquals(5, cfg.lifecycle().drainTimeoutSeconds());
+        assertEquals("amqp://guest:guest@127.0.0.1:5672", cfg.broker().uri());
+        assertEquals("term_abc123", cfg.primary().terminal());
+        assertEquals(5, cfg.primary().remindersOrDefault());
+        assertEquals(15000L, cfg.primary().backoffMsOrDefault());
+    }
 }

bridged/src/test/java/dev/ltms/bridged/metrics/MetricsTest.java

@@ -0,0 +1,127 @@
+package dev.ltms.bridged.metrics;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/** CB-502 — the zero-dependency Prometheus text renderer. */
+class MetricsTest {
+
+    @Test
+    void countersAccumulatePerLabelSet() {
+        Metrics m = new Metrics();
+        m.inc("bridged_sends_total", "outcome", "replied");
+        m.inc("bridged_sends_total", "outcome", "replied");
+        m.inc("bridged_sends_total", "outcome", "timeout");
+
+        assertEquals(2, m.count("bridged_sends_total", "outcome", "replied"));
+        assertEquals(1, m.count("bridged_sends_total", "outcome", "timeout"));
+        assertEquals(0, m.count("bridged_sends_total", "outcome", "failed"),
+                "an untouched series reads as zero, not an error");
+    }
+
+    @Test
+    void rendersHelpAndTypeOncePerFamily() {
+        Metrics m = new Metrics();
+        m.describe("bridged_sends_total", "counter", "Delegated sends by outcome.");
+        m.inc("bridged_sends_total", "outcome", "replied");
+        m.inc("bridged_sends_total", "outcome", "timeout");
+
+        String out = m.render();
+        assertEquals(1, countOccurrences(out, "# HELP bridged_sends_total"),
+                "HELP is per family, not per series");
+        assertEquals(1, countOccurrences(out, "# TYPE bridged_sends_total counter"));
+        assertTrue(out.contains("bridged_sends_total{outcome=\"replied\"} 1"));
+        assertTrue(out.contains("bridged_sends_total{outcome=\"timeout\"} 1"));
+    }
+
+    @Test
+    void labelsAreSortedSoScrapesAreByteStable() {
+        Metrics a = new Metrics();
+        a.inc("m", "b", "2", "a", "1");
+        Metrics b = new Metrics();
+        b.inc("m", "a", "1", "b", "2");
+
+        assertEquals(a.render(), b.render(), "label order in the call must not change the output");
+        assertTrue(a.render().contains("m{a=\"1\",b=\"2\"}"));
+    }
+
+    @Test
+    void gaugesAreEvaluatedAtScrapeTimeNotRegistrationTime() {
+        Metrics m = new Metrics();
+        int[] live = {1};
+        m.gauge("bridged_sessions", () -> live[0], "state", "ready");
+
+        assertTrue(m.render().contains("bridged_sessions{state=\"ready\"} 1"));
+        live[0] = 5;
+        assertTrue(m.render().contains("bridged_sessions{state=\"ready\"} 5"),
+                "the gauge must read current state on every scrape");
+    }
+
+    @Test
+    void aThrowingGaugeDoesNotBreakTheWholeScrape() {
+        Metrics m = new Metrics();
+        m.inc("good_total");
+        m.gauge("bad_gauge", () -> {
+            throw new IllegalStateException("herdr is down");
+        });
+
+        String out = assertDoesNotThrow(m::render);
+        assertTrue(out.contains("good_total 1"), "healthy series must still be exported");
+        assertFalse(out.contains("bad_gauge"), "the broken series is simply absent");
+    }
+
+    @Test
+    void collectorsDiscoverTheirLabelSetPerScrape() {
+        Metrics m = new Metrics();
+        Map<String, Number> depths = new LinkedHashMap<>();
+        m.collector("bridged_inbox_depth", "target", () -> depths);
+
+        assertFalse(m.render().contains("bridged_inbox_depth"), "no targets yet ⇒ no series");
+
+        depths.put("term_a", 2);
+        depths.put("term_b", 0);
+        String out = m.render();
+        assertTrue(out.contains("bridged_inbox_depth{target=\"term_a\"} 2"));
+        assertTrue(out.contains("bridged_inbox_depth{target=\"term_b\"} 0"));
+    }
+
+    @Test
+    void labelValuesAreEscaped() {
+        Metrics m = new Metrics();
+        m.inc("m", "detail", "he said \"hi\"\nand \\left");
+
+        String out = m.render();
+        assertTrue(out.contains("\\\""), "quotes escaped");
+        assertTrue(out.contains("\\n"), "newlines escaped — a raw one would corrupt the exposition");
+        assertTrue(out.contains("\\\\"), "backslashes escaped");
+    }
+
+    @Test
+    void wholeNumberGaugesRenderWithoutADecimalPoint() {
+        Metrics m = new Metrics();
+        m.gauge("whole", () -> 3.0);
+        m.gauge("fractional", () -> 1.5);
+
+        String out = m.render();
+        assertTrue(out.contains("whole 3"), "3.0 should not render as 3.0");
+        assertTrue(out.contains("fractional 1.5"));
+    }
+
+    @Test
+    void oddLabelCountIsRejected() {
+        Metrics m = new Metrics();
+        assertThrows(IllegalArgumentException.class, () -> m.inc("m", "dangling"));
+    }
+
+    private static int countOccurrences(String haystack, String needle) {
+        int n = 0;
+        for (int i = haystack.indexOf(needle); i >= 0; i = haystack.indexOf(needle, i + 1)) {
+            n++;
+        }
+        return n;
+    }
+}

bridged/src/test/java/dev/ltms/bridged/rest/BridgedAppAuthTest.java

@@ -0,0 +1,225 @@
+package dev.ltms.bridged.rest;
+
+import dev.ltms.bridged.auth.CallerResolver;
+import dev.ltms.bridged.config.BridgedConfig;
+import dev.ltms.bridged.guard.SubscriptionGuard;
+import dev.ltms.bridged.herdr.AgentControl;
+import dev.ltms.bridged.herdr.FakeHerdr;
+import dev.ltms.bridged.herdr.PaneLocator;
+import dev.ltms.bridged.herdr.WorkspaceControl;
+import dev.ltms.bridged.inject.Injector;
+import dev.ltms.bridged.mcp.ConnectionIdentity;
+import dev.ltms.bridged.metrics.BridgedMetrics;
+import dev.ltms.bridged.metrics.Metrics;
+import dev.ltms.bridged.msg.MessageService;
+import dev.ltms.bridged.msg.Rendezvous;
+import dev.ltms.bridged.session.FakeWorktrees;
+import dev.ltms.bridged.session.SessionManager;
+import dev.ltms.bridged.worker.ClaudeCodeLauncher;
+import io.javalin.Javalin;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * CB-501/505 enforcement over real HTTP. The unit tests pin the policy; these pin that the policy
+ * is actually reached from a request — a rule enforced nowhere is not a control.
+ */
+class BridgedAppAuthTest {
+
+    private final HttpClient http = HttpClient.newHttpClient();
+    private Javalin app;
+    private Metrics metrics;
+
+    @AfterEach
+    void stop() {
+        if (app != null) app.stop();
+    }
+
+    /**
+     * Start the app with the given identity/auth wiring.
+     *
+     * @param pid the PID every connection resolves to — {@link FakeHerdr#WORKER_PID} makes the
+     *            caller worker {@code term_a}, anything else makes it a non-worker
+     */
+    private int start(long pid, boolean tokenMode, String token) {
+        FakeHerdr herdr = new FakeHerdr();
+        BridgedConfig.Worker wcfg = new BridgedConfig.Worker(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "BRIDGED_WORKER_TOKEN", null,
+                "tab", "bridged-workers", "worker: {profile} #{n}", null, null, null);
+        AgentControl agents = new AgentControl(herdr);
+        ClaudeCodeLauncher workers = new ClaudeCodeLauncher(
+                agents, new WorkspaceControl(herdr), new SubscriptionGuard(Set.of("gx00.gw")),
+                Map.of(wcfg.profile(), wcfg), wcfg.profile(),
+                k -> "BRIDGED_WORKER_TOKEN".equals(k) ? "tok-abc" : null);
+        SessionManager sessions = new SessionManager(workers, new FakeWorktrees());
+        Injector injector = new Injector(agents);
+        MessageService messages = new MessageService(agents, injector, new Rendezvous());
+
+        ConnectionIdentity identity = new ConnectionIdentity(new PaneLocator(herdr), _ -> pid);
+        CallerResolver callers = tokenMode
+                ? new CallerResolver(identity, true, token)
+                : new CallerResolver(identity);
+        metrics = BridgedMetrics.create(sessions, new dev.ltms.bridged.msg.InMemoryReplyInbox());
+
+        app = new BridgedApp(herdr, workers, sessions, messages, sessions.asPresence(), null,
+                callers, metrics).build().start("127.0.0.1", 0);
+        return app.port();
+    }
+
+    private HttpResponse<String> send(int port, String method, String path, String body, String auth)
+            throws Exception {
+        HttpRequest.Builder b = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + port + path))
+                .header("Content-Type", "application/json");
+        if (auth != null) {
+            b.header("Authorization", auth);
+        }
+        b = switch (method) {
+            case "POST" -> b.POST(body == null
+                    ? HttpRequest.BodyPublishers.noBody()
+                    : HttpRequest.BodyPublishers.ofString(body));
+            case "DELETE" -> b.DELETE();
+            default -> b.GET();
+        };
+        return http.send(b.build(), HttpResponse.BodyHandlers.ofString());
+    }
+
+    // --- loopback-trust: the caller is the primary -------------------------------------------
+
+    @Test
+    void thePrimaryMayOrchestrateButMayNotForgeAWorkerReply() throws Exception {
+        int port = start(999_999, false, null); // no pane ⇒ primary
+
+        HttpResponse<String> read = send(port, "GET", "/profiles", null, null);
+        assertEquals(200, read.statusCode(), "the primary may observe");
+
+        HttpResponse<String> reply = send(port, "POST", "/sessions/term_a/reply",
+                "{\"content\":\"forged\"}", null);
+        assertEquals(403, reply.statusCode(),
+                "a forged reply would resolve the rendezvous the primary is itself waiting on");
+        assertTrue(reply.body().contains("forbidden"));
+    }
+
+    // --- loopback-trust: the caller is a worker ------------------------------------------------
+
+    @Test
+    void aWorkerMayReplyAsItselfButNotAsAnother() throws Exception {
+        int port = start(FakeHerdr.WORKER_PID, false, null); // resolves to term_a
+
+        HttpResponse<String> own = send(port, "POST", "/sessions/term_a/reply",
+                "{\"content\":\"done\"}", null);
+        assertEquals(200, own.statusCode(), "a worker replies on its own session");
+
+        HttpResponse<String> other = send(port, "POST", "/sessions/term_b/reply",
+                "{\"content\":\"not mine\"}", null);
+        assertEquals(403, other.statusCode(),
+                "REST trusted the path id before CB-505; this is the hole being closed");
+    }
+
+    @Test
+    void aWorkerMayNotOrchestrate() throws Exception {
+        int port = start(FakeHerdr.WORKER_PID, false, null);
+
+        assertEquals(403, send(port, "POST", "/workers", null, null).statusCode(),
+                "a worker spawning workers would be escalating into the orchestrator role");
+        assertEquals(403, send(port, "DELETE", "/workers/w2:p7", null, null).statusCode());
+        assertEquals(403, send(port, "POST", "/sessions/term_b/message",
+                "{\"content\":\"hi\"}", null).statusCode());
+        assertEquals(403, send(port, "GET", "/sessions/term_a/replies", null, null).statusCode(),
+                "draining an inbox is the primary's collection step");
+    }
+
+    // --- token mode ---------------------------------------------------------------------------
+
+    @Test
+    void tokenModeRejectsAnUncredentialedNonWorkerWith401() throws Exception {
+        int port = start(999_999, true, "s3cret");
+
+        HttpResponse<String> res = send(port, "GET", "/profiles", null, null);
+        assertEquals(401, res.statusCode(), "no credential ⇒ authenticated as nothing");
+        assertTrue(res.body().contains("unauthenticated"));
+    }
+
+    @Test
+    void tokenModeAcceptsAValidBearerToken() throws Exception {
+        int port = start(999_999, true, "s3cret");
+
+        assertEquals(200, send(port, "GET", "/profiles", null, "Bearer s3cret").statusCode());
+    }
+
+    @Test
+    void tokenModeStillHonoursConnectionDerivedWorkerIdentity() throws Exception {
+        // The fleet must keep working when auth is switched on: a worker presents no token, and
+        // must still be able to reply.
+        int port = start(FakeHerdr.WORKER_PID, true, "s3cret");
+
+        assertEquals(200, send(port, "POST", "/sessions/term_a/reply",
+                "{\"content\":\"done\"}", null).statusCode());
+    }
+
+    // --- health, metrics ----------------------------------------------------------------------
+
+    @Test
+    void healthzStaysOpenWithoutCredentials() throws Exception {
+        int port = start(999_999, true, "s3cret");
+
+        assertEquals(200, send(port, "GET", "/healthz", null, null).statusCode(),
+                "a supervisor must be able to probe liveness before any credential is configured");
+    }
+
+    @Test
+    void metricsRequireAuthenticationAndRenderPrometheusText() throws Exception {
+        int port = start(999_999, true, "s3cret");
+
+        assertEquals(401, send(port, "GET", "/metrics", null, null).statusCode());
+
+        HttpResponse<String> ok = send(port, "GET", "/metrics", null, "Bearer s3cret");
+        assertEquals(200, ok.statusCode());
+        assertTrue(ok.headers().firstValue("Content-Type").orElse("").startsWith("text/plain"));
+        assertTrue(ok.body().contains("bridged_sessions{state=\"ready\"}"),
+                "the session census gauge is exported even when empty");
+    }
+
+    @Test
+    void refusalsAreCounted() throws Exception {
+        int port = start(999_999, true, "s3cret");
+
+        send(port, "GET", "/profiles", null, null);              // 401
+        send(port, "POST", "/sessions/term_a/reply", "{}", "Bearer s3cret"); // 403
+
+        assertEquals(1, metrics.count(BridgedMetrics.AUTH_FAILURES, "reason", "unauthenticated"));
+        assertEquals(1, metrics.count(BridgedMetrics.AUTH_FAILURES, "reason", "forbidden"));
+    }
+
+    // --- legacy constructor -------------------------------------------------------------------
+
+    @Test
+    void theLegacyConstructorLeavesAuthorizationOff() throws Exception {
+        // The 29 pre-existing acceptance tests rely on this: no auth fixture, no enforcement.
+        FakeHerdr herdr = new FakeHerdr();
+        AgentControl agents = new AgentControl(herdr);
+        BridgedConfig.Worker wcfg = new BridgedConfig.Worker(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "BRIDGED_WORKER_TOKEN", null,
+                "tab", "bridged-workers", "worker: {profile} #{n}", null, null, null);
+        ClaudeCodeLauncher workers = new ClaudeCodeLauncher(
+                agents, new WorkspaceControl(herdr), new SubscriptionGuard(Set.of("gx00.gw")),
+                Map.of(wcfg.profile(), wcfg), wcfg.profile(), _ -> "tok");
+        SessionManager sessions = new SessionManager(workers, new FakeWorktrees());
+        MessageService messages = new MessageService(agents, new Injector(agents), new Rendezvous());
+        app = new BridgedApp(herdr, workers, sessions, messages, sessions.asPresence(), null)
+                .build().start("127.0.0.1", 0);
+
+        assertEquals(200, send(app.port(), "POST", "/sessions/term_a/reply",
+                "{\"content\":\"x\"}", null).statusCode());
+        assertEquals(404, send(app.port(), "GET", "/metrics", null, null).statusCode(),
+                "no registry supplied ⇒ the endpoint is not mounted at all");
+    }
+}

deploy/bridged.service

@@ -0,0 +1,57 @@
+# CB-504 — systemd unit for bridged (Linux).
+#
+# The macOS launchd agent (deploy/dev.ltms.bridged.plist) is the supervision target for the
+# current single-host deployment. This unit exists for the per-host gateways CB-308 introduces,
+# which will run on Linux.
+#
+# Install (user service — bridged drives the user's herdr, not a system daemon):
+#   mkdir -p ~/.config/systemd/user
+#   cp deploy/bridged.service ~/.config/systemd/user/
+#   # edit ExecStart / WorkingDirectory / Environment below, then:
+#   systemctl --user daemon-reload
+#   systemctl --user enable --now bridged
+#   journalctl --user -u bridged -f
+
+[Unit]
+Description=bridged — claude-bridge message server
+Documentation=https://git.ltms.dev/lms/claude-bridge/wiki
+# Ordering only: herdr is a user process and its socket may appear after us. This is advisory —
+# bridged retries the herdr socket rather than exiting, which is what actually makes a late
+# socket survivable. Do NOT add Requires=: a herdr restart must not take bridged down with it.
+After=herdr.service
+Wants=herdr.service
+
+[Service]
+Type=simple
+WorkingDirectory=%h/src/claude-bridge/bridged
+ExecStart=/usr/lib/jvm/temurin-25-jdk/bin/java -jar target/bridged.jar bridged.yaml
+
+Environment=HERDR_SOCKET_PATH=%h/.config/herdr/herdr.sock
+# Secrets are NOT set here — this file is committed. Put the API/worker tokens in a private
+# drop-in that systemd reads with restrictive permissions:
+#   systemctl --user edit bridged     →  [Service] / Environment=BRIDGED_API_TOKEN=...
+# or point EnvironmentFile at a 0600 file:
+# EnvironmentFile=%h/.config/bridged/env
+
+Restart=on-failure
+RestartSec=10s
+# A bad config (e.g. a non-loopback bind without token auth) makes bridged fail fast by design.
+# Give up rather than restart-loop on a permanent error.
+StartLimitBurst=5
+StartLimitIntervalSec=120
+
+# The daemon reads the repo, writes worktrees, and talks to a Unix socket — it needs no more.
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=read-write
+ProtectKernelTunables=true
+ProtectControlGroups=true
+RestrictSUIDSGID=true
+
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=bridged
+
+[Install]
+WantedBy=default.target

deploy/dev.ltms.bridged.plist

@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<!--
+  CB-504 — launchd agent for bridged (macOS).
+
+  This is the real supervision target today: the dogfooded daemon runs on macOS, where there is
+  no systemd. A systemd unit ships alongside (deploy/bridged.service) for the Linux gateways
+  CB-308 introduces.
+
+  Install:
+    cp deploy/dev.ltms.bridged.plist ~/Library/LaunchAgents/
+    # edit the paths + JAVA_HOME below to match this host, then:
+    launchctl load -w ~/Library/LaunchAgents/dev.ltms.bridged.plist
+    launchctl list | grep bridged
+
+  Note on ordering: launchd has no "start after herdr" primitive for user agents, and neither
+  does systemd in a way that survives a socket appearing late. bridged retries the herdr socket
+  on startup instead, so an agent that comes up before herdr converges rather than dying — that
+  retry is the actual fix; KeepAlive below is the backstop.
+-->
+<plist version="1.0">
+<dict>
+    <key>Label</key>
+    <string>dev.ltms.bridged</string>
+
+    <key>ProgramArguments</key>
+    <array>
+        <string>/Users/CHANGEME/Tool/jdk-25.0.2.jdk/Contents/Home/bin/java</string>
+        <string>-jar</string>
+        <string>/Users/CHANGEME/src/claude-bridge/bridged/target/bridged.jar</string>
+        <string>bridged.yaml</string>
+    </array>
+
+    <!-- Config path in ProgramArguments is relative, so the working directory must be the module. -->
+    <key>WorkingDirectory</key>
+    <string>/Users/CHANGEME/src/claude-bridge/bridged</string>
+
+    <key>EnvironmentVariables</key>
+    <dict>
+        <key>JAVA_HOME</key>
+        <string>/Users/CHANGEME/Tool/jdk-25.0.2.jdk/Contents/Home</string>
+        <key>HERDR_SOCKET_PATH</key>
+        <string>/Users/CHANGEME/.config/herdr/herdr.sock</string>
+        <!--
+          Worker/API tokens are NOT set here: this file is committed. Export them from a private
+          launchd override or a wrapper script. bridged reads the API token from the env var named
+          by auth.tokenEnv (default BRIDGED_API_TOKEN) and only in auth.mode: token.
+        -->
+    </dict>
+
+    <key>RunAtLoad</key>
+    <true/>
+
+    <!-- Restart on crash, but not in a tight loop if the config is bad (bridged fails fast on a
+         non-loopback bind without token auth — that is a config error, not a transient one). -->
+    <key>KeepAlive</key>
+    <dict>
+        <key>SuccessfulExit</key>
+        <false/>
+    </dict>
+    <key>ThrottleInterval</key>
+    <integer>10</integer>
+
+    <key>StandardOutPath</key>
+    <string>/Users/CHANGEME/src/claude-bridge/bridged/logs/bridged.out.log</string>
+    <key>StandardErrorPath</key>
+    <string>/Users/CHANGEME/src/claude-bridge/bridged/logs/bridged.err.log</string>
+
+    <key>ProcessType</key>
+    <string>Background</string>
+</dict>
+</plist>

docs/CB-301-ext-Worktree-Provisioning.md

@@ -1,6 +1,10 @@
 # CB-301-ext — Worktree provisioning + config-parity overlay
 
-**Status:** design spec for review → delegate implementation.
+**Status:** ✅ shipped — implemented at commit `97ecc71` (per-worker git worktree + config-parity
+overlay). As-built: `session/GitWorktrees.java` behind the `Worktrees` port, wired in
+`Bridged.main` and configurable via `worktreeRoot` / per-profile `parityOverlay`
+(see `bridged.example.yaml`). Branch/worktree surface in `bridge_list` landed with CB-304
+(`9fe04bf`); the worker-opened-PR checkpoint landed as CB-302 (`64e70ef`).
 **Extends:** [CB-301 Session Manager](CB-301-Session-Manager.md) (shipped, commit `54d907c`).
 **Realizes:** the config-parity requirement in [Worker Git Workflow](Worker-Git-Workflow.md).
 **Grounded in:** `SessionManager`, `WorkerService.spawn/effectiveCwd`, `BridgedConfig.Worker`,

docs/CB-402-OpenCode-Adapter.md

@@ -1,6 +1,11 @@
 # CB-402 — Second peer adapter: opencode (Stage B of the Peer Launcher SPI)
 
-**Status:** design note (pre-implementation)
+**Status:** ✅ **implemented and merged** (`ded226a`) — increments 1–4 of §4 all landed
+(`HerdrPeerLauncher` base, `kind:` discriminator, `OpenCodeLauncher`, `CompositePeerLauncher`).
+⚠️ **Increment 5 — the live dogfood (§5) — has NOT run.** It was deferred at merge time pending a
+daemon restart and a resolved provider, and §7 Q1 (which provider this host has credentials for) is
+still open; `opencode` is not installed on the dev host. Gitea issue #7 stays open until the §5
+checklist is executed. This is the single known-unverified item going into the cross-host stage.
 **Depends on:** CB-401 Stage A (`PeerLauncher` SPI, merged `3aa69a9`)
 **Stage:** 4 (Pluggable peers) · Stage B
 **Owner action:** design-note → file issue → delegate → primary-verify (per CB-401/306/307)

docs/CB-5xx-Hardening.md

@@ -0,0 +1,242 @@
+# CB-5xx — Stage 5 Hardening (auth · metrics · CI · supervision · authz+audit)
+
+**Status:** design note (pre-implementation) — the single-host close-out before cross-host work.
+**Covers:** CB-501 (bearer auth + TLS) · CB-502 (`/metrics`) · CB-503 (mock-socket CI) ·
+CB-504 (service supervision) · CB-505 (per-session authz + audit log).
+**Depends on:** everything shipped through CB-402. Nothing here changes messaging semantics.
+**Blocks:** CB-308. The cross-host trust model is CB-308's own gating concern, and it inherits
+whatever identity/authz shape lands here — so this stage is deliberately *before* federation,
+not after it.
+
+---
+
+## 1. Why this stage is not optional bookkeeping
+
+`bridged` today has **exactly one security control: the loopback bind**. Every other guarantee
+rests on it.
+
+The identity model (`mcp/ConnectionIdentity.java`) resolves a caller from the connection alone —
+the OS reports the connecting PID, herdr owns the PID→pane map, so a worker cannot forge another
+worker. Its own javadoc is explicit: *"Single-host only (the herd shares the `bridged` host); the
+token path is the split-host fallback."* The token path does not exist yet.
+
+That leaves a seam that is **latent today and load-bearing the moment the bind moves**:
+
+```java
+// ConnectionIdentity.resolve — non-loopback callers get a null terminal
+if (!isLoopback(remoteAddr)) return new Caller(null, -1);
+```
+
+…and `null` terminal is interpreted downstream as **"this caller is the primary"**. Combined:
+
+> Any caller that is not a recognised on-host worker pane is treated as the primary — including,
+> if `bind.host` is ever widened, an arbitrary remote client.
+
+Today `bind` defaults to `127.0.0.1` so this is unreachable. But CB-308 exists precisely to widen
+the boundary, and the primary is the *most* privileged role on the bus (it spawns, stops, sends to
+any session, and drains any inbox). Shipping federation on top of "unauthenticated ⇒ primary"
+would be building the security boundary backwards.
+
+**So CB-501 is not "add a token header". It is: make identity explicit, and make the absence of
+identity mean *nothing*, not *everything*.**
+
+---
+
+## 2. Decisions (locked)
+
+### D1 — Three caller roles, one resolution path
+
+Introduce `Role { PRIMARY, WORKER, ANONYMOUS }` resolved by a single `CallerResolver` that both
+REST and MCP go through. Resolution order:
+
+1. **Connection identity wins where it applies.** A loopback peer PID that maps to a herdr worker
+   pane ⇒ `WORKER` with that terminal. Unforgeable, unchanged from today, zero config.
+2. **Token, if presented.** A valid bearer token ⇒ the role that token is provisioned for.
+3. **Otherwise `ANONYMOUS`** — *not* `PRIMARY`.
+
+This inverts today's default. `PRIMARY` becomes something you must *prove* (by being a loopback
+non-worker process when auth is disabled, or by presenting a primary-scoped token when it is
+enabled), rather than something you get by failing every other check.
+
+### D2 — Auth is opt-in by config, but the *default* must stay zero-friction on loopback
+
+The daemon is dogfooded constantly on one machine. If enabling hardening breaks the local setup,
+it will be disabled and the stage is wasted. So:
+
+```yaml
+auth:
+  mode: loopback-trust   # default — behaves exactly like today: loopback ⇒ PRIMARY, no token needed
+  # mode: token          # every non-worker caller must present a valid bearer token
+  # tokenEnv: BRIDGED_API_TOKEN   # host env var holding the token; never the literal value
+```
+
+`mode: loopback-trust` is the current behaviour, named honestly and now *chosen* rather than
+implied. `mode: token` is what a non-loopback bind requires. **A non-loopback `bind.host` with
+`mode: loopback-trust` must fail fast at startup** — that check is the single highest-value line
+in this stage, because it makes the dangerous configuration unrepresentable rather than merely
+discouraged.
+
+### D3 — TLS terminates *outside* the JVM
+
+Do **not** add TLS config to Javalin/Jetty. The deployment story for a cross-host gateway is a
+reverse proxy (or an SSH/WireGuard tunnel) in front of the daemon; the AMQP link has its own TLS
+via the broker URI (`amqps://`). Adding keystore handling here would mean certificate lifecycle
+code in a daemon whose whole value is being small, and would duplicate what the proxy does better.
+
+**CB-501 therefore ships bearer auth + the fail-fast bind check, and documents TLS as a
+deployment concern with a worked reverse-proxy example.** This is a deliberate narrowing of the
+roadmap's "auth/TLS" wording — flagged in §6 for the lead.
+
+### D4 — Metrics without a new dependency
+
+The roadmap's tech-stack table says Micrometer→Prometheus. Recommend **not** taking that dep:
+
+- This pom already carries an unusually heavy dependency-reconciliation burden (a hand-pinned
+  `jackson-annotations` 3.0-rc5 to reconcile the MCP SDK's Jackson 3 with our Jackson 2.19, a
+  Jetty BOM import to stop version skew, plus four documented accepted-CVE advisories). Every new
+  transitive tree is a real cost here, not a hypothetical one.
+- The CVE gate that CLAUDE.md mandates for dependency changes (`jetbrains get_file_problems` →
+  Mend.io) **cannot currently be run** — no JetBrains MCP server is connected. Adding a dependency
+  tree we cannot scan violates the project's own stated policy.
+- The metric set is small and fully known (§4). Prometheus text exposition is a trivial,
+  stable, well-specified format.
+
+So: a ~120-line `metrics/Metrics.java` holding `LongAdder` counters and gauge suppliers, rendered
+to the Prometheus text format at `GET /metrics`. If Micrometer is wanted later for its
+registry/push ecosystem, this stays a drop-in swap behind the same endpoint. **Flagged in §6 —
+this deviates from a documented tech-stack choice.**
+
+### D5 — Supervision targets launchd first, systemd second
+
+The roadmap says "systemd unit". **This host is macOS — there is no systemd on it** (`systemctl`
+not found), and the daemon that has been dogfooded for weeks runs as a bare foreground
+`java -jar`. Ship **both**:
+
+- `deploy/dev.ltms.bridged.plist` — launchd agent, the *actual* runtime here, with `KeepAlive` and
+  ordered start after herdr.
+- `deploy/bridged.service` — systemd unit for the Linux gateways CB-308 introduces.
+
+Ordering after herdr is advisory in both: the herdr socket may not exist at boot, so the daemon
+must **retry the socket rather than exit** — supervision ordering is a nicety, socket-retry is the
+actual fix. That retry behaviour is part of CB-504, not a separate ticket.
+
+### D6 — Audit log is a separate append-only stream, not the app log
+
+Privileged actions (spawn, stop, send, reply-drain, ack) emit a structured JSON line to a
+dedicated `audit` SLF4J logger with its own appender, carrying `{ts, role, terminal, pid, action,
+target, outcome}`. Keeping it off the chatty app logger is what makes it greppable and, later,
+shippable. **No message *content* in the audit record** — the bridge carries the user's source
+code and prompts; an audit trail that quietly becomes a transcript archive is a liability, not a
+control. Content stays out; correlation ids go in.
+
+---
+
+## 3. Authorization model (CB-505)
+
+With D1's roles, the rules are small enough to state completely:
+
+| Action | REST | PRIMARY | WORKER | ANONYMOUS |
+|---|---|---|---|---|
+| spawn worker | `POST /workers` | ✅ | ❌ | ❌ |
+| stop worker | `DELETE /workers/{paneId}` | ✅ | ❌ | ❌ |
+| send to a session | `POST /sessions/{id}/message` | ✅ | ❌ | ❌ |
+| reply | `POST /sessions/{id}/reply` | ❌ | ✅ **own session only** | ❌ |
+| ask | `POST /sessions/{id}/ask` | ❌ | ✅ **own session only** | ❌ |
+| drain replies | `GET /sessions/{id}/replies` | ✅ | ❌ | ❌ |
+| status / list / profiles | `GET …` | ✅ | ✅ | ❌ |
+| health | `GET /healthz` | ✅ | ✅ | ✅ (unauthenticated by design) |
+| metrics | `GET /metrics` | ✅ | ✅ | ❌ |
+
+The load-bearing row is **"own session only"**: a worker may only reply or ask *as itself*. That is
+already true de facto — `ConnectionIdentity` derives the terminal rather than reading it from the
+body — so CB-505 mostly **asserts an existing invariant explicitly** and adds the test that pins
+it. The one real change is rejecting a worker that names a *different* session id in the path.
+
+`/healthz` stays open: it must answer for a load balancer or supervisor before any credential is
+configured. It already leaks nothing but herdr's version and up/down.
+
+### 3.1 There are TWO entry paths, and only one of them has identity today
+
+The wiki describes MCP as "a thin adapter over the REST core". **At the code level that is not
+literally true, and the difference is security-relevant.** `BridgeMcp` calls `MessageService` /
+`SessionManager` *directly*; it never issues an HTTP request against a Javalin route. And `/mcp` is
+mounted as a raw servlet on Jetty's `ServletContextHandler`
+(`BridgedApp.build → cfg.jetty.modifyServletContextHandler`), so it does **not** pass through
+Javalin's `before` filters at all.
+
+The current split is the mirror image of what you'd expect:
+
+| Path | Caller identity today | Authz today |
+|---|---|---|
+| MCP `/mcp` | ✅ resolved per call (`ConnectionIdentity` via the transport-context extractor) | ❌ none |
+| REST routes | ❌ **none at all** — the session id is taken from the URL path and trusted | ❌ none |
+
+So REST is the *more* exposed surface: `POST /sessions/{id}/reply` accepts any `{id}` from the
+path, whereas the MCP `bridge_reply` derives the worker from the connection and refuses to read it
+from an argument. Loopback-only bind is what makes this safe today.
+
+**Therefore CB-505 must enforce on both paths against one shared resolver** — not at a single
+choke point. Concretely: a Javalin `before` filter for REST, and the existing transport-context
+extractor for MCP, both delegating to `auth.CallerResolver`. Any authz check that lives in only
+one of the two is not a control.
+
+---
+
+## 4. Metric set (CB-502)
+
+Deliberately small; every one maps to a failure mode we have actually hit.
+
+| Metric | Type | Why it exists |
+|---|---|---|
+| `bridged_sends_total{outcome}` | counter | outcome ∈ replied\|completion_fallback\|timeout\|failed — the completion-fallback rate is the health signal for turn detection (CB-115/116/118) |
+| `bridged_send_duration_seconds` | histogram | delegated turn latency |
+| `bridged_replies_total{path}` | counter | path ∈ rendezvous\|inbox — how often a reply strands (CB-307's whole reason to exist) |
+| `bridged_inbox_depth{target}` | gauge | undrained replies; steady-state should be 0 |
+| `bridged_push_nudges_total{outcome}` | counter | outcome ∈ delivered\|exhausted — a rising `exhausted` means the primary is not draining |
+| `bridged_spawns_total{kind,outcome}` | counter | outcome ∈ ready\|timeout\|guard_rejected; per peer kind (CB-402) |
+| `bridged_sessions{state}` | gauge | SPAWNING/READY/BUSY/DONE census |
+| `bridged_herdr_calls_total{method,outcome}` | counter | socket health — the dependency everything rests on |
+| `bridged_auth_failures_total{reason}` | counter | only meaningful once CB-501 lands; catches misconfigured workers |
+
+---
+
+## 5. Increment plan
+
+Ordered so each step is independently mergeable and the risky one lands first.
+
+1. **CB-501a — `CallerResolver` + `Role`.** Pure refactor: route today's connection identity through
+   the new type, `ANONYMOUS` not yet reachable (loopback-trust default preserves behaviour).
+   Green build, no behaviour change.
+2. **CB-501b — token mode + fail-fast bind check.** Config block, bearer parsing, the
+   non-loopback-bind guard. This is the security-relevant commit; keep it small and reviewable.
+3. **CB-505 — authz table + audit logger.** Enforce §3 on **both** entry paths (see §3.1); add the
+   audit appender.
+4. **CB-502 — `Metrics` + `/metrics`.** Instrument the paths in §4.
+5. **CB-503 — CI.** Runs `mvn -B clean install` with `-Dgroups='!contract'` so the live-herdr and
+   RabbitMQ contract tests are excluded; the mock-UDS suite is the CI surface, exactly as the
+   roadmap's testability section intends.
+6. **CB-504 — launchd plist + systemd unit + herdr-socket retry.**
+
+---
+
+## 6. Open questions for the lead
+
+*All four resolved 2026-07-29 — the lead confirmed D3, D4, and the D6 sink; the CI runner question
+was answered from the forge itself. Kept here as the decision record.*
+
+1. ✅ **TLS scope (D3) — confirmed.** Bearer auth + the fail-fast bind guard ship in the daemon;
+   TLS terminates at a reverse proxy, documented with a worked example. AMQP gets TLS via an
+   `amqps://` URI. No keystore handling in `bridged`.
+2. ✅ **Micrometer (D4) — confirmed dropped.** Zero-dependency Prometheus text renderer, for the
+   reasons in D4 (pom reconciliation burden + the mandated CVE gate being un-runnable this
+   session). Revisit if a push-gateway or JVM-metrics requirement appears; the endpoint is the
+   swap seam.
+3. ~~**CI runner (CB-503).**~~ ✅ **Resolved during design** — a Gitea Actions runner *is*
+   registered and healthy (`lms/alms-memory` has 28 completed runs; `lms/alms` runs on push and
+   pull_request). CB-503 targets `.gitea/workflows/ci.yml` with `runs-on: ubuntu-latest`, matching
+   the sibling repo's convention. Note the runner's image ships an older `default-jdk`, so the
+   workflow must provision **JDK 25** explicitly rather than apt-installing the default.
+   Contract-test exclusion needs no CI flag: the pom's `default-excludes` profile already sets
+   `excludedGroups=contract`, so a plain `mvn -B clean install` *is* the mock-socket surface.
+4. ✅ **Audit sink — confirmed dedicated file.** Its own logback appender writing JSON lines beside
+   the daemon, separate from the app log, per D6. Content still never enters the record.