diff --git a/.env.example b/.env.example
index 0512a316..d22439c5 100644
--- a/.env.example
+++ b/.env.example
@@ -35,6 +35,32 @@ API_RATE_LIMIT=200
 # Increase for environments with many concurrent browser sessions behind shared NAT.
 API_POLLING_RATE_LIMIT=300
 
+# ─── Pilot agent (remote host only) ──────────────────────────────
+# These three vars are required ONLY on a remote host running as a
+# pilot-agent reverse-tunnel container. The primary instance does not
+# read them. See docs/features/pilot-agent.mdx for the full setup.
+
+# Switches the container into agent mode. The primary instance leaves
+# this unset.
+SENCHO_MODE=
+
+# WebSocket-capable URL of the controlling Sencho instance. Use https://
+# scheme; the agent rewrites it to wss:// for the tunnel upgrade.
+SENCHO_PRIMARY_URL=
+
+# Single-use, 15-minute enrollment token issued by the primary when the
+# pilot-agent node is created. After the first successful connect the
+# agent persists a long-lived tunnel credential at /app/data/pilot.jwt
+# and ignores this var on subsequent restarts.
+SENCHO_ENROLL_TOKEN=
+
+# Optional: path inside the agent container to a PEM CA bundle the
+# agent should trust when validating the primary's TLS cert. Use this
+# for self-hosted deployments terminating TLS with an internal CA.
+# Leave unset to fall back to the system trust store. There is no
+# escape hatch to disable TLS verification.
+SENCHO_PILOT_CA_FILE=
+
 # ─── SSO / LDAP Configuration ────────────────────────────────────
 
 # LDAP / Active Directory
diff --git a/backend/src/__tests__/pilot-bridge-limits.test.ts b/backend/src/__tests__/pilot-bridge-limits.test.ts
new file mode 100644
index 00000000..11c8dc04
--- /dev/null
+++ b/backend/src/__tests__/pilot-bridge-limits.test.ts
@@ -0,0 +1,140 @@
+/**
+ * Tests for the pilot-tunnel bridge resource limits introduced in the
+ * hardening pass:
+ *   - Frame-size cap enforced at the protocol decoders.
+ *   - Per-tunnel concurrent stream cap enforced by the bridge.
+ *   - Per-stream idle timeout (verified via internal helpers; the real
+ *     timer fires at 10 minutes which is too long for a unit test, so we
+ *     drive the timer paths via reduced state).
+ *
+ * The bridge is exercised through its loopback HTTP server with a mock
+ * tunnel WebSocket; this keeps the test in-process and avoids needing a
+ * full pilot agent for the cap surfaces.
+ */
+import { describe, it, expect, beforeAll, afterAll } from 'vitest';
+import http from 'http';
+import { EventEmitter } from 'events';
+import { WebSocket } from 'ws';
+import {
+    MAX_FRAME_SIZE_BYTES,
+    MAX_STREAMS_PER_TUNNEL,
+    decodeBinaryFrame,
+    decodeJsonFrame,
+} from '../pilot/protocol';
+import { PilotTunnelBridge } from '../services/PilotTunnelBridge';
+
+describe('protocol decoders', () => {
+    it('rejects oversized binary frames', () => {
+        const buf = Buffer.alloc(MAX_FRAME_SIZE_BYTES + 1);
+        expect(() => decodeBinaryFrame(buf)).toThrow(/too large/);
+    });
+
+    it('rejects undersized binary frames', () => {
+        expect(() => decodeBinaryFrame(Buffer.alloc(3))).toThrow(/too short/);
+    });
+
+    it('rejects oversized JSON frames', () => {
+        const big = '"' + 'x'.repeat(MAX_FRAME_SIZE_BYTES) + '"';
+        expect(() => decodeJsonFrame(big)).toThrow(/too large/);
+    });
+
+    it('rejects malformed JSON', () => {
+        expect(() => decodeJsonFrame('not json')).toThrow();
+    });
+
+    it('rejects JSON without a type discriminator', () => {
+        expect(() => decodeJsonFrame('{"hello":"world"}')).toThrow(/discriminator/);
+    });
+});
+
+/**
+ * Minimal mock that satisfies PilotTunnelBridge's contract with its
+ * tunnelWs argument. The bridge calls `send`, reads `bufferedAmount`,
+ * checks `readyState`, and listens for `message`, `close`, `error`.
+ */
+function makeMockTunnelWs(): EventEmitter & {
+    sent: unknown[];
+    readyState: number;
+    bufferedAmount: number;
+    send: (data: unknown) => void;
+    ping: () => void;
+    close: () => void;
+} {
+    const ws = new EventEmitter() as EventEmitter & {
+        sent: unknown[];
+        readyState: number;
+        bufferedAmount: number;
+        send: (data: unknown) => void;
+        ping: () => void;
+        close: () => void;
+    };
+    ws.sent = [];
+    ws.readyState = WebSocket.OPEN;
+    ws.bufferedAmount = 0;
+    ws.send = (data: unknown) => { ws.sent.push(data); };
+    ws.ping = () => { /* no-op */ };
+    ws.close = () => { ws.readyState = WebSocket.CLOSED; ws.emit('close'); };
+    return ws;
+}
+
+describe('PilotTunnelBridge concurrent stream cap', () => {
+    let bridge: PilotTunnelBridge;
+    let mockWs: ReturnType<typeof makeMockTunnelWs>;
+    let loopbackUrl: string;
+
+    beforeAll(async () => {
+        mockWs = makeMockTunnelWs();
+        // The bridge constructor type-checks against `WebSocket`; the runtime
+        // only needs the surface exercised above, so the cast keeps the test
+        // honest without dragging in a real ws pair just to fill the cap.
+        bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket);
+        await bridge.start();
+        loopbackUrl = bridge.getLoopbackUrl();
+    });
+
+    afterAll(() => {
+        bridge.close();
+    });
+
+    it('refuses HTTP requests with 503 once the cap is reached', async () => {
+        // Use openTcpStream to fill the stream map without putting any load
+        // on the loopback HTTP socket pool. Each TCP stream consumes a slot
+        // exactly the same way an HTTP request would.
+        const filled: Array<ReturnType<PilotTunnelBridge['openTcpStream']>> = [];
+        for (let i = 0; i < MAX_STREAMS_PER_TUNNEL; i++) {
+            const handle = bridge.openTcpStream({ stack: 's', service: 'svc', port: 80 });
+            // Suppress the 'error' that fires later when we close the
+            // bridge; the handle is used purely for the slot.
+            handle?.on('error', () => { /* expected on cleanup */ });
+            filled.push(handle);
+        }
+
+        // Confirm all 1024 slot allocations actually serialized a tcp_open
+        // frame on the tunnel. If a slot were short-circuited without
+        // sending the frame, this would catch the regression.
+        const sentCount = mockWs.sent.length;
+        expect(sentCount).toBe(MAX_STREAMS_PER_TUNNEL);
+
+        // The (cap+1)th openTcpStream returns null per the cap check, and
+        // does NOT consume a stream id (no tcp_open is sent).
+        expect(bridge.openTcpStream({ stack: 's', service: 'svc', port: 80 })).toBeNull();
+        expect(mockWs.sent.length).toBe(sentCount);
+
+        // And a loopback HTTP request lands on the 503 branch of
+        // handleLoopbackRequest.
+        const url = new URL(loopbackUrl);
+        const overflow = await new Promise<number>((resolve, reject) => {
+            const req = http.request({
+                host: url.hostname,
+                port: Number(url.port),
+                method: 'GET',
+                path: '/over-the-cap',
+            }, (res) => resolve(res.statusCode || 0));
+            req.on('error', reject);
+            req.end();
+        });
+        expect(overflow).toBe(503);
+        // The 503 path also does not allocate a stream or send any frame.
+        expect(mockWs.sent.length).toBe(sentCount);
+    }, 15_000);
+});
diff --git a/backend/src/__tests__/pilot-enrollment.test.ts b/backend/src/__tests__/pilot-enrollment.test.ts
new file mode 100644
index 00000000..f1b95f98
--- /dev/null
+++ b/backend/src/__tests__/pilot-enrollment.test.ts
@@ -0,0 +1,204 @@
+/**
+ * Tests for pilot-agent enrollment lifecycle and rate limiting.
+ *
+ * Covers:
+ *   - POST /api/nodes with mode=pilot_agent mints an enrollment token, persists
+ *     the SHA256 hash into pilot_enrollments, and returns a docker run command
+ *     containing the bearer token.
+ *   - consumePilotEnrollment is one-shot: a second consume on the same hash
+ *     returns null (replay protection).
+ *   - Expired enrollments are not consumable.
+ *   - The enrollment rate limiter (10/min in production, 100/min in dev) is
+ *     wired only on routes that mint a pilot enrollment.
+ */
+import { describe, it, expect, beforeAll, afterAll } from 'vitest';
+import request from 'supertest';
+import crypto from 'crypto';
+import { setupTestDb, cleanupTestDb, loginAsTestAdmin } from './helpers/setupTestDb';
+
+let tmpDir: string;
+let app: import('express').Express;
+let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
+let adminCookie: string;
+
+beforeAll(async () => {
+  tmpDir = await setupTestDb();
+  ({ DatabaseService } = await import('../services/DatabaseService'));
+  ({ app } = await import('../index'));
+  adminCookie = await loginAsTestAdmin(app);
+});
+
+afterAll(() => cleanupTestDb(tmpDir));
+
+describe('POST /api/nodes (pilot_agent mode)', () => {
+  it('mints an enrollment token and returns a docker run command', async () => {
+    const res = await request(app)
+      .post('/api/nodes')
+      .set('Cookie', adminCookie)
+      .send({ name: 'pilot-1', type: 'remote', mode: 'pilot_agent' });
+
+    expect(res.status).toBe(200);
+    expect(res.body.success).toBe(true);
+    expect(res.body.enrollment).toBeDefined();
+    expect(res.body.enrollment.token).toMatch(/^[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$/);
+    expect(res.body.enrollment.dockerRun).toContain('SENCHO_MODE=pilot');
+    expect(res.body.enrollment.dockerRun).toContain(`SENCHO_ENROLL_TOKEN=${res.body.enrollment.token}`);
+    expect(res.body.enrollment.expiresAt).toBeGreaterThan(Date.now());
+  });
+
+  it('persists the token hash, not the raw token', async () => {
+    const res = await request(app)
+      .post('/api/nodes')
+      .set('Cookie', adminCookie)
+      .send({ name: 'pilot-2', type: 'remote', mode: 'pilot_agent' });
+
+    const expectedHash = crypto.createHash('sha256').update(res.body.enrollment.token).digest('hex');
+    const row = DatabaseService.getInstance().getDb()
+      .prepare('SELECT token_hash, used_at FROM pilot_enrollments WHERE node_id = ?')
+      .get(res.body.id) as { token_hash: string; used_at: number | null } | undefined;
+
+    expect(row).toBeDefined();
+    expect(row?.token_hash).toBe(expectedHash);
+    expect(row?.used_at).toBeNull();
+    // The raw token must not appear anywhere in the row.
+    expect(row?.token_hash).not.toBe(res.body.enrollment.token);
+  });
+
+  it('rejects unauthenticated callers with 401', async () => {
+    const res = await request(app)
+      .post('/api/nodes')
+      .send({ name: 'pilot-anon', type: 'remote', mode: 'pilot_agent' });
+    expect(res.status).toBe(401);
+  });
+});
+
+describe('POST /api/nodes/:id/pilot/enroll', () => {
+  it('regenerates the enrollment for an existing pilot node', async () => {
+    const create = await request(app)
+      .post('/api/nodes')
+      .set('Cookie', adminCookie)
+      .send({ name: 'pilot-regen', type: 'remote', mode: 'pilot_agent' });
+    const original = create.body.enrollment.token;
+
+    const regen = await request(app)
+      .post(`/api/nodes/${create.body.id}/pilot/enroll`)
+      .set('Cookie', adminCookie);
+
+    expect(regen.status).toBe(200);
+    expect(regen.body.enrollment.token).not.toBe(original);
+  });
+
+  it('rejects regeneration for proxy-mode remote nodes', async () => {
+    const create = await request(app)
+      .post('/api/nodes')
+      .set('Cookie', adminCookie)
+      .send({
+        name: 'proxy-not-pilot',
+        type: 'remote',
+        api_url: 'http://192.168.1.50:1852',
+        api_token: 'tok',
+      });
+
+    const regen = await request(app)
+      .post(`/api/nodes/${create.body.id}/pilot/enroll`)
+      .set('Cookie', adminCookie);
+
+    expect(regen.status).toBe(400);
+    expect(regen.body.error).toMatch(/pilot/i);
+  });
+
+  it('rejects unknown nodes with 404', async () => {
+    const res = await request(app)
+      .post('/api/nodes/999999/pilot/enroll')
+      .set('Cookie', adminCookie);
+    expect(res.status).toBe(404);
+  });
+});
+
+describe('consumePilotEnrollment replay protection', () => {
+  it('marks the row used and rejects the second consume', () => {
+    const db = DatabaseService.getInstance();
+    const create = db.addNode({
+      name: 'pilot-replay',
+      type: 'remote',
+      compose_dir: '/tmp/x',
+      mode: 'pilot_agent',
+      is_default: false,
+      api_url: '',
+      api_token: '',
+    });
+    const tokenHash = crypto.createHash('sha256').update('synthetic-test-token').digest('hex');
+    db.createPilotEnrollment(create, tokenHash, Date.now() + 60_000);
+
+    const first = db.consumePilotEnrollment(tokenHash);
+    expect(first).toBeDefined();
+    expect(first?.node_id).toBe(create);
+
+    const second = db.consumePilotEnrollment(tokenHash);
+    expect(second).toBeUndefined();
+  });
+
+  it('rejects expired enrollments', () => {
+    const db = DatabaseService.getInstance();
+    const nodeId = db.addNode({
+      name: 'pilot-expired',
+      type: 'remote',
+      compose_dir: '/tmp/x',
+      mode: 'pilot_agent',
+      is_default: false,
+      api_url: '',
+      api_token: '',
+    });
+    const tokenHash = crypto.createHash('sha256').update('expired-test-token').digest('hex');
+    db.createPilotEnrollment(nodeId, tokenHash, Date.now() - 1_000);
+
+    expect(db.consumePilotEnrollment(tokenHash)).toBeUndefined();
+  });
+
+  it('rejects unknown token hashes', () => {
+    const db = DatabaseService.getInstance();
+    expect(db.consumePilotEnrollment('0'.repeat(64))).toBeUndefined();
+  });
+});
+
+describe('Enrollment rate limiter wiring', () => {
+  it('POST /api/nodes (pilot mode) advertises a tighter limit than the global limiter', async () => {
+    const res = await request(app)
+      .post('/api/nodes')
+      .set('Cookie', adminCookie)
+      .send({ name: 'pilot-headers', type: 'remote', mode: 'pilot_agent' });
+
+    const limit = parseInt(res.headers['ratelimit-limit'], 10);
+    // Dev limit is 100/min for the enrollment limiter; global is 1000/min.
+    expect(limit).toBe(100);
+  });
+
+  it('POST /api/nodes (proxy mode) falls back to the global limiter', async () => {
+    const res = await request(app)
+      .post('/api/nodes')
+      .set('Cookie', adminCookie)
+      .send({
+        name: 'proxy-global-limit',
+        type: 'remote',
+        api_url: 'http://192.168.1.51:1852',
+        api_token: 'tok',
+      });
+
+    const limit = parseInt(res.headers['ratelimit-limit'], 10);
+    expect(limit).toBe(1000);
+  });
+
+  it('POST /api/nodes/:id/pilot/enroll uses the enrollment limiter', async () => {
+    const create = await request(app)
+      .post('/api/nodes')
+      .set('Cookie', adminCookie)
+      .send({ name: 'pilot-regen-headers', type: 'remote', mode: 'pilot_agent' });
+
+    const regen = await request(app)
+      .post(`/api/nodes/${create.body.id}/pilot/enroll`)
+      .set('Cookie', adminCookie);
+
+    const limit = parseInt(regen.headers['ratelimit-limit'], 10);
+    expect(limit).toBe(100);
+  });
+});
diff --git a/backend/src/__tests__/pilot-manager-cap.test.ts b/backend/src/__tests__/pilot-manager-cap.test.ts
new file mode 100644
index 00000000..54c124c4
--- /dev/null
+++ b/backend/src/__tests__/pilot-manager-cap.test.ts
@@ -0,0 +1,116 @@
+/**
+ * Tests for PilotTunnelManager system-wide cap (M3) and the metrics
+ * snapshot exposed via /api/system/pilot-tunnels (M2).
+ */
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { EventEmitter } from 'events';
+import { WebSocket } from 'ws';
+import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
+import { PilotTunnelManager, PilotTunnelCapacityError } from '../services/PilotTunnelManager';
+import { PilotMetrics } from '../services/PilotMetrics';
+
+let tmpDir: string;
+let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
+
+beforeAll(async () => {
+    tmpDir = await setupTestDb();
+    ({ DatabaseService } = await import('../services/DatabaseService'));
+});
+
+afterAll(() => cleanupTestDb(tmpDir));
+
+function makeMockTunnelWs(): EventEmitter & {
+    sent: unknown[];
+    readyState: number;
+    bufferedAmount: number;
+    send: (data: unknown) => void;
+    ping: () => void;
+    close: () => void;
+} {
+    const ws = new EventEmitter() as EventEmitter & {
+        sent: unknown[];
+        readyState: number;
+        bufferedAmount: number;
+        send: (data: unknown) => void;
+        ping: () => void;
+        close: () => void;
+    };
+    ws.sent = [];
+    ws.readyState = WebSocket.OPEN;
+    ws.bufferedAmount = 0;
+    ws.send = (data: unknown) => { ws.sent.push(data); };
+    ws.ping = () => { /* no-op */ };
+    ws.close = () => { ws.readyState = WebSocket.CLOSED; ws.emit('close'); };
+    return ws;
+}
+
+describe('PilotTunnelCapacityError', () => {
+    it('carries the limit that triggered the rejection', () => {
+        const err = new PilotTunnelCapacityError(256);
+        expect(err.name).toBe('PilotTunnelCapacityError');
+        expect(err.limit).toBe(256);
+        expect(err.message).toContain('256');
+    });
+});
+
+describe('PilotMetrics counters', () => {
+    beforeEach(() => {
+        // Counters are process-singleton; we verify deltas rather than
+        // absolute values so the test is order-independent.
+    });
+
+    it('increment then snapshot returns the bumped value', () => {
+        const before = PilotMetrics.snapshot();
+        PilotMetrics.increment('enroll_acks');
+        const after = PilotMetrics.snapshot();
+        expect(after.enroll_acks).toBe(before.enroll_acks + 1);
+    });
+
+    it('snapshot returns a copy, not the live object', () => {
+        const snap = PilotMetrics.snapshot();
+        const original = snap.tunnels_total;
+        PilotMetrics.increment('tunnels_total');
+        // The earlier snapshot must remain unchanged.
+        expect(snap.tunnels_total).toBe(original);
+    });
+});
+
+describe('PilotTunnelManager.getMetricsSnapshot', () => {
+    it('returns the current open count and counter set', () => {
+        const mgr = PilotTunnelManager.getInstance();
+        const snap = mgr.getMetricsSnapshot();
+
+        expect(snap).toHaveProperty('counters');
+        expect(snap).toHaveProperty('tunnels_open');
+        expect(snap).toHaveProperty('per_node');
+        expect(Array.isArray(snap.per_node)).toBe(true);
+        expect(snap.tunnels_open).toBe(snap.per_node.length);
+    });
+
+    it('reflects per-tunnel breakdown after registerTunnel succeeds', async () => {
+        const mgr = PilotTunnelManager.getInstance();
+        const before = mgr.getMetricsSnapshot();
+
+        // Seed a real pilot-mode node so updateNode does not throw.
+        const nodeId = DatabaseService.getInstance().addNode({
+            name: `pilot-mgr-${Date.now()}`,
+            type: 'remote',
+            mode: 'pilot_agent',
+            compose_dir: '/tmp/x',
+            is_default: false,
+            api_url: '',
+            api_token: '',
+        });
+
+        const ws = makeMockTunnelWs();
+        await mgr.registerTunnel(nodeId, ws as unknown as WebSocket, 'test-1.0.0');
+
+        const after = mgr.getMetricsSnapshot();
+        expect(after.tunnels_open).toBe(before.tunnels_open + 1);
+        expect(after.per_node.some((p) => p.nodeId === nodeId)).toBe(true);
+        expect(after.counters.tunnels_total).toBe(before.counters.tunnels_total + 1);
+
+        // Cleanup so this test does not leak a tunnel into the next.
+        mgr.closeTunnel(nodeId);
+    });
+});
diff --git a/backend/src/middleware/rateLimiters.ts b/backend/src/middleware/rateLimiters.ts
index 69139516..c0c82ff4 100644
--- a/backend/src/middleware/rateLimiters.ts
+++ b/backend/src/middleware/rateLimiters.ts
@@ -145,6 +145,33 @@ export const ssoRateLimiter = rateLimit({
   message: { error: 'Too many SSO attempts. Please try again later.' },
 });
 
+// Pilot enrollment limiter. Enrollment mints a JWT and writes a
+// pilot_enrollments row, so it deserves a stricter ceiling than the global
+// API limiter (200/min). Applied on the regenerate route directly and on
+// `POST /api/nodes` only when the body resolves to pilot_agent mode (the
+// limiter's `skip` function reads the parsed body).
+export const enrollmentLimiter = rateLimit({
+  ...rateLimitBase,
+  max: process.env.NODE_ENV === 'production' ? 10 : 100,
+  keyGenerator: rateLimitKeyGenerator,
+  message: { error: 'Too many enrollment requests. Please try again shortly.' },
+  skip: (req: Request) => {
+    // Only gate `POST /api/nodes` calls that actually create a pilot agent;
+    // proxy-mode node creation falls back to the global limiter. The
+    // dedicated `/pilot/enroll` route applies this limiter unconditionally,
+    // so the body check is bypassed there by skipping the skip when the
+    // path already targets enrollment.
+    if (req.path.endsWith('/pilot/enroll')) return false;
+    // Express.json() runs globally before route handlers in app.ts, so
+    // req.body is parsed by the time this fires. If a future refactor moves
+    // body parsing per-route the worst case is "limiter applies even to
+    // proxy-mode" rather than "limiter is bypassed entirely".
+    if (!req.body) return false;
+    const body = req.body as { mode?: string; type?: string };
+    return !(body.type === 'remote' && body.mode === 'pilot_agent');
+  },
+});
+
 // Trivy install/update limiter. Install + update are expensive (binary download,
 // sha256 verification) so a 10-minute window prevents accidental thrashing.
 export const trivyInstallLimiter = rateLimit({
diff --git a/backend/src/pilot/agent.ts b/backend/src/pilot/agent.ts
index 9c3e916e..d2531921 100644
--- a/backend/src/pilot/agent.ts
+++ b/backend/src/pilot/agent.ts
@@ -6,8 +6,11 @@ import WebSocket from 'ws';
 import { getSenchoVersion } from '../services/CapabilityRegistry';
 import {
     BinaryFrameType,
+    MAX_FRAME_SIZE_BYTES,
+    MAX_STREAMS_PER_TUNNEL,
     MeshErrCode,
     PROTOCOL_VERSION,
+    STREAM_IDLE_TIMEOUT_MS,
     decodeBinaryFrame,
     decodeJsonFrame,
     encodeBinaryFrame,
@@ -16,6 +19,7 @@ import {
     wsDataToString,
 } from './protocol';
 import { sanitizeForLog } from '../utils/safeLog';
+import { isDebugEnabled } from '../utils/debug';
 
 const RECONNECT_MIN_MS = 1_000;
 const RECONNECT_MAX_MS = 60_000;
@@ -69,13 +73,22 @@ class PilotAgent {
     private readonly httpStreams = new Map<number, { req: http.ClientRequest }>();
     private readonly wsStreams = new Map<number, WebSocket>();
     private readonly tcpStreams = new Map<number, MeshTcpStream>();
+    private readonly idleTimers = new Map<number, NodeJS.Timeout>();
     private shuttingDown = false;
     private readonly agentVersion: string;
+    /**
+     * Optional CA bundle read once at agent construction. Cached so that a
+     * later rotation (file renamed, secret rotated) does not surprise the
+     * agent with a process exit on the next reconnect; container restart is
+     * the documented way to pick up a new CA bundle.
+     */
+    private readonly customCa: Buffer | null;
 
     constructor(options: AgentOptions) {
         this.options = options;
         this.token = options.initialToken;
         this.agentVersion = getSenchoVersion() || '0.0.0';
+        this.customCa = readPilotCaBundle();
     }
 
     public start(): void {
@@ -101,12 +114,24 @@ class PilotAgent {
                 'x-sencho-agent-version': this.agentVersion,
             },
             handshakeTimeout: 15_000,
+            maxPayload: MAX_FRAME_SIZE_BYTES,
+            // Self-signed deployments can supply an internal CA bundle via
+            // SENCHO_PILOT_CA_FILE; rejectUnauthorized stays true. There is
+            // intentionally no env var to disable TLS verification — that
+            // would defeat the entire trust model of the tunnel credential.
+            // The bundle is read once at agent construction (this.customCa);
+            // rotate by restarting the container.
+            ...(this.customCa ? { ca: this.customCa } : {}),
         });
         this.ws = ws;
 
         ws.on('open', () => {
-            console.log('[Pilot] Tunnel connected to', this.options.primaryUrl);
-            this.backoff = RECONNECT_MIN_MS;
+            // Backoff intentionally NOT reset here: a TCP-level connect that
+            // immediately fails the protocol handshake (incompatible version,
+            // bad token consumed at upgrade) would otherwise reset the
+            // backoff and tight-loop reconnects. The reset moves to the
+            // handleJsonFrame 'hello' case once we have a clean handshake.
+            console.log('[Pilot] Tunnel connected to', sanitizeForLog(this.options.primaryUrl));
             try {
                 ws.send(encodeJsonFrame({
                     t: 'hello',
@@ -150,6 +175,58 @@ class PilotAgent {
             try { stream.socket.destroy(); } catch { /* ignore */ }
         }
         this.tcpStreams.clear();
+        for (const [, timer] of this.idleTimers) clearTimeout(timer);
+        this.idleTimers.clear();
+    }
+
+    private streamCount(): number {
+        return this.httpStreams.size + this.wsStreams.size + this.tcpStreams.size;
+    }
+
+    private refreshIdleTimer(streamId: number): void {
+        const existing = this.idleTimers.get(streamId);
+        if (existing) clearTimeout(existing);
+        const timer = setTimeout(() => this.onStreamIdle(streamId), STREAM_IDLE_TIMEOUT_MS);
+        this.idleTimers.set(streamId, timer);
+    }
+
+    private clearIdleTimer(streamId: number): void {
+        const timer = this.idleTimers.get(streamId);
+        if (timer) {
+            clearTimeout(timer);
+            this.idleTimers.delete(streamId);
+        }
+    }
+
+    private onStreamIdle(streamId: number): void {
+        this.idleTimers.delete(streamId);
+        const ws = this.ws;
+        const httpEntry = this.httpStreams.get(streamId);
+        if (httpEntry) {
+            try { httpEntry.req.destroy(); } catch { /* ignore */ }
+            this.httpStreams.delete(streamId);
+            if (ws) {
+                try { ws.send(encodeJsonFrame({ t: 'http_err', s: streamId, code: 'timeout', message: 'agent idle timeout' })); } catch { /* ignore */ }
+            }
+            return;
+        }
+        const wsEntry = this.wsStreams.get(streamId);
+        if (wsEntry) {
+            try { wsEntry.close(1001, 'idle'); } catch { /* ignore */ }
+            this.wsStreams.delete(streamId);
+            if (ws) {
+                try { ws.send(encodeJsonFrame({ t: 'ws_close', s: streamId, code: 1001, reason: 'idle' })); } catch { /* ignore */ }
+            }
+            return;
+        }
+        const tcpEntry = this.tcpStreams.get(streamId);
+        if (tcpEntry) {
+            try { tcpEntry.socket.destroy(); } catch { /* ignore */ }
+            this.tcpStreams.delete(streamId);
+            if (ws) {
+                try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ }
+            }
+        }
     }
 
     private scheduleReconnect(): void {
@@ -176,7 +253,9 @@ class PilotAgent {
                 this.handleJsonFrame(frame);
             }
         } catch (err) {
-            console.warn('[Pilot] Malformed frame from primary:', sanitizeForLog((err as Error).message));
+            // Per-frame; diag-gated to avoid log floods from a misbehaving
+            // primary or a malformed frame arriving in a tight loop.
+            if (isDebugEnabled()) console.warn('[Pilot:diag] Malformed frame from primary:', sanitizeForLog((err as Error).message));
         }
     }
 
@@ -191,6 +270,11 @@ class PilotAgent {
                     try { ws.close(1002, 'incompatible version'); } catch { /* ignore */ }
                     process.exit(1);
                 }
+                // Clean handshake: it is now safe to reset the reconnect
+                // backoff. Doing this earlier (in 'open') would let a peer
+                // that always rejects the handshake drive us into a tight
+                // reconnect loop.
+                this.backoff = RECONNECT_MIN_MS;
                 break;
             }
             case 'ctrl': {
@@ -220,18 +304,21 @@ class PilotAgent {
                 const entry = this.httpStreams.get(frame.streamId);
                 if (!entry) return;
                 try { entry.req.write(frame.payload); } catch { /* ignore */ }
+                this.refreshIdleTimer(frame.streamId);
                 break;
             }
             case BinaryFrameType.WsMessageBinary: {
                 const ws = this.wsStreams.get(frame.streamId);
                 if (!ws) return;
                 try { ws.send(frame.payload, { binary: true }); } catch { /* ignore */ }
+                this.refreshIdleTimer(frame.streamId);
                 break;
             }
             case BinaryFrameType.TcpData: {
                 const stream = this.tcpStreams.get(frame.streamId);
                 if (!stream) return;
                 try { stream.socket.write(frame.payload); } catch { /* ignore */ }
+                this.refreshIdleTimer(frame.streamId);
                 break;
             }
             default:
@@ -244,6 +331,17 @@ class PilotAgent {
     private onHttpReq(frame: Extract<ReturnType<typeof decodeJsonFrame>, { t: 'http_req' }>): void {
         const ws = this.ws;
         if (!ws) return;
+        if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) {
+            try {
+                ws.send(encodeJsonFrame({
+                    t: 'http_err',
+                    s: frame.s,
+                    code: 'agent_error',
+                    message: 'agent stream cap reached',
+                }));
+            } catch { /* ignore */ }
+            return;
+        }
 
         const req = http.request({
             host: '127.0.0.1',
@@ -265,17 +363,21 @@ class PilotAgent {
                     headers: outHeaders,
                 }));
             } catch { /* ignore */ }
+            this.refreshIdleTimer(frame.s);
 
             res.on('data', (chunk: Buffer) => {
                 try { ws.send(encodeBinaryFrame(BinaryFrameType.HttpResBody, frame.s, chunk), { binary: true }); } catch { /* ignore */ }
+                this.refreshIdleTimer(frame.s);
             });
             res.on('end', () => {
                 try { ws.send(encodeJsonFrame({ t: 'http_res_end', s: frame.s })); } catch { /* ignore */ }
                 this.httpStreams.delete(frame.s);
+                this.clearIdleTimer(frame.s);
             });
             res.on('error', () => {
                 try { ws.send(encodeJsonFrame({ t: 'http_err', s: frame.s, code: 'bad_response', message: 'upstream error' })); } catch { /* ignore */ }
                 this.httpStreams.delete(frame.s);
+                this.clearIdleTimer(frame.s);
             });
         });
 
@@ -289,15 +391,18 @@ class PilotAgent {
                 }));
             } catch { /* ignore */ }
             this.httpStreams.delete(frame.s);
+            this.clearIdleTimer(frame.s);
         });
 
         this.httpStreams.set(frame.s, { req });
+        this.refreshIdleTimer(frame.s);
     }
 
     private onHttpReqEnd(streamId: number): void {
         const entry = this.httpStreams.get(streamId);
         if (!entry) return;
         try { entry.req.end(); } catch { /* ignore */ }
+        this.refreshIdleTimer(streamId);
     }
 
     // --- WebSocket dispatch (tunnel -> loopback) ---
@@ -305,15 +410,28 @@ class PilotAgent {
     private onWsOpen(frame: Extract<ReturnType<typeof decodeJsonFrame>, { t: 'ws_open' }>): void {
         const ws = this.ws;
         if (!ws) return;
+        if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) {
+            try {
+                ws.send(encodeJsonFrame({
+                    t: 'ws_reject',
+                    s: frame.s,
+                    status: 503,
+                    message: 'agent stream cap reached',
+                }));
+            } catch { /* ignore */ }
+            return;
+        }
 
         const target = `ws://127.0.0.1:${this.options.loopbackPort}${frame.path}`;
         const client = new WebSocket(target, {
             headers: { ...frame.headers, host: `127.0.0.1:${this.options.loopbackPort}` },
+            maxPayload: MAX_FRAME_SIZE_BYTES,
         });
 
         client.on('open', () => {
             try { ws.send(encodeJsonFrame({ t: 'ws_accept', s: frame.s, headers: {} })); } catch { /* ignore */ }
             this.wsStreams.set(frame.s, client);
+            this.refreshIdleTimer(frame.s);
         });
         client.on('message', (data, isBinary) => {
             if (isBinary) {
@@ -321,14 +439,17 @@ class PilotAgent {
             } else {
                 try { ws.send(encodeJsonFrame({ t: 'ws_msg_text', s: frame.s, data: wsDataToString(data) ?? '' })); } catch { /* ignore */ }
             }
+            this.refreshIdleTimer(frame.s);
         });
         client.on('close', (code, reason) => {
             try { ws.send(encodeJsonFrame({ t: 'ws_close', s: frame.s, code, reason: reason?.toString?.() })); } catch { /* ignore */ }
             this.wsStreams.delete(frame.s);
+            this.clearIdleTimer(frame.s);
         });
         client.on('error', () => {
             try { ws.send(encodeJsonFrame({ t: 'ws_reject', s: frame.s, status: 502, message: 'agent websocket failed' })); } catch { /* ignore */ }
             this.wsStreams.delete(frame.s);
+            this.clearIdleTimer(frame.s);
         });
     }
 
@@ -336,6 +457,7 @@ class PilotAgent {
         const ws = this.wsStreams.get(streamId);
         if (!ws) return;
         try { ws.send(data); } catch { /* ignore */ }
+        this.refreshIdleTimer(streamId);
     }
 
     private onWsClose(streamId: number, code: number, reason?: string): void {
@@ -343,6 +465,7 @@ class PilotAgent {
         if (!ws) return;
         try { ws.close(code, reason); } catch { /* ignore */ }
         this.wsStreams.delete(streamId);
+        this.clearIdleTimer(streamId);
     }
 
     // --- Sencho Mesh TCP dispatch (tunnel -> Compose service container) ---
@@ -354,6 +477,12 @@ class PilotAgent {
     private async onTcpOpen(frame: Extract<ReturnType<typeof decodeJsonFrame>, { t: 'tcp_open' }>): Promise<void> {
         const ws = this.ws;
         if (!ws) return;
+        if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) {
+            try {
+                ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: 'agent_error' }));
+            } catch { /* ignore */ }
+            return;
+        }
 
         const target = await this.resolveMeshTarget(frame.stack, frame.service, frame.port);
         if (!target.ok) {
@@ -367,6 +496,7 @@ class PilotAgent {
         socket.setTimeout(MESH_CONNECT_TIMEOUT_MS);
         const entry: MeshTcpStream = { socket, accepted: false };
         this.tcpStreams.set(frame.s, entry);
+        this.refreshIdleTimer(frame.s);
 
         const sendAck = (ok: boolean, err?: MeshErrCode) => {
             try { ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok, err })); } catch { /* ignore */ }
@@ -376,17 +506,20 @@ class PilotAgent {
             entry.accepted = true;
             socket.setTimeout(0);
             sendAck(true);
+            this.refreshIdleTimer(frame.s);
         });
         socket.on('data', (chunk: Buffer) => {
             try {
                 ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, frame.s, chunk), { binary: true });
             } catch { /* ignore */ }
+            this.refreshIdleTimer(frame.s);
         });
         socket.on('timeout', () => {
             if (entry.accepted) return;
             entry.accepted = true;
             sendAck(false, 'unreachable');
             this.tcpStreams.delete(frame.s);
+            this.clearIdleTimer(frame.s);
             try { socket.destroy(); } catch { /* ignore */ }
         });
         socket.on('error', (err) => {
@@ -394,15 +527,18 @@ class PilotAgent {
                 entry.accepted = true;
                 sendAck(false, 'unreachable');
                 this.tcpStreams.delete(frame.s);
+                this.clearIdleTimer(frame.s);
                 return;
             }
             console.warn('[Pilot] tcp stream error:', sanitizeForLog(err.message));
             if (this.tcpStreams.delete(frame.s)) {
+                this.clearIdleTimer(frame.s);
                 try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
             }
         });
         socket.on('close', () => {
             if (this.tcpStreams.delete(frame.s)) {
+                this.clearIdleTimer(frame.s);
                 try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ }
             }
         });
@@ -412,6 +548,7 @@ class PilotAgent {
         const entry = this.tcpStreams.get(streamId);
         if (!entry) return;
         this.tcpStreams.delete(streamId);
+        this.clearIdleTimer(streamId);
         try { entry.socket.destroy(); } catch { /* ignore */ }
     }
 
@@ -475,6 +612,24 @@ function readPersistedToken(): string | null {
     return null;
 }
 
+/**
+ * Load the optional CA bundle pointed at by SENCHO_PILOT_CA_FILE so a pilot
+ * agent can verify a self-signed primary cert without disabling TLS
+ * verification globally. Returns the file contents or null if the var is
+ * unset; surfaces a clear error and exits if the file cannot be read so the
+ * operator does not silently fall back to the default trust store.
+ */
+function readPilotCaBundle(): Buffer | null {
+    const caFile = process.env.SENCHO_PILOT_CA_FILE;
+    if (!caFile) return null;
+    try {
+        return fs.readFileSync(caFile);
+    } catch (err) {
+        console.error('[Pilot] Failed to read SENCHO_PILOT_CA_FILE:', sanitizeForLog((err as Error).message));
+        process.exit(1);
+    }
+}
+
 function persistToken(token: string): void {
     try {
         const dir = path.dirname(TOKEN_PATH);
diff --git a/backend/src/pilot/protocol.ts b/backend/src/pilot/protocol.ts
index bbd337ac..1ec79728 100644
--- a/backend/src/pilot/protocol.ts
+++ b/backend/src/pilot/protocol.ts
@@ -17,6 +17,39 @@
 
 export const PROTOCOL_VERSION = 1;
 
+/**
+ * Hard ceiling on a single tunnel WebSocket frame, in bytes. Authoritative
+ * enforcement is at the WebSocket layer: `maxPayload` on the gateway-side
+ * `WebSocketServer` and the agent-side `WebSocket` client both reject
+ * oversized frames before they reach the decoder. The decoder also enforces
+ * the same cap for two narrow cases: (1) tests that build Buffers locally
+ * and skip the WebSocket layer, and (2) defense-in-depth if a future
+ * codepath ever passes user-controlled bytes to the decoder directly.
+ *
+ * 8 MB comfortably accommodates compose YAML, image-list responses, and
+ * exec stream chunks while bounding the decode buffer a buggy or malicious
+ * peer can force the other side to allocate.
+ */
+export const MAX_FRAME_SIZE_BYTES = 8 * 1024 * 1024;
+
+/**
+ * Maximum concurrent multiplexed streams on a single tunnel. Beyond this the
+ * bridge refuses new loopback requests with 503 and the agent rejects new
+ * incoming streams with the appropriate error frame. Sized for normal Sencho
+ * fanout (UI tabs polling stats, logs, stack lifecycle) with substantial
+ * headroom; the realistic ceiling under load is closer to single digits per
+ * tunnel.
+ */
+export const MAX_STREAMS_PER_TUNNEL = 1024;
+
+/**
+ * Per-stream idle timeout. A stream that sees no inbound or outbound activity
+ * for this long is closed and removed from the stream map. Protects against
+ * leaked streams (one side crashed, peer never noticed) leaking memory over
+ * long uptimes.
+ */
+export const STREAM_IDLE_TIMEOUT_MS = 10 * 60 * 1000;
+
 // --- Binary frame types (first byte of a binary WS frame) ---
 
 export enum BinaryFrameType {
@@ -157,6 +190,14 @@ export function encodeJsonFrame(frame: JsonFrame): string {
 }
 
 export function decodeJsonFrame(raw: string): JsonFrame {
+    // Compare against UTF-8 byte length, not the string's UTF-16 code-unit
+    // count: multi-byte payloads can otherwise sneak ~3x the byte budget
+    // through a `raw.length` check. Cheap because Buffer.byteLength does
+    // not allocate.
+    const byteLen = Buffer.byteLength(raw, 'utf8');
+    if (byteLen > MAX_FRAME_SIZE_BYTES) {
+        throw new Error(`json frame too large: ${byteLen} bytes`);
+    }
     const parsed = JSON.parse(raw);
     if (!parsed || typeof parsed !== 'object' || typeof parsed.t !== 'string') {
         throw new Error('invalid frame: missing type discriminator');
@@ -189,6 +230,9 @@ export function decodeBinaryFrame(buf: Buffer): DecodedBinaryFrame {
     if (buf.length < 5) {
         throw new Error(`binary frame too short: ${buf.length} bytes`);
     }
+    if (buf.length > MAX_FRAME_SIZE_BYTES) {
+        throw new Error(`binary frame too large: ${buf.length} bytes`);
+    }
     const type = buf.readUInt8(0) as BinaryFrameType;
     if (type !== BinaryFrameType.HttpReqBody &&
         type !== BinaryFrameType.HttpResBody &&
@@ -228,7 +272,12 @@ export const PilotCloseCode = {
 
 /**
  * Monotonic stream id generator. Primary is the sole allocator.
- * Wraps at 2^31 (well above practical per-tunnel concurrency).
+ *
+ * Wraps at 2^31. With MAX_STREAMS_PER_TUNNEL = 1024 the allocator
+ * cannot collide with a still-live stream during a single tunnel
+ * lifetime: the wrap distance (~2.1 billion) is more than six orders
+ * of magnitude larger than the cap. A new tunnel restarts the
+ * sequence at 1, so cross-tunnel reuse is also harmless.
  */
 export class StreamIdAllocator {
     private next: number;
diff --git a/backend/src/routes/metrics.ts b/backend/src/routes/metrics.ts
index 77f60e0f..18c384fe 100644
--- a/backend/src/routes/metrics.ts
+++ b/backend/src/routes/metrics.ts
@@ -5,6 +5,7 @@ import DockerController, { globalDockerNetwork } from '../services/DockerControl
 import { DatabaseService } from '../services/DatabaseService';
 import { CacheService } from '../services/CacheService';
 import { NodeRegistry } from '../services/NodeRegistry';
+import { PilotTunnelManager } from '../services/PilotTunnelManager';
 import { authMiddleware } from '../middleware/auth';
 import { requireAdmin } from '../middleware/tierGates';
 import { STATS_CACHE_TTL_MS, SYSTEM_STATS_CACHE_TTL_MS } from '../helpers/constants';
@@ -249,3 +250,19 @@ metricsRouter.get('/system/cache-stats', authMiddleware, async (req: Request, re
     res.status(500).json({ error: 'Failed to fetch cache stats' });
   }
 });
+
+/**
+ * Admin-only pilot tunnel observability. Counters reset on process restart
+ * by design (see PilotMetrics.ts). The per_node array is the load-bearing
+ * field: aggregate counters can hide a single tunnel that is flapping or
+ * sitting on a stuck write buffer.
+ */
+metricsRouter.get('/system/pilot-tunnels', authMiddleware, async (req: Request, res: Response): Promise<void> => {
+  if (!requireAdmin(req, res)) return;
+  try {
+    res.json(PilotTunnelManager.getInstance().getMetricsSnapshot());
+  } catch (error) {
+    console.error('Failed to fetch pilot tunnel metrics:', error);
+    res.status(500).json({ error: 'Failed to fetch pilot tunnel metrics' });
+  }
+});
diff --git a/backend/src/routes/nodes.ts b/backend/src/routes/nodes.ts
index cc31ba5c..d0fef48b 100644
--- a/backend/src/routes/nodes.ts
+++ b/backend/src/routes/nodes.ts
@@ -5,6 +5,7 @@ import { authMiddleware } from '../middleware/auth';
 import { requirePermission } from '../middleware/permissions';
 import { rejectApiTokenScope } from '../middleware/apiTokenScope';
 import { requireAdmiral } from '../middleware/tierGates';
+import { enrollmentLimiter } from '../middleware/rateLimiters';
 import { DatabaseService } from '../services/DatabaseService';
 import { NodeRegistry } from '../services/NodeRegistry';
 import { CacheService } from '../services/CacheService';
@@ -120,7 +121,7 @@ nodesRouter.get('/:id', async (req: Request, res: Response) => {
   }
 });
 
-nodesRouter.post('/', async (req: Request, res: Response) => {
+nodesRouter.post('/', enrollmentLimiter, async (req: Request, res: Response) => {
   if (rejectApiTokenScope(req, res, NODE_SCOPE_MESSAGE)) return;
   if (!requirePermission(req, res, 'node:manage')) return;
   try {
@@ -190,7 +191,7 @@ nodesRouter.post('/', async (req: Request, res: Response) => {
   }
 });
 
-nodesRouter.post('/:id/pilot/enroll', async (req: Request, res: Response) => {
+nodesRouter.post('/:id/pilot/enroll', enrollmentLimiter, async (req: Request, res: Response) => {
   if (rejectApiTokenScope(req, res, NODE_SCOPE_MESSAGE)) return;
   const nodeIdStr = req.params.id as string;
   if (!requirePermission(req, res, 'node:manage', 'node', nodeIdStr)) return;
diff --git a/backend/src/server.ts b/backend/src/server.ts
index 4e9423ef..a1b48eff 100644
--- a/backend/src/server.ts
+++ b/backend/src/server.ts
@@ -1,6 +1,7 @@
 import http from 'http';
 import type { Express } from 'express';
 import { WebSocketServer } from 'ws';
+import { MAX_FRAME_SIZE_BYTES } from './pilot/protocol';
 
 export interface SenchoServer {
   server: http.Server;
@@ -21,8 +22,12 @@ export function createServer(app: Express): SenchoServer {
   const wss = new WebSocketServer({ noServer: true });
 
   // Agents dial /api/pilot/tunnel; the handshake verifies a pilot_enroll or
-  // pilot_tunnel JWT, then hands the socket off to PilotTunnelManager.
-  const pilotTunnelWss = new WebSocketServer({ noServer: true });
+  // pilot_tunnel JWT, then hands the socket off to PilotTunnelManager. The
+  // pilot tunnel multiplexes every HTTP / WS / Mesh-TCP byte for a remote
+  // node, so a single oversized frame from a buggy or malicious agent can
+  // bloat the gateway's decode buffer; cap the payload at the protocol-level
+  // ceiling rather than relying on the ws default (100 MB).
+  const pilotTunnelWss = new WebSocketServer({ noServer: true, maxPayload: MAX_FRAME_SIZE_BYTES });
 
   return { server, wss, pilotTunnelWss };
 }
diff --git a/backend/src/services/PilotMetrics.ts b/backend/src/services/PilotMetrics.ts
new file mode 100644
index 00000000..42e5976f
--- /dev/null
+++ b/backend/src/services/PilotMetrics.ts
@@ -0,0 +1,39 @@
+/**
+ * PilotMetrics: in-memory counters for the pilot-agent reverse-tunnel
+ * subsystem. Strictly process-local — Sencho does not export metrics to any
+ * external sink (privacy posture, see CLAUDE.md). Counters reset on process
+ * restart by design; their purpose is operator support and debug, not
+ * long-term trend analysis.
+ *
+ * No general in-process metrics facility exists in the backend today, so this
+ * is a per-feature pattern. When a shared facility lands, this module should
+ * be replaced by an instance of it rather than grown.
+ */
+
+interface Counters {
+    tunnels_total: number;
+    tunnels_replaced: number;
+    tunnels_rejected_capacity: number;
+    enroll_acks: number;
+    frame_decode_errors: number;
+}
+
+class PilotMetricsImpl {
+    private counters: Counters = {
+        tunnels_total: 0,
+        tunnels_replaced: 0,
+        tunnels_rejected_capacity: 0,
+        enroll_acks: 0,
+        frame_decode_errors: 0,
+    };
+
+    public increment<K extends keyof Counters>(name: K): void {
+        this.counters[name] += 1;
+    }
+
+    public snapshot(): Counters {
+        return { ...this.counters };
+    }
+}
+
+export const PilotMetrics = new PilotMetricsImpl();
diff --git a/backend/src/services/PilotTunnelBridge.ts b/backend/src/services/PilotTunnelBridge.ts
index 79f8029c..d4ef2656 100644
--- a/backend/src/services/PilotTunnelBridge.ts
+++ b/backend/src/services/PilotTunnelBridge.ts
@@ -5,6 +5,8 @@ import { WebSocket, WebSocketServer } from 'ws';
 import {
     BinaryFrameType,
     DecodedBinaryFrame,
+    MAX_STREAMS_PER_TUNNEL,
+    STREAM_IDLE_TIMEOUT_MS,
     StreamIdAllocator,
     decodeBinaryFrame,
     decodeJsonFrame,
@@ -13,17 +15,32 @@ import {
     wsDataToBuffer,
     wsDataToString,
 } from '../pilot/protocol';
+import { isDebugEnabled } from '../utils/debug';
+import { sanitizeForLog } from '../utils/safeLog';
+import { PilotMetrics } from './PilotMetrics';
 
 const BUFFER_HIGH_WATER_MARK = 4 * 1024 * 1024;
 const PING_INTERVAL_MS = 30_000;
+/**
+ * Poll cadence for the backpressure drain check. The `ws` WebSocket does not
+ * surface a usable 'drain' event, so when at least one stream is paused we
+ * sample `bufferedAmount` at this rate and resume / fan out 'drain' as soon
+ * as the buffer drops below the high-water mark. Dormant when nothing is
+ * paused, so steady-state cost is zero.
+ */
+const DRAIN_CHECK_INTERVAL_MS = 100;
 
-interface HttpStreamState {
+interface StreamMeta {
+    idleTimer?: NodeJS.Timeout;
+}
+
+interface HttpStreamState extends StreamMeta {
     kind: 'http';
     res: ServerResponse;
     headersWritten: boolean;
 }
 
-interface WsStreamState {
+interface WsStreamState extends StreamMeta {
     kind: 'ws';
     rawSocket?: Socket;
     rawHead?: Buffer;
@@ -31,7 +48,7 @@ interface WsStreamState {
     clientWs?: WebSocket;
 }
 
-interface TcpStreamState {
+interface TcpStreamState extends StreamMeta {
     kind: 'tcp';
     handle: TcpStream;
     bytesIn: number;
@@ -104,8 +121,11 @@ export class PilotTunnelBridge extends EventEmitter {
     private readonly streamIds = new StreamIdAllocator();
     private readonly streams = new Map<number, StreamState>();
     private readonly connectedAt = Date.now();
+    private readonly pausedReqs = new Map<number, IncomingMessage>();
+    private readonly tcpAwaitingDrain = new Set<number>();
     private loopbackUrl = '';
     private pingTimer?: NodeJS.Timeout;
+    private drainTimer?: NodeJS.Timeout;
     private closed = false;
 
     constructor(_nodeId: number, tunnelWs: WebSocket) {
@@ -143,14 +163,6 @@ export class PilotTunnelBridge extends EventEmitter {
         this.pingTimer = setInterval(() => {
             if (this.tunnelWs.readyState !== WebSocket.OPEN) return;
             try { this.tunnelWs.ping(); } catch { /* surfaced via 'error' */ }
-            // Coarse drain fan-out for TCP streams: ws does not expose a
-            // socket-level 'drain' we can hook, so we let backpressure clear
-            // by the next ping cycle.
-            if (this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK) {
-                for (const s of this.streams.values()) {
-                    if (s.kind === 'tcp' && s.accepted) s.handle.emit('drain');
-                }
-            }
         }, PING_INTERVAL_MS);
     }
 
@@ -166,16 +178,19 @@ export class PilotTunnelBridge extends EventEmitter {
      */
     public openTcpStream(target: { stack: string; service: string; port: number }): TcpStream | null {
         if (!this.isOpen()) return null;
+        if (this.streams.size >= MAX_STREAMS_PER_TUNNEL) return null;
         const streamId = this.streamIds.allocate();
         const handle = new TcpStream(streamId, this);
-        this.streams.set(streamId, {
+        const state: TcpStreamState = {
             kind: 'tcp',
             handle,
             bytesIn: 0,
             bytesOut: 0,
             openedAt: Date.now(),
             accepted: false,
-        });
+        };
+        this.streams.set(streamId, state);
+        this.refreshIdleTimer(streamId, state);
         this.sendJson({
             t: 'tcp_open',
             s: streamId,
@@ -197,13 +212,22 @@ export class PilotTunnelBridge extends EventEmitter {
         if (!this.isOpen()) return false;
         this.sendBinary(BinaryFrameType.TcpData, streamId, payload);
         s.bytesOut += payload.length;
-        return this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK;
+        this.refreshIdleTimer(streamId, s);
+        const ok = this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK;
+        if (!ok) {
+            // Backpressure: caller will pause until 'drain'. Track this TCP
+            // stream so checkDrain knows to fire 'drain' on it (and so the
+            // drain timer keeps running for TCP-only backpressure scenarios).
+            this.tcpAwaitingDrain.add(streamId);
+            this.ensureDrainTimer();
+        }
+        return ok;
     }
 
     /** @internal Called only by TcpStream.end / .destroy. */
     public _closeTcpStream(streamId: number): void {
         if (!this.streams.has(streamId)) return;
-        this.streams.delete(streamId);
+        this.removeStream(streamId);
         this.sendJson({ t: 'tcp_close', s: streamId });
     }
 
@@ -225,8 +249,21 @@ export class PilotTunnelBridge extends EventEmitter {
         if (this.closed) return;
         this.closed = true;
         if (this.pingTimer) { clearInterval(this.pingTimer); this.pingTimer = undefined; }
+        this.stopDrainTimer();
+        // Resume any paused IncomingMessage so its end event can fire and
+        // its parser unwinds; the loopback close below will tear down the
+        // socket either way, but this avoids holding a dangling parser
+        // state across teardown.
+        for (const [, req] of this.pausedReqs) {
+            try { req.resume(); } catch { /* ignore */ }
+        }
+        this.pausedReqs.clear();
+        this.tcpAwaitingDrain.clear();
 
-        for (const [, state] of this.streams) this.teardownStream(state);
+        for (const [, state] of this.streams) {
+            this.clearIdleTimer(state);
+            this.teardownStream(state);
+        }
         this.streams.clear();
 
         try { this.tunnelWs.close(code, reason); } catch { /* ignore */ }
@@ -243,9 +280,17 @@ export class PilotTunnelBridge extends EventEmitter {
             res.end('pilot tunnel not ready');
             return;
         }
+        if (this.streams.size >= MAX_STREAMS_PER_TUNNEL) {
+            res.statusCode = 503;
+            res.setHeader('content-type', 'text/plain');
+            res.end('pilot tunnel: stream cap reached');
+            return;
+        }
 
         const streamId = this.streamIds.allocate();
-        this.streams.set(streamId, { kind: 'http', res, headersWritten: false });
+        const state: HttpStreamState = { kind: 'http', res, headersWritten: false };
+        this.streams.set(streamId, state);
+        this.refreshIdleTimer(streamId, state);
 
         const headers: Record<string, string> = {};
         for (const [k, v] of Object.entries(req.headers)) {
@@ -262,9 +307,13 @@ export class PilotTunnelBridge extends EventEmitter {
         });
 
         req.on('data', (chunk: Buffer) => {
-            if (!this.streams.has(streamId)) return;
+            const s = this.streams.get(streamId);
+            if (!s) return;
             this.sendBinary(BinaryFrameType.HttpReqBody, streamId, chunk);
-            if (this.tunnelWs.bufferedAmount > BUFFER_HIGH_WATER_MARK) req.pause();
+            this.refreshIdleTimer(streamId, s);
+            if (this.tunnelWs.bufferedAmount > BUFFER_HIGH_WATER_MARK) {
+                this.pauseRequest(streamId, req);
+            }
         });
         req.on('end', () => {
             if (!this.streams.has(streamId)) return;
@@ -273,13 +322,13 @@ export class PilotTunnelBridge extends EventEmitter {
         req.on('error', () => {
             const s = this.streams.get(streamId);
             if (s) this.teardownStream(s);
-            this.streams.delete(streamId);
+            this.removeStream(streamId);
         });
 
         res.on('close', () => {
             // Client disconnected before response finished.
             if (this.streams.has(streamId)) {
-                this.streams.delete(streamId);
+                this.removeStream(streamId);
                 this.sendJson({ t: 'http_err', s: streamId, code: 'tunnel_down', message: 'client aborted' });
             }
         });
@@ -291,14 +340,21 @@ export class PilotTunnelBridge extends EventEmitter {
             socket.destroy();
             return;
         }
+        if (this.streams.size >= MAX_STREAMS_PER_TUNNEL) {
+            socket.write('HTTP/1.1 503 Service Unavailable\r\n\r\n');
+            socket.destroy();
+            return;
+        }
 
         const streamId = this.streamIds.allocate();
-        this.streams.set(streamId, {
+        const state: WsStreamState = {
             kind: 'ws',
             rawSocket: socket,
             rawHead: head,
             upgradeRequest: req,
-        });
+        };
+        this.streams.set(streamId, state);
+        this.refreshIdleTimer(streamId, state);
 
         const headers: Record<string, string> = {};
         for (const [k, v] of Object.entries(req.headers)) {
@@ -316,12 +372,12 @@ export class PilotTunnelBridge extends EventEmitter {
         socket.on('error', () => {
             const s = this.streams.get(streamId);
             if (s) this.teardownStream(s);
-            this.streams.delete(streamId);
+            this.removeStream(streamId);
         });
         socket.on('close', () => {
             if (this.streams.has(streamId)) {
                 this.sendJson({ t: 'ws_close', s: streamId, code: 1006, reason: 'client closed' });
-                this.streams.delete(streamId);
+                this.removeStream(streamId);
             }
         });
     }
@@ -341,8 +397,14 @@ export class PilotTunnelBridge extends EventEmitter {
                 const frame = decodeJsonFrame(text);
                 this.handleJsonFrame(frame);
             }
-        } catch {
-            // Malformed frame: kill the tunnel to force re-sync.
+        } catch (err) {
+            // Malformed frame: kill the tunnel to force re-sync. Diag-gated
+            // so a flood of malformed frames cannot drown the log; the
+            // tunnel close itself is the loud signal.
+            PilotMetrics.increment('frame_decode_errors');
+            if (isDebugEnabled()) {
+                console.warn('[PilotBridge:diag] Malformed frame from agent:', sanitizeForLog((err as Error).message));
+            }
             this.close(1002, 'protocol error');
         }
     }
@@ -358,13 +420,14 @@ export class PilotTunnelBridge extends EventEmitter {
                     } catch { /* headers already sent or invalid */ }
                     s.headersWritten = true;
                 }
+                this.refreshIdleTimer(frame.s, s);
                 break;
             }
             case 'http_res_end': {
                 const s = this.streams.get(frame.s);
                 if (!s || s.kind !== 'http') return;
                 try { s.res.end(); } catch { /* ignore */ }
-                this.streams.delete(frame.s);
+                this.removeStream(frame.s);
                 break;
             }
             case 'http_err': {
@@ -378,7 +441,7 @@ export class PilotTunnelBridge extends EventEmitter {
                 } else {
                     this.teardownStream(s);
                 }
-                this.streams.delete(frame.s);
+                this.removeStream(frame.s);
                 break;
             }
             case 'ws_accept': {
@@ -388,7 +451,10 @@ export class PilotTunnelBridge extends EventEmitter {
                     s.clientWs = ws;
                     s.rawSocket = undefined;
                     s.rawHead = undefined;
+                    this.refreshIdleTimer(frame.s, s);
                     ws.on('message', (msg, isBin) => {
+                        const cur = this.streams.get(frame.s);
+                        if (cur) this.refreshIdleTimer(frame.s, cur);
                         if (isBin) {
                             this.sendBinary(BinaryFrameType.WsMessageBinary, frame.s, wsDataToBuffer(msg) ?? Buffer.alloc(0));
                         } else {
@@ -398,11 +464,11 @@ export class PilotTunnelBridge extends EventEmitter {
                     ws.on('close', (code, reason) => {
                         if (this.streams.has(frame.s)) {
                             this.sendJson({ t: 'ws_close', s: frame.s, code, reason: reason?.toString?.() });
-                            this.streams.delete(frame.s);
+                            this.removeStream(frame.s);
                         }
                     });
                     ws.on('error', () => {
-                        if (this.streams.has(frame.s)) this.streams.delete(frame.s);
+                        if (this.streams.has(frame.s)) this.removeStream(frame.s);
                     });
                 });
                 break;
@@ -414,13 +480,14 @@ export class PilotTunnelBridge extends EventEmitter {
                     s.rawSocket.write(`HTTP/1.1 ${frame.status} ${frame.message}\r\n\r\n`);
                     s.rawSocket.destroy();
                 } catch { /* ignore */ }
-                this.streams.delete(frame.s);
+                this.removeStream(frame.s);
                 break;
             }
             case 'ws_msg_text': {
                 const s = this.streams.get(frame.s);
                 if (!s || s.kind !== 'ws' || !s.clientWs) return;
                 try { s.clientWs.send(frame.data); } catch { /* ignore */ }
+                this.refreshIdleTimer(frame.s, s);
                 break;
             }
             case 'ws_close': {
@@ -431,7 +498,7 @@ export class PilotTunnelBridge extends EventEmitter {
                 } else if (s.rawSocket) {
                     try { s.rawSocket.destroy(); } catch { /* ignore */ }
                 }
-                this.streams.delete(frame.s);
+                this.removeStream(frame.s);
                 break;
             }
             case 'ctrl': {
@@ -445,9 +512,10 @@ export class PilotTunnelBridge extends EventEmitter {
                 if (!s || s.kind !== 'tcp') return;
                 if (frame.ok) {
                     s.accepted = true;
+                    this.refreshIdleTimer(frame.s, s);
                     s.handle.emit('open');
                 } else {
-                    this.streams.delete(frame.s);
+                    this.removeStream(frame.s);
                     s.handle.emit('error', new Error(frame.err ?? 'tcp_open rejected'));
                     s.handle.emit('close');
                 }
@@ -456,7 +524,7 @@ export class PilotTunnelBridge extends EventEmitter {
             case 'tcp_close': {
                 const s = this.streams.get(frame.s);
                 if (!s || s.kind !== 'tcp') return;
-                this.streams.delete(frame.s);
+                this.removeStream(frame.s);
                 s.handle.emit('close');
                 break;
             }
@@ -478,11 +546,13 @@ export class PilotTunnelBridge extends EventEmitter {
                     s.headersWritten = true;
                 }
                 try { s.res.write(frame.payload); } catch { /* ignore */ }
+                this.refreshIdleTimer(frame.streamId, s);
                 break;
             }
             case BinaryFrameType.WsMessageBinary: {
                 if (s.kind !== 'ws' || !s.clientWs) return;
                 try { s.clientWs.send(frame.payload, { binary: true }); } catch { /* ignore */ }
+                this.refreshIdleTimer(frame.streamId, s);
                 break;
             }
             case BinaryFrameType.HttpReqBody:
@@ -491,6 +561,7 @@ export class PilotTunnelBridge extends EventEmitter {
             case BinaryFrameType.TcpData: {
                 if (s.kind !== 'tcp') return;
                 s.bytesIn += frame.payload.length;
+                this.refreshIdleTimer(frame.streamId, s);
                 s.handle.emit('data', frame.payload);
                 break;
             }
@@ -506,6 +577,84 @@ export class PilotTunnelBridge extends EventEmitter {
 
     // --- Helpers ---
 
+    private pauseRequest(streamId: number, req: IncomingMessage): void {
+        if (this.pausedReqs.has(streamId)) return;
+        try { req.pause(); } catch { /* ignore */ }
+        this.pausedReqs.set(streamId, req);
+        this.ensureDrainTimer();
+    }
+
+    private ensureDrainTimer(): void {
+        if (this.drainTimer || this.closed) return;
+        this.drainTimer = setInterval(() => this.checkDrain(), DRAIN_CHECK_INTERVAL_MS);
+    }
+
+    private checkDrain(): void {
+        if (this.closed) {
+            this.stopDrainTimer();
+            return;
+        }
+        if (this.tunnelWs.bufferedAmount > BUFFER_HIGH_WATER_MARK) return;
+        for (const [, req] of this.pausedReqs) {
+            try { req.resume(); } catch { /* ignore */ }
+        }
+        this.pausedReqs.clear();
+        // Fire 'drain' only on TCP streams that signaled backpressure; do not
+        // wake every accepted TCP stream because that violates the documented
+        // event contract on TcpStream.
+        for (const streamId of this.tcpAwaitingDrain) {
+            const s = this.streams.get(streamId);
+            if (s && s.kind === 'tcp' && s.accepted) s.handle.emit('drain');
+        }
+        this.tcpAwaitingDrain.clear();
+        this.stopDrainTimer();
+    }
+
+    private stopDrainTimer(): void {
+        if (this.drainTimer) {
+            clearInterval(this.drainTimer);
+            this.drainTimer = undefined;
+        }
+    }
+
+    private refreshIdleTimer(streamId: number, state: StreamState): void {
+        if (state.idleTimer) clearTimeout(state.idleTimer);
+        state.idleTimer = setTimeout(() => this.onStreamIdle(streamId), STREAM_IDLE_TIMEOUT_MS);
+    }
+
+    private clearIdleTimer(state: StreamState): void {
+        if (state.idleTimer) {
+            clearTimeout(state.idleTimer);
+            state.idleTimer = undefined;
+        }
+    }
+
+    private removeStream(streamId: number): void {
+        const s = this.streams.get(streamId);
+        if (!s) return;
+        this.clearIdleTimer(s);
+        this.streams.delete(streamId);
+        this.pausedReqs.delete(streamId);
+        this.tcpAwaitingDrain.delete(streamId);
+    }
+
+    private onStreamIdle(streamId: number): void {
+        const s = this.streams.get(streamId);
+        if (!s) return;
+        // Tear down the loopback side and tell the agent to release its half.
+        this.teardownStream(s);
+        this.streams.delete(streamId);
+        this.pausedReqs.delete(streamId);
+        this.tcpAwaitingDrain.delete(streamId);
+        if (s.kind === 'tcp') {
+            this.sendJson({ t: 'tcp_close', s: streamId });
+        } else if (s.kind === 'ws') {
+            this.sendJson({ t: 'ws_close', s: streamId, code: 1001, reason: 'idle' });
+        } else {
+            this.sendJson({ t: 'http_err', s: streamId, code: 'timeout', message: 'stream idle timeout' });
+        }
+    }
+
     private sendJson(frame: Parameters<typeof encodeJsonFrame>[0]): void {
         if (this.tunnelWs.readyState !== WebSocket.OPEN) return;
         try { this.tunnelWs.send(encodeJsonFrame(frame)); } catch { /* ignore */ }
diff --git a/backend/src/services/PilotTunnelManager.ts b/backend/src/services/PilotTunnelManager.ts
index 576cbee5..48a2d849 100644
--- a/backend/src/services/PilotTunnelManager.ts
+++ b/backend/src/services/PilotTunnelManager.ts
@@ -3,6 +3,35 @@ import WebSocket from 'ws';
 import { PilotTunnelBridge } from './PilotTunnelBridge';
 import { DatabaseService } from './DatabaseService';
 import { PilotCloseCode } from '../pilot/protocol';
+import { isDebugEnabled } from '../utils/debug';
+import { PilotMetrics } from './PilotMetrics';
+
+/**
+ * Soft warning threshold: a single instance handling more than this many
+ * concurrent pilot tunnels is unusual and likely indicates a reconnect storm
+ * or operator misconfiguration. Logged at WARN.
+ */
+const PILOT_TUNNEL_SOFT_LIMIT = 128;
+
+/**
+ * Hard ceiling on concurrent pilot tunnels per primary. Beyond this the
+ * gateway refuses to register new tunnels (the upgrade handler closes the
+ * socket with 1013 try-again-later) so a runaway reconnect storm cannot
+ * exhaust gateway memory.
+ */
+const PILOT_TUNNEL_HARD_LIMIT = 256;
+
+/**
+ * Thrown by registerTunnel when the system-wide cap is exceeded. The pilot
+ * upgrade handler catches this and closes the WebSocket cleanly so the agent
+ * backs off rather than tight-looping.
+ */
+export class PilotTunnelCapacityError extends Error {
+    constructor(public readonly limit: number) {
+        super(`pilot tunnel cap (${limit}) reached`);
+        this.name = 'PilotTunnelCapacityError';
+    }
+}
 
 /**
  * PilotTunnelManager: singleton registry of active pilot tunnels.
@@ -20,6 +49,7 @@ import { PilotCloseCode } from '../pilot/protocol';
 export class PilotTunnelManager extends EventEmitter {
     private static instance: PilotTunnelManager;
     private bridges: Map<number, PilotTunnelBridge> = new Map();
+    private softWarned = false;
 
     private constructor() {
         super();
@@ -42,11 +72,31 @@ export class PilotTunnelManager extends EventEmitter {
      */
     public async registerTunnel(nodeId: number, ws: WebSocket, agentVersion?: string): Promise<void> {
         const existing = this.bridges.get(nodeId);
+        const replaced = existing != null;
         if (existing) {
             existing.close(PilotCloseCode.Replaced, 'replaced by newer tunnel');
             this.bridges.delete(nodeId);
         }
 
+        // Hard cap: only counts tunnels for *other* nodes since we just
+        // released the matching slot above. A reconnect by the same node
+        // does not consume new capacity.
+        if (this.bridges.size >= PILOT_TUNNEL_HARD_LIMIT) {
+            PilotMetrics.increment('tunnels_rejected_capacity');
+            throw new PilotTunnelCapacityError(PILOT_TUNNEL_HARD_LIMIT);
+        }
+
+        // Bump the replaced counter only after the cap check passes, so a
+        // rejection does not double-count as both a replacement and a
+        // capacity rejection.
+        if (replaced) PilotMetrics.increment('tunnels_replaced');
+        if (this.bridges.size >= PILOT_TUNNEL_SOFT_LIMIT && !this.softWarned) {
+            console.warn(`[Pilot] Active tunnel count at soft limit (${this.bridges.size}/${PILOT_TUNNEL_HARD_LIMIT}); reconnect storm or runaway enrollment likely.`);
+            this.softWarned = true;
+        } else if (this.bridges.size < PILOT_TUNNEL_SOFT_LIMIT) {
+            this.softWarned = false;
+        }
+
         const bridge = new PilotTunnelBridge(nodeId, ws);
         bridge.once('closed', () => {
             if (this.bridges.get(nodeId) === bridge) {
@@ -64,9 +114,35 @@ export class PilotTunnelManager extends EventEmitter {
             pilot_last_seen: Date.now(),
             pilot_agent_version: agentVersion ?? null,
         });
+        PilotMetrics.increment('tunnels_total');
+        if (isDebugEnabled()) {
+            console.log('[PilotMgr:diag] Tunnel registered:', { nodeId, active: this.bridges.size });
+        }
         this.emit('tunnel-up', nodeId);
     }
 
+    /**
+     * Per-tunnel breakdown for the metrics endpoint. Includes the
+     * loopback-relative connectedAt and bufferedAmount so one-bad-node cases
+     * stay visible (an aggregate hides a single tunnel sitting on a stuck
+     * write buffer).
+     */
+    public getMetricsSnapshot(): {
+        counters: ReturnType<typeof PilotMetrics.snapshot>;
+        tunnels_open: number;
+        per_node: Array<{ nodeId: number; connectedAt: number; bufferedAmount: number }>;
+    } {
+        return {
+            counters: PilotMetrics.snapshot(),
+            tunnels_open: this.bridges.size,
+            per_node: Array.from(this.bridges.entries()).map(([nodeId, bridge]) => ({
+                nodeId,
+                connectedAt: bridge.getConnectedAt(),
+                bufferedAmount: bridge.getBufferedAmount(),
+            })),
+        };
+    }
+
     /**
      * Return the loopback base URL (http://127.0.0.1:PORT) for a node's active
      * tunnel, or null if no tunnel is currently registered.
diff --git a/backend/src/websocket/pilotTunnel.ts b/backend/src/websocket/pilotTunnel.ts
index 51420747..08680374 100644
--- a/backend/src/websocket/pilotTunnel.ts
+++ b/backend/src/websocket/pilotTunnel.ts
@@ -4,7 +4,8 @@ import type { WebSocketServer } from 'ws';
 import jwt from 'jsonwebtoken';
 import crypto from 'crypto';
 import { DatabaseService } from '../services/DatabaseService';
-import { PilotTunnelManager } from '../services/PilotTunnelManager';
+import { PilotTunnelCapacityError, PilotTunnelManager } from '../services/PilotTunnelManager';
+import { PilotMetrics } from '../services/PilotMetrics';
 import { encodeJsonFrame as encodePilotJsonFrame, PROTOCOL_VERSION as PILOT_PROTOCOL_VERSION } from '../pilot/protocol';
 import { getErrorMessage } from '../utils/errors';
 import { rejectUpgrade as rejectSocket } from './reject';
@@ -63,6 +64,7 @@ export async function handlePilotTunnel(
       jwtSecret,
       { expiresIn: '365d' },
     );
+    PilotMetrics.increment('enroll_acks');
   }
 
   const agentVersionHeader = req.headers['x-sencho-agent-version'];
@@ -90,6 +92,12 @@ export async function handlePilotTunnel(
     try {
       await PilotTunnelManager.getInstance().registerTunnel(decoded.nodeId!, ws, agentVersion);
     } catch (err) {
+      if (err instanceof PilotTunnelCapacityError) {
+        // 1013 (Try Again Later) signals the agent to back off rather than
+        // tight-loop reconnect on a saturated gateway.
+        try { ws.close(1013, 'pilot tunnel cap reached'); } catch { /* ignore */ }
+        return;
+      }
       console.error('[Pilot] Failed to register tunnel:', getErrorMessage(err, 'unknown'));
       try { ws.close(1011, 'registration failed'); } catch { /* ignore */ }
     }
diff --git a/docs/features/pilot-agent.mdx b/docs/features/pilot-agent.mdx
index 60bdbde2..7964f7eb 100644
--- a/docs/features/pilot-agent.mdx
+++ b/docs/features/pilot-agent.mdx
@@ -71,6 +71,34 @@ The agent reconnects automatically if the tunnel drops (transient network blip,
 
 If the agent container is destroyed before its first successful connect, or the enrollment token expires before you paste it on the remote, open the node's edit dialog on the primary and click **Regenerate enrollment token**. A fresh 15-minute token is issued and the previous tunnel (if any) is disconnected so the new agent replaces it cleanly.
 
+## Self-signed primary TLS certs
+
+If your primary terminates TLS with an internal CA and the agent cannot validate the certificate against the system trust store, point the agent at the CA bundle with `SENCHO_PILOT_CA_FILE`. Mount the bundle into the agent container and add the env var:
+
+```bash
+docker run -d --restart=unless-stopped --name sencho-agent \
+  -v /var/run/docker.sock:/var/run/docker.sock \
+  -v sencho-agent-data:/app/data \
+  -v /opt/docker/sencho:/app/compose \
+  -v /etc/ssl/internal-ca.pem:/etc/ssl/internal-ca.pem:ro \
+  -e SENCHO_MODE=pilot \
+  -e SENCHO_PRIMARY_URL=https://sencho.internal.example.com \
+  -e SENCHO_ENROLL_TOKEN=<short-lived token> \
+  -e SENCHO_PILOT_CA_FILE=/etc/ssl/internal-ca.pem \
+  saelix/sencho:latest
+```
+
+The agent uses the bundle as the only trust anchor for the tunnel WebSocket. TLS verification stays on; there is no env var to disable verification globally because that would defeat the credential trust model.
+
+## Resource limits
+
+Each tunnel has fixed protocol-level ceilings to keep one misbehaving agent from impacting the primary:
+
+- **Frame size:** individual WebSocket frames are capped at 8 MB. Single requests and responses larger than that are rejected and the tunnel reconnects.
+- **Concurrent streams:** at most 1024 multiplexed HTTP, WebSocket, or Mesh-TCP streams per tunnel. Above the cap the bridge returns 503 and the agent rejects new incoming streams with a typed error frame.
+- **Stream idle:** any stream with no activity for 10 minutes is closed and removed.
+- **System-wide tunnels:** a single primary accepts up to 256 concurrent pilot tunnels. Past the soft warning at 128 the primary logs a WARN; past the hard cap of 256 new tunnels are refused with WebSocket close code 1013 (Try Again Later) so the agent backs off rather than tight-looping.
+
 ## Troubleshooting
 
 <AccordionGroup>
@@ -93,4 +121,20 @@ If the agent container is destroyed before its first successful connect, or the
   <Accordion title="Primary was restored from backup and the agent will not reconnect">
     The agent's persisted tunnel credential is signed with the primary's JWT secret. If that secret is rotated or the primary is rebuilt from scratch, existing tunnels stop verifying. Regenerate enrollment for the affected node and restart the agent container so it consumes the fresh token.
   </Accordion>
+
+  <Accordion title="Tunnel rejected with 'pilot tunnel cap reached' (close code 1013)">
+    The primary is at the system-wide concurrent tunnel cap of 256. A reconnect storm or runaway enrollment is the usual cause. Inspect the primary's logs for the `[Pilot] Active tunnel count at soft limit` warning that fires at 128 to find the trend, and check `GET /api/system/pilot-tunnels` (admin-only) for the per-node breakdown to identify a flapping agent. The agent backs off and retries automatically once headroom returns.
+  </Accordion>
+
+  <Accordion title="Bridge returns 503 'pilot tunnel: stream cap reached'">
+    A single tunnel is at the 1024 concurrent stream cap. The most common trigger is a UI session that opened many long-poll streams and never closed them, or a script firing parallel API calls without bound. Reload the affected browser tab so its WebSocket and SSE streams reset. If the cap is being hit by automation, throttle the caller; the cap protects the primary's memory from a runaway agent or proxy.
+  </Accordion>
+
+  <Accordion title="Tunnel closes immediately with 'protocol error' (close code 1002)">
+    The agent sent a frame larger than the 8 MB ceiling, or the wire decoder rejected a malformed frame. Run the agent with developer mode enabled on the primary (Settings → Developer) to surface the diagnostic decode log, then check the primary's logs for `[PilotBridge:diag] Malformed frame from agent`. The agent reconnects automatically; persistent failures usually mean a version mismatch between primary and agent images.
+  </Accordion>
+
+  <Accordion title="HTTP 429 'Too many enrollment requests'">
+    Enrollment endpoints are rate-limited to 10 requests per minute per user (or per IP, when unauthenticated) so a script accidentally minting tokens in a loop cannot exhaust the database. Wait sixty seconds and retry. If you genuinely need to bulk-enroll many nodes, space the requests out or contact the operator who runs the primary about raising the limit.
+  </Accordion>
 </AccordionGroup>