refactor(pilot): narrow Mesh handle, prune dead code, add replay-route test (#982)

* refactor(pilot): narrow PilotTunnelManager.getBridge to MeshTunnelHandle

The manager handed out the entire PilotTunnelBridge to its only consumer,
MeshService, which let any current or future caller reach into transport
internals: loopback URL, per-stream maps, the close API, the underscored
_writeTcpData / _closeTcpStream, the diagnostic helpers. None of that
was load-bearing for Mesh.

Introduce a MeshTunnelHandle interface alongside TcpStream in
PilotTunnelBridge.ts that exposes only the two methods Mesh actually
calls (openTcpStream and getBufferedAmount), have PilotTunnelBridge
declare implements MeshTunnelHandle, and change getBridge to return
the interface. MeshService continues to compile unchanged because its
existing call sites only touch the narrowed surface.

A future alternative transport (a stub for tests, a different routing
strategy) now has a one-method-and-a-getter contract to satisfy
instead of the full bridge.

* refactor(pilot): drop unused tunnel manager and bridge surface

Three public methods predating the hardening pass had zero callers
across the entire codebase (verified via grep across backend, frontend,
e2e):

  - PilotTunnelManager.touch(nodeId): never invoked. The pilot_last_seen
    timestamp is updated by the manager itself on registerTunnel and by
    the persistence layer on heartbeat events.
  - PilotTunnelManager.listActive(): never invoked. The metrics endpoint
    added in PR #979 returns its own per-node breakdown via
    getMetricsSnapshot, which is the canonical observability surface.
  - PilotTunnelBridge.listTcpStreams() and the supporting
    TcpStreamSummary interface: never invoked. The Mesh diagnostics
    sheet that would have consumed it is not wired and would use
    getMetricsSnapshot if/when it ships.

Removing them tightens the public surface and prevents accidental new
dependencies on speculative future-proofing.

* test(pilot): cover enrollment replay rejection at the route layer

The DB-level test in pilot-enrollment.test.ts already verifies that
consumePilotEnrollment is one-shot. That guards the persistence layer
but not the route handler: a refactor of handlePilotTunnel that swaps
the order of consume vs upgrade, or that grants the WebSocket
upgrade before checking the consume result, would silently break the
security invariant while DB-layer tests stay green.

Drive handlePilotTunnel directly with a stub IncomingMessage and
Duplex socket. Six cases:
  - Already-consumed enrollment row -> 401.
  - Token whose hash matches no row -> 401.
  - Row whose expires_at has passed -> 401.
  - Missing Authorization header -> 401.
  - JWT signed with a wrong secret -> 401.
  - pilot_tunnel JWT for an unknown node -> 404.

The stub captures HTTP/1.1 status writes so the test asserts the
exact rejection lands on the wire, not just that the function
returned without throwing.

* docs(debug): warn future committers off per-frame isDebugEnabled calls

Code-review feedback on PR #979 noted that isDebugEnabled is fine in
the cadences it has today (per-tunnel, per-request, error paths) but
is fragile against a future commit that drops it into a per-frame
WebSocket loop. The function does a try/catch + Node require cache
lookup + a method call into DatabaseService on every invocation;
acceptable at hundreds of calls per second, expensive at thousands.

Add a comment block above the function spelling out the acceptable
and unacceptable cadences and the snapshot-outside-the-loop
mitigation, so the next person to add a diag log there sees the
constraint.

* fix(pilot): address PR A code-review findings

Code review on this branch surfaced four items:

  - Em-dash directive (CLAUDE.md Directive 18) violations in four
    comment sites; replaced with colons or restructured.
  - debug.ts perf comment claimed try/catch frame setup as the cost
    driver. V8 inlines those; the load-bearing cost is the require
    lookup and singleton dispatch. Reword.
  - Test file header described coverage as 'replay rejection at the
    route layer' but the file now also covers missing-header,
    wrong-secret, expired-row, never-stored, and unknown-node
    rejection paths. Widen the JSDoc and the describe label to match.
  - Stub-cast comment in the replay test now explicitly lists the
    IncomingMessage and Duplex surface the stub satisfies, so a
    future commit that grows handlePilotTunnel's surface (rate
    limiting, socket options) updates both stubs instead of
    silently no-opping against them.

No behavior change.

backend/src/__tests__/pilot-enrollment-replay.test.ts

@@ -0,0 +1,231 @@
+/**
+ * Route-level coverage for the rejection paths inside handlePilotTunnel.
+ *
+ * The headline invariant is "a pilot_enroll JWT is one-shot at the upgrade
+ * handler, not just at the DB layer". pilot-enrollment.test.ts already
+ * verifies the DB layer. That is necessary but not sufficient: a future
+ * refactor of handlePilotTunnel that re-orders mint and consume, or that
+ * grants the WebSocket upgrade before checking the consume result, would
+ * silently break the invariant while the DB-layer test stays green.
+ *
+ * This file drives handlePilotTunnel directly with a stub IncomingMessage
+ * and Duplex socket. Each test exercises a different early-return branch
+ * (missing header, wrong-secret JWT, never-stored row, expired row,
+ * already-consumed row, unknown node) and asserts the rejection lands on
+ * the wire with the right HTTP status before any upgrade attempt.
+ */
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { EventEmitter } from 'events';
+import type { IncomingMessage } from 'http';
+import type { Duplex } from 'stream';
+import crypto from 'crypto';
+import jwt from 'jsonwebtoken';
+import { WebSocketServer } from 'ws';
+import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
+import { handlePilotTunnel } from '../websocket/pilotTunnel';
+
+let tmpDir: string;
+let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
+let pilotTunnelWss: WebSocketServer;
+
+beforeAll(async () => {
+    tmpDir = await setupTestDb();
+    ({ DatabaseService } = await import('../services/DatabaseService'));
+    pilotTunnelWss = new WebSocketServer({ noServer: true });
+});
+
+afterAll(() => {
+    pilotTunnelWss.close();
+    cleanupTestDb(tmpDir);
+});
+
+interface StubSocket extends EventEmitter {
+    writes: string[];
+    destroyed: boolean;
+    write(chunk: string): boolean;
+    destroy(): void;
+}
+
+function makeStubSocket(): StubSocket {
+    const sock = new EventEmitter() as StubSocket;
+    sock.writes = [];
+    sock.destroyed = false;
+    sock.write = (chunk: string) => { sock.writes.push(chunk); return true; };
+    sock.destroy = () => { sock.destroyed = true; };
+    return sock;
+}
+
+function makeStubReq(authHeader: string | undefined, agentVersion = 'test-1.0'): IncomingMessage {
+    const headers: Record<string, string> = { 'x-sencho-agent-version': agentVersion };
+    if (authHeader !== undefined) headers['authorization'] = authHeader;
+    // Cast through unknown: handlePilotTunnel only reads `headers` from
+    // req, and on the reject path only calls socket.write(string) and
+    // socket.destroy() on the Duplex. Anything else we add to the handler
+    // (e.g. reading req.socket.remoteAddress for rate-limiting) silently
+    // no-ops against the stub, so update both stubs when the surface grows.
+    return { headers } as unknown as IncomingMessage;
+}
+
+/**
+ * Mint a real pilot_enroll JWT against the test DB's auth secret. Mirrors
+ * mintPilotEnrollment in routes/nodes.ts but inlined to avoid pulling the
+ * full Express request shape into the test.
+ */
+function mintEnroll(nodeId: number, ttlSeconds = 15 * 60): { token: string; hash: string } {
+    const db = DatabaseService.getInstance();
+    const jwtSecret = db.getGlobalSettings().auth_jwt_secret;
+    if (!jwtSecret) throw new Error('test DB has no auth_jwt_secret');
+    const token = jwt.sign(
+        { scope: 'pilot_enroll', nodeId, enrollNonce: crypto.randomUUID() },
+        jwtSecret,
+        { expiresIn: ttlSeconds },
+    );
+    const hash = crypto.createHash('sha256').update(token).digest('hex');
+    db.createPilotEnrollment(nodeId, hash, Date.now() + ttlSeconds * 1000);
+    return { token, hash };
+}
+
+let nodeId: number;
+
+beforeEach(() => {
+    // Fresh pilot-mode node per test so prior enrollments do not bleed.
+    nodeId = DatabaseService.getInstance().addNode({
+        name: `pilot-replay-${Date.now()}-${Math.random()}`,
+        type: 'remote',
+        mode: 'pilot_agent',
+        compose_dir: '/tmp/x',
+        is_default: false,
+        api_url: '',
+        api_token: '',
+    });
+});
+
+describe('handlePilotTunnel: rejection paths in the upgrade handler', () => {
+    it('rejects an enrollment token whose row was already consumed', async () => {
+        const { token, hash } = mintEnroll(nodeId);
+
+        // Simulate that a prior successful enrollment consumed the row.
+        const firstConsume = DatabaseService.getInstance().consumePilotEnrollment(hash);
+        expect(firstConsume).toBeDefined();
+        expect(firstConsume?.node_id).toBe(nodeId);
+
+        // Now drive the upgrade handler with the same token. The route
+        // must reject with 401 before granting the WebSocket upgrade.
+        const socket = makeStubSocket();
+        await handlePilotTunnel(
+            makeStubReq(`Bearer ${token}`),
+            socket as unknown as Duplex,
+            Buffer.alloc(0),
+            pilotTunnelWss,
+        );
+
+        expect(socket.destroyed).toBe(true);
+        expect(socket.writes.length).toBeGreaterThan(0);
+        expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/);
+    });
+
+    it('rejects an enrollment token whose row was never created', async () => {
+        // Hand-rolled token whose hash matches no row in pilot_enrollments.
+        const db = DatabaseService.getInstance();
+        const jwtSecret = db.getGlobalSettings().auth_jwt_secret;
+        const orphan = jwt.sign(
+            { scope: 'pilot_enroll', nodeId, enrollNonce: 'never-stored' },
+            jwtSecret,
+            { expiresIn: 60 },
+        );
+
+        const socket = makeStubSocket();
+        await handlePilotTunnel(
+            makeStubReq(`Bearer ${orphan}`),
+            socket as unknown as Duplex,
+            Buffer.alloc(0),
+            pilotTunnelWss,
+        );
+
+        expect(socket.destroyed).toBe(true);
+        expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/);
+    });
+
+    it('rejects an enrollment token whose row has expired', async () => {
+        const db = DatabaseService.getInstance();
+        const jwtSecret = db.getGlobalSettings().auth_jwt_secret;
+        // JWT TTL still valid (60 s) but the DB row's expires_at is in the
+        // past. consumePilotEnrollment filters on expires_at > now, so this
+        // tests the DB-side window, not the JWT-side one.
+        const token = jwt.sign(
+            { scope: 'pilot_enroll', nodeId, enrollNonce: 'expired-row' },
+            jwtSecret,
+            { expiresIn: 60 },
+        );
+        const hash = crypto.createHash('sha256').update(token).digest('hex');
+        db.createPilotEnrollment(nodeId, hash, Date.now() - 1_000);
+
+        const socket = makeStubSocket();
+        await handlePilotTunnel(
+            makeStubReq(`Bearer ${token}`),
+            socket as unknown as Duplex,
+            Buffer.alloc(0),
+            pilotTunnelWss,
+        );
+
+        expect(socket.destroyed).toBe(true);
+        expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/);
+    });
+
+    it('rejects a missing Authorization header before touching the DB', async () => {
+        const socket = makeStubSocket();
+        await handlePilotTunnel(
+            makeStubReq(undefined),
+            socket as unknown as Duplex,
+            Buffer.alloc(0),
+            pilotTunnelWss,
+        );
+
+        expect(socket.destroyed).toBe(true);
+        expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/);
+    });
+
+    it('rejects a JWT signed with a wrong secret', async () => {
+        const wrongToken = jwt.sign(
+            { scope: 'pilot_enroll', nodeId, enrollNonce: 'wrong-secret' },
+            'this-is-not-the-real-secret',
+            { expiresIn: 60 },
+        );
+
+        const socket = makeStubSocket();
+        await handlePilotTunnel(
+            makeStubReq(`Bearer ${wrongToken}`),
+            socket as unknown as Duplex,
+            Buffer.alloc(0),
+            pilotTunnelWss,
+        );
+
+        expect(socket.destroyed).toBe(true);
+        expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/);
+    });
+
+    it('rejects a pilot_tunnel-scoped JWT for an unknown node', async () => {
+        // pilot_tunnel scope is the long-lived credential persisted on the
+        // agent. A pilot_tunnel JWT for a node that has been deleted (or
+        // never existed) must be rejected at the upgrade handler: the
+        // node lookup is the gate, not the JWT signature alone.
+        const db = DatabaseService.getInstance();
+        const jwtSecret = db.getGlobalSettings().auth_jwt_secret;
+        const ghost = jwt.sign(
+            { scope: 'pilot_tunnel', nodeId: 99_999_999 },
+            jwtSecret,
+            { expiresIn: '365d' },
+        );
+
+        const socket = makeStubSocket();
+        await handlePilotTunnel(
+            makeStubReq(`Bearer ${ghost}`),
+            socket as unknown as Duplex,
+            Buffer.alloc(0),
+            pilotTunnelWss,
+        );
+
+        expect(socket.destroyed).toBe(true);
+        expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 404 /);
+    });
+});

backend/src/services/PilotTunnelBridge.ts

@@ -59,13 +59,6 @@ interface TcpStreamState extends StreamMeta {
 
 type StreamState = HttpStreamState | WsStreamState | TcpStreamState;
 
-export interface TcpStreamSummary {
-    streamId: number;
-    bytesIn: number;
-    bytesOut: number;
-    openedAt: number;
-}
-
 /**
  * Sencho Mesh TCP stream handle. EventEmitter-based duplex-like surface that
  * MeshService consumes to bridge a local socket to a Compose service on the
@@ -105,6 +98,19 @@ export class TcpStream extends EventEmitter {
     }
 }
 
+/**
+ * Narrow surface that Sencho Mesh consumes from a registered pilot tunnel.
+ * `PilotTunnelManager.getBridge` returns this interface (not the concrete
+ * bridge) so MeshService cannot reach into transport internals: close code,
+ * loopback URL, the per-stream maps, the underscored helpers, etc.
+ * Keep this set minimal: it is the contract a future alternative transport
+ * (a stub for tests, a different routing strategy) would have to implement.
+ */
+export interface MeshTunnelHandle {
+    openTcpStream(target: { stack: string; service: string; port: number }): TcpStream | null;
+    getBufferedAmount(): number;
+}
+
 /**
  * Per-tunnel bridge: hosts a loopback HTTP server that demuxes requests into
  * wire frames sent over the pilot WebSocket, and remuxes response frames back
@@ -114,7 +120,7 @@ export class TcpStream extends EventEmitter {
  * as just another remote target, so HTTP and WebSocket proxy logic, header
  * stripping/injection, and license-tier propagation all work unchanged.
  */
-export class PilotTunnelBridge extends EventEmitter {
+export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle {
     private readonly tunnelWs: WebSocket;
     private readonly loopback: HttpServer;
     private readonly wsUpgradeServer: WebSocketServer;
@@ -231,20 +237,6 @@ export class PilotTunnelBridge extends EventEmitter {
         this.sendJson({ t: 'tcp_close', s: streamId });
     }
 
-    /**
-     * Snapshot of active TCP streams for the diagnostics sheet. Cheap; called
-     * on demand by MeshService.
-     */
-    public listTcpStreams(): TcpStreamSummary[] {
-        const out: TcpStreamSummary[] = [];
-        for (const [streamId, s] of this.streams) {
-            if (s.kind === 'tcp') {
-                out.push({ streamId, bytesIn: s.bytesIn, bytesOut: s.bytesOut, openedAt: s.openedAt });
-            }
-        }
-        return out;
-    }
-
     public close(code = 1000, reason = 'closed by primary'): void {
         if (this.closed) return;
         this.closed = true;

backend/src/services/PilotTunnelManager.ts

@@ -1,6 +1,6 @@
 import { EventEmitter } from 'events';
 import WebSocket from 'ws';
-import { PilotTunnelBridge } from './PilotTunnelBridge';
+import { PilotTunnelBridge, type MeshTunnelHandle } from './PilotTunnelBridge';
 import { DatabaseService } from './DatabaseService';
 import { PilotCloseCode } from '../pilot/protocol';
 import { isDebugEnabled } from '../utils/debug';
@@ -160,11 +160,11 @@ export class PilotTunnelManager extends EventEmitter {
     }
 
     /**
-     * Direct accessor for the per-node bridge. Returns null when no tunnel is
-     * registered. Used by Sencho Mesh to open TCP streams without going
-     * through the loopback HTTP path.
+     * Per-node tunnel handle, returned to MeshService. Returns the bridge
+     * narrowed to the MeshTunnelHandle surface so callers cannot reach
+     * into transport internals (loopback URL, per-stream maps, close API).
      */
-    public getBridge(nodeId: number): PilotTunnelBridge | null {
+    public getBridge(nodeId: number): MeshTunnelHandle | null {
         return this.bridges.get(nodeId) ?? null;
     }
 
@@ -178,22 +178,4 @@ export class PilotTunnelManager extends EventEmitter {
         this.bridges.delete(nodeId);
     }
 
-    /**
-     * Snapshot of currently active tunnels.
-     */
-    public listActive(): Array<{ nodeId: number; loopbackUrl: string; connectedAt: number }> {
-        return Array.from(this.bridges.entries()).map(([nodeId, bridge]) => ({
-            nodeId,
-            loopbackUrl: bridge.getLoopbackUrl(),
-            connectedAt: bridge.getConnectedAt(),
-        }));
-    }
-
-    /**
-     * Record an application-level heartbeat from the agent.
-     */
-    public touch(nodeId: number): void {
-        if (!this.bridges.has(nodeId)) return;
-        DatabaseService.getInstance().updateNode(nodeId, { pilot_last_seen: Date.now() });
-    }
 }

backend/src/utils/debug.ts

@@ -2,6 +2,18 @@
  * Shared diagnostic logging gate. Reads `developer_mode` from
  * DatabaseService, which caches the global_settings snapshot internally
  * and invalidates on write, so hot-path callers can query freely.
+ *
+ * Acceptable: error paths, per-request paths, per-tunnel paths,
+ * once-per-stream paths. Each call costs one Map lookup in Node's
+ * require cache, one method call into the cached settings object, and
+ * a property access. Fine for these cadences.
+ *
+ * NOT acceptable: per-frame paths in steady state (WebSocket message
+ * loops, container stats streams, log-tail demuxers). The require
+ * lookup and singleton dispatch add up at thousands of calls per
+ * second. If you need diagnostic logging in a per-frame loop,
+ * snapshot the result once outside the loop and read the snapshot
+ * inside.
  */
 
 export function isDebugEnabled(): boolean {