fix(pilot): preserve tunnel JWT across upgrade 401 reconnects (#1818)

Stop deleting pilot.jwt on any WebSocket 401 and falling back to an already-consumed enroll token. Attempt enroll recovery only on upgrade 401/404 when SENCHO_ENROLL_TOKEN is still fresh, restore the in-memory tunnel credential if that attempt fails, and surface a clear re-enrollment prompt instead of looping silently.

Closes #1817

backend/src/__tests__/pilot-agent-auth-fallback-unit.test.ts

@@ -2,6 +2,7 @@
  * Unit tests for PilotAgent auth fallback event handling using a stub WebSocket.
  */
 import { describe, it, expect, vi, beforeEach, afterEach, beforeAll, afterAll } from 'vitest';
+import jwt from 'jsonwebtoken';
 import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
 
 const { wsInstances, mockAttachSwitchboard } = vi.hoisted(() => ({
@@ -9,6 +10,8 @@ const { wsInstances, mockAttachSwitchboard } = vi.hoisted(() => ({
         emit: (event: string, ...args: unknown[]) => boolean;
         readyState: number;
         close: ReturnType<typeof vi.fn>;
+        terminate: ReturnType<typeof vi.fn>;
+        on: (event: string, listener: (...args: unknown[]) => void) => unknown;
     }>,
     mockAttachSwitchboard: vi.fn(() => ({
         handleJsonFrame: vi.fn(() => false),
@@ -30,9 +33,10 @@ vi.mock('ws', () => {
     class MockWebSocket extends EventEmitter {
         readyState = 0;
         close = vi.fn();
+        terminate = vi.fn();
         constructor(..._args: unknown[]) {
             super();
-            wsInstances.push(this);
+            wsInstances.push(this as never);
         }
     }
     return { default: MockWebSocket };
@@ -43,87 +47,294 @@ let PilotAgent: typeof import('../pilot/agent').PilotAgent;
 let readPersistedToken: typeof import('../pilot/agent').readPersistedToken;
 let persistToken: typeof import('../pilot/agent').persistToken;
 let clearPersistedToken: typeof import('../pilot/agent').clearPersistedToken;
+let isFreshEnrollToken: typeof import('../pilot/agent').isFreshEnrollToken;
 
 beforeAll(async () => {
     tmpDir = await setupTestDb();
-    ({ PilotAgent, readPersistedToken, persistToken, clearPersistedToken } = await import('../pilot/agent'));
+    ({
+        PilotAgent,
+        readPersistedToken,
+        persistToken,
+        clearPersistedToken,
+        isFreshEnrollToken,
+    } = await import('../pilot/agent'));
 });
 
 afterAll(() => {
     cleanupTestDb(tmpDir);
 });
 
+function mintFreshEnroll(): string {
+    return jwt.sign({ scope: 'pilot_enroll', nodeId: 1 }, 'unit-test-secret', { expiresIn: '15m' });
+}
+
+function mintExpiredEnroll(): string {
+    // Explicit exp: negative expiresIn is unreliable in jsonwebtoken.
+    return jwt.sign(
+        { scope: 'pilot_enroll', nodeId: 1, exp: Math.floor(Date.now() / 1000) - 120 },
+        'unit-test-secret',
+    );
+}
+
+function emitUpgradeReject(
+    ws: (typeof wsInstances)[number],
+    status: number,
+    reason?: string,
+): void {
+    const headers: Record<string, string> = {};
+    if (reason) headers['x-sencho-pilot-reject'] = reason;
+    ws.emit('unexpected-response', {}, {
+        statusCode: status,
+        headers,
+        resume: vi.fn(),
+    });
+}
+
+describe('isFreshEnrollToken', () => {
+    it('accepts an unexpired pilot_enroll JWT', () => {
+        expect(isFreshEnrollToken(mintFreshEnroll())).toBe(true);
+    });
+
+    it('rejects an expired pilot_enroll JWT', () => {
+        expect(isFreshEnrollToken(mintExpiredEnroll())).toBe(false);
+    });
+
+    it('rejects a non-JWT string', () => {
+        expect(isFreshEnrollToken('not-a-jwt')).toBe(false);
+    });
+});
+
 describe('PilotAgent auth fallback (stub WebSocket)', () => {
     beforeEach(() => {
         wsInstances.length = 0;
         vi.clearAllMocks();
         vi.spyOn(console, 'log').mockImplementation(() => { /* swallow */ });
         vi.spyOn(console, 'warn').mockImplementation(() => { /* swallow */ });
+        vi.useFakeTimers();
     });
 
     afterEach(() => {
+        vi.runOnlyPendingTimers();
+        vi.useRealTimers();
         vi.restoreAllMocks();
         clearPersistedToken();
     });
 
-    it('swaps to the enroll token and clears pilot.jwt after HTTP 401 on upgrade', () => {
+    it('swaps to a fresh enroll token after HTTP 401 and leaves pilot.jwt on disk', () => {
         persistToken('stale-on-disk');
+        const enroll = mintFreshEnroll();
         const agent = new PilotAgent({
             primaryUrl: 'http://primary.invalid',
             loopbackPort: 1,
             initialToken: 'stale-token',
-            enrollToken: 'fresh-enroll-token',
+            enrollToken: enroll,
             enrolling: false,
         });
 
         (agent as unknown as { connect: () => void }).connect();
 
         const firstWs = wsInstances[0]!;
-        // Real ws (no unexpected-response listener): abortHandshake emits error then close.
-        firstWs.emit('error', new Error('Unexpected server response: 401'));
-        firstWs.emit('close', 1006, Buffer.from(''));
+        emitUpgradeReject(firstWs, 401, 'invalid_token');
 
-        expect(readPersistedToken()).toBeNull();
-        expect((agent as unknown as { token: string }).token).toBe('fresh-enroll-token');
+        expect(readPersistedToken()).toBe('stale-on-disk');
+        expect((agent as unknown as { token: string }).token).toBe(enroll);
         // scheduleReconnect doubles backoff after scheduling the imminent retry.
         expect((agent as unknown as { backoff: number }).backoff).toBe(2_000);
     });
 
     it('swaps to the enroll token after HTTP 404 on upgrade', () => {
         persistToken('stale-on-disk');
+        const enroll = mintFreshEnroll();
         const agent = new PilotAgent({
             primaryUrl: 'http://primary.invalid',
             loopbackPort: 1,
             initialToken: 'stale-token',
-            enrollToken: 'fresh-enroll-token',
+            enrollToken: enroll,
             enrolling: false,
         });
 
         (agent as unknown as { connect: () => void }).connect();
+        emitUpgradeReject(wsInstances[0]!, 404, 'unknown_node');
 
-        const firstWs = wsInstances[0]!;
-        firstWs.emit('error', new Error('Unexpected server response: 404'));
-        firstWs.emit('close', 1006, Buffer.from(''));
-
-        expect(readPersistedToken()).toBeNull();
-        expect((agent as unknown as { token: string }).token).toBe('fresh-enroll-token');
+        expect(readPersistedToken()).toBe('stale-on-disk');
+        expect((agent as unknown as { token: string }).token).toBe(enroll);
     });
 
-    it('does not swap when already connecting with the enroll token', () => {
+    it('does not swap when the enroll token is expired', () => {
+        persistToken('stale-on-disk');
         const agent = new PilotAgent({
             primaryUrl: 'http://primary.invalid',
             loopbackPort: 1,
-            initialToken: 'only-enroll-token',
-            enrollToken: 'only-enroll-token',
+            initialToken: 'stale-token',
+            enrollToken: mintExpiredEnroll(),
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        emitUpgradeReject(wsInstances[0]!, 401, 'invalid_token');
+
+        expect(readPersistedToken()).toBe('stale-on-disk');
+        expect((agent as unknown as { token: string }).token).toBe('stale-token');
+    });
+
+    it('does not swap on a clean close even with a fresh enroll token', () => {
+        persistToken('good-on-disk');
+        const enroll = mintFreshEnroll();
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'good-token',
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        wsInstances[0]!.emit('close', 1000, Buffer.from(''));
+
+        expect(readPersistedToken()).toBe('good-on-disk');
+        expect((agent as unknown as { token: string }).token).toBe('good-token');
+    });
+
+    it('does not swap when already connecting with the enroll token', () => {
+        const enroll = mintFreshEnroll();
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: enroll,
+            enrollToken: enroll,
             enrolling: true,
         });
 
         (agent as unknown as { connect: () => void }).connect();
+        emitUpgradeReject(wsInstances[0]!, 401, 'enrollment_used');
 
+        // Failed enroll dial with no prior tunnelToken leaves token as enroll.
+        expect((agent as unknown as { token: string }).token).toBe(enroll);
+    });
+
+    it('restores the in-memory tunnel token when enroll fallback also 401s', () => {
+        const enroll = mintFreshEnroll();
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'tunnel-in-memory',
+            enrollToken: enroll,
+            enrolling: false,
+        });
+        // Simulate persist failure: no file on disk.
+        clearPersistedToken();
+
+        (agent as unknown as { connect: () => void }).connect();
+        emitUpgradeReject(wsInstances[0]!, 401);
+        expect((agent as unknown as { token: string }).token).toBe(enroll);
+
+        // Advance into the reconnect attempt with the enroll token.
+        vi.runOnlyPendingTimers();
+        expect(wsInstances.length).toBe(2);
+        emitUpgradeReject(wsInstances[1]!, 401, 'enrollment_used');
+
+        expect((agent as unknown as { token: string }).token).toBe('tunnel-in-memory');
+        expect(readPersistedToken()).toBeNull();
+    });
+
+    it('restores the tunnel token when the enroll dial drops without a 401/404', () => {
+        const enroll = mintFreshEnroll();
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'tunnel-in-memory',
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        emitUpgradeReject(wsInstances[0]!, 401);
+        expect((agent as unknown as { token: string }).token).toBe(enroll);
+
+        vi.runOnlyPendingTimers();
+        // Clean close / network drop while dialing enroll (no rejectInfo).
+        wsInstances[1]!.emit('close', 1006, Buffer.from(''));
+
+        expect((agent as unknown as { token: string }).token).toBe('tunnel-in-memory');
+    });
+
+    it('still swaps when the reject reason is enrollment_used (header is diagnostic only)', () => {
+        persistToken('stale-on-disk');
+        const enroll = mintFreshEnroll();
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'stale-token',
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        emitUpgradeReject(wsInstances[0]!, 401, 'enrollment_used');
+
+        expect(readPersistedToken()).toBe('stale-on-disk');
+        expect((agent as unknown as { token: string }).token).toBe(enroll);
+    });
+
+    it('does not swap on hub 403', () => {
+        persistToken('good-on-disk');
+        const enroll = mintFreshEnroll();
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'good-token',
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        emitUpgradeReject(wsInstances[0]!, 403, 'bad_scope');
+
+        expect(readPersistedToken()).toBe('good-on-disk');
+        expect((agent as unknown as { token: string }).token).toBe('good-token');
+    });
+
+    it('does not double-schedule reconnect when a stale socket closes after the next connect', () => {
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'tunnel-token',
+            enrollToken: mintFreshEnroll(),
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
         const firstWs = wsInstances[0]!;
-        firstWs.emit('error', new Error('Unexpected server response: 401'));
-        firstWs.emit('close', 1006, Buffer.from(''));
+        emitUpgradeReject(firstWs, 401);
+        const afterFirst = wsInstances.length;
 
-        expect((agent as unknown as { token: string }).token).toBe('only-enroll-token');
+        vi.runOnlyPendingTimers();
+        expect(wsInstances.length).toBe(afterFirst + 1);
+
+        // Late close from the first socket must be a no-op for reconnect.
+        firstWs.emit('close', 1006, Buffer.from(''));
+        const beforeAdvance = wsInstances.length;
+        vi.runOnlyPendingTimers();
+        expect(wsInstances.length).toBe(beforeAdvance);
+    });
+
+    it('keeps the tunnel token on opaque 401 when enroll is fresh (fallback still swaps in memory)', () => {
+        persistToken('stale-on-disk');
+        const enroll = mintFreshEnroll();
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'stale-token',
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        // No X-Sencho-Pilot-Reject header (proxy / older hub).
+        emitUpgradeReject(wsInstances[0]!, 401);
+
+        expect(readPersistedToken()).toBe('stale-on-disk');
+        expect((agent as unknown as { token: string }).token).toBe(enroll);
+        expect((agent as unknown as { backoff: number }).backoff).toBe(2_000);
     });
 });

backend/src/__tests__/pilot-agent-auth-fallback.test.ts

@@ -2,12 +2,12 @@
  * Regression tests for pilot agent auth fallback when a persisted tunnel
  * token is rejected at WebSocket upgrade (401 invalid JWT, 404 unknown node).
  *
- * Without fallback the agent reconnects forever with the same stale
- * pilot.jwt credential even when SENCHO_ENROLL_TOKEN carries a fresh
- * enrollment JWT.
+ * Preserves pilot.jwt until a successful enroll_ack; enroll fallback requires
+ * upgrade 401/404 plus a still-fresh SENCHO_ENROLL_TOKEN.
  */
-import { describe, it, expect, beforeAll, afterAll, afterEach } from 'vitest';
+import { describe, it, expect, beforeAll, afterAll, afterEach, vi } from 'vitest';
 import http from 'http';
+import crypto from 'crypto';
 import jwt from 'jsonwebtoken';
 import { WebSocketServer } from 'ws';
 import { setupTestDb, cleanupTestDb, TEST_JWT_SECRET } from './helpers/setupTestDb';
@@ -18,11 +18,10 @@ import { WebSocket } from 'ws';
 let tmpDir: string;
 let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
 
-// agent.ts freezes its pilot.jwt path from DATA_DIR at module load, so it must
-// be imported only after setupTestDb() points DATA_DIR at the writable tmp dir.
 let readPersistedToken: typeof import('../pilot/agent').readPersistedToken;
 let persistToken: typeof import('../pilot/agent').persistToken;
 let clearPersistedToken: typeof import('../pilot/agent').clearPersistedToken;
+let PilotAgent: typeof import('../pilot/agent').PilotAgent;
 
 let server: http.Server;
 let port: number;
@@ -33,7 +32,7 @@ let nodeId: number;
 beforeAll(async () => {
     tmpDir = await setupTestDb();
     ({ DatabaseService } = await import('../services/DatabaseService'));
-    ({ readPersistedToken, persistToken, clearPersistedToken } = await import('../pilot/agent'));
+    ({ readPersistedToken, persistToken, clearPersistedToken, PilotAgent } = await import('../pilot/agent'));
 
     server = http.createServer();
     mainWss = new WebSocketServer({ noServer: true });
@@ -78,6 +77,7 @@ afterAll(async () => {
 afterEach(() => {
     PilotTunnelManager.getInstance().closeTunnel(nodeId);
     clearPersistedToken();
+    vi.useRealTimers();
 });
 
 function mintStaleTunnelTokenWrongSecret(): string {
@@ -96,6 +96,68 @@ function mintStaleTunnelTokenWrongNode(): string {
     );
 }
 
+function mintFreshEnrollForNode(): string {
+    const expiresAt = Date.now() + 15 * 60 * 1000;
+    const token = jwt.sign(
+        { scope: 'pilot_enroll', nodeId, enrollNonce: crypto.randomUUID() },
+        TEST_JWT_SECRET,
+        { expiresIn: '15m' },
+    );
+    const tokenHash = crypto.createHash('sha256').update(token).digest('hex');
+    DatabaseService.getInstance().createPilotEnrollment(nodeId, tokenHash, expiresAt);
+    return token;
+}
+
+function mintExpiredEnroll(): string {
+    return jwt.sign(
+        {
+            scope: 'pilot_enroll',
+            nodeId,
+            enrollNonce: crypto.randomUUID(),
+            exp: Math.floor(Date.now() / 1000) - 120,
+        },
+        TEST_JWT_SECRET,
+    );
+}
+
+function stopAgent(agent: InstanceType<typeof PilotAgent>): void {
+    (agent as unknown as { shuttingDown: boolean }).shuttingDown = true;
+    const timer = (agent as unknown as { reconnectTimer?: NodeJS.Timeout }).reconnectTimer;
+    if (timer) clearTimeout(timer);
+    (agent as unknown as { reconnectTimer?: NodeJS.Timeout }).reconnectTimer = undefined;
+    try { (agent as unknown as { ws: WebSocket | null }).ws?.terminate(); } catch { /* ignore */ }
+}
+
+function enrollmentRowUsed(): boolean {
+    const row = DatabaseService.getInstance().getDb()
+        .prepare('SELECT used_at FROM pilot_enrollments WHERE node_id = ?')
+        .get(nodeId) as { used_at: number | null } | undefined;
+    return row?.used_at != null;
+}
+
+function waitForToken(
+    agent: InstanceType<typeof PilotAgent>,
+    predicate: (token: string) => boolean,
+    timeoutMs = 3_000,
+): Promise<string> {
+    return new Promise((resolve, reject) => {
+        const started = Date.now();
+        const tick = () => {
+            const token = (agent as unknown as { token: string }).token;
+            if (predicate(token)) {
+                resolve(token);
+                return;
+            }
+            if (Date.now() - started > timeoutMs) {
+                reject(new Error(`timed out waiting for token change; still ${token.slice(0, 24)}…`));
+                return;
+            }
+            setTimeout(tick, 25);
+        };
+        tick();
+    });
+}
+
 describe('clearPersistedToken', () => {
     it('removes an existing pilot.jwt file', () => {
         persistToken('stale-token');
@@ -112,7 +174,7 @@ describe('clearPersistedToken', () => {
 });
 
 describe('pilot tunnel upgrade rejection (in-process integration)', () => {
-    it('rejects a stale tunnel JWT signed with the wrong secret at upgrade', async () => {
+    it('rejects a stale tunnel JWT signed with the wrong secret at upgrade with reject header', async () => {
         const staleToken = mintStaleTunnelTokenWrongSecret();
         const ws = new WebSocket(`ws://127.0.0.1:${port}/api/pilot/tunnel`, {
             headers: {
@@ -120,14 +182,19 @@ describe('pilot tunnel upgrade rejection (in-process integration)', () => {
                 'x-sencho-agent-version': 'auth-fallback-test/1.0',
             },
         });
-        const result = await new Promise<{ status?: number }>((resolve) => {
+        const result = await new Promise<{ status?: number; reason?: string }>((resolve) => {
             ws.on('unexpected-response', (_req, res) => {
-                resolve({ status: res.statusCode });
+                const reasonHeader = res.headers['x-sencho-pilot-reject'];
+                resolve({
+                    status: res.statusCode,
+                    reason: Array.isArray(reasonHeader) ? reasonHeader[0] : reasonHeader,
+                });
                 res.destroy();
             });
             ws.on('error', () => { /* close follows */ });
         });
         expect(result.status).toBe(401);
+        expect(result.reason).toBe('invalid_token');
     });
 
     it('rejects a tunnel JWT for an unknown node with HTTP 404 at upgrade', async () => {
@@ -138,13 +205,173 @@ describe('pilot tunnel upgrade rejection (in-process integration)', () => {
                 'x-sencho-agent-version': 'auth-fallback-test/1.0',
             },
         });
-        const result = await new Promise<{ status?: number }>((resolve) => {
+        const result = await new Promise<{ status?: number; reason?: string }>((resolve) => {
             ws.on('unexpected-response', (_req, res) => {
-                resolve({ status: res.statusCode });
+                const reasonHeader = res.headers['x-sencho-pilot-reject'];
+                resolve({
+                    status: res.statusCode,
+                    reason: Array.isArray(reasonHeader) ? reasonHeader[0] : reasonHeader,
+                });
                 res.destroy();
             });
             ws.on('error', () => { /* close follows */ });
         });
         expect(result.status).toBe(404);
+        expect(result.reason).toBe('unknown_node');
+    });
+});
+
+describe('PilotAgent reconnect recovery (real hub)', () => {
+    it('swaps to a fresh enroll token on hub 401 and leaves pilot.jwt unchanged until enroll_ack', async () => {
+        const diskToken = mintStaleTunnelTokenWrongSecret();
+        persistToken(diskToken);
+        const enroll = mintFreshEnrollForNode();
+
+        const agent = new PilotAgent({
+            primaryUrl: `http://127.0.0.1:${port}`,
+            loopbackPort: 1,
+            initialToken: diskToken,
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        await waitForToken(agent, (t) => t === enroll);
+
+        expect(readPersistedToken()).toBe(diskToken);
+        expect(enrollmentRowUsed()).toBe(false);
+
+        // Next reconnect should complete enrollment and overwrite disk.
+        await waitForToken(agent, (t) => t !== enroll && t !== diskToken, 5_000);
+        // Allow persistToken + any in-flight sibling dial to settle.
+        await new Promise((r) => setTimeout(r, 100));
+        const finalToken = (agent as unknown as { token: string }).token;
+        expect(finalToken).not.toBe(diskToken);
+        expect(finalToken).not.toBe(enroll);
+        expect(readPersistedToken()).toBe(finalToken);
+        expect(enrollmentRowUsed()).toBe(true);
+
+        stopAgent(agent);
+    });
+
+    it('keeps pilot.jwt and dial token when enroll is expired', async () => {
+        const diskToken = mintStaleTunnelTokenWrongSecret();
+        persistToken(diskToken);
+        const agent = new PilotAgent({
+            primaryUrl: `http://127.0.0.1:${port}`,
+            loopbackPort: 1,
+            initialToken: diskToken,
+            enrollToken: mintExpiredEnroll(),
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        await new Promise((r) => setTimeout(r, 400));
+
+        expect(readPersistedToken()).toBe(diskToken);
+        expect((agent as unknown as { token: string }).token).toBe(diskToken);
+
+        stopAgent(agent);
+    });
+
+    it('does not consume enrollment or swap on a post-handshake capacity close (1013)', async () => {
+        const goodTunnel = jwt.sign(
+            { scope: 'pilot_tunnel', nodeId },
+            TEST_JWT_SECRET,
+            { expiresIn: '365d' },
+        );
+        persistToken(goodTunnel);
+        const enroll = mintFreshEnrollForNode();
+
+        const agent = new PilotAgent({
+            primaryUrl: `http://127.0.0.1:${port}`,
+            loopbackPort: 1,
+            initialToken: goodTunnel,
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        const up = new Promise<void>((resolve) => {
+            PilotTunnelManager.getInstance().once('tunnel-up', () => resolve());
+        });
+        (agent as unknown as { connect: () => void }).connect();
+        await up;
+
+        // Force a post-handshake close that mimics capacity (1013).
+        PilotTunnelManager.getInstance().closeTunnel(nodeId);
+        await new Promise((r) => setTimeout(r, 300));
+
+        expect((agent as unknown as { token: string }).token).toBe(goodTunnel);
+        expect(readPersistedToken()).toBe(goodTunnel);
+        expect(enrollmentRowUsed()).toBe(false);
+
+        stopAgent(agent);
+    });
+
+    it('schedules exactly one reconnect per rejected upgrade', async () => {
+        const diskToken = mintStaleTunnelTokenWrongSecret();
+        persistToken(diskToken);
+
+        const agent = new PilotAgent({
+            primaryUrl: `http://127.0.0.1:${port}`,
+            loopbackPort: 1,
+            initialToken: diskToken,
+            enrollToken: mintExpiredEnroll(),
+            enrolling: false,
+        });
+
+        const connectSpy = vi.spyOn(agent as unknown as { connect: () => void }, 'connect');
+        (agent as unknown as { connect: () => void }).connect();
+        // Initial call + wait for one scheduled reconnect.
+        await new Promise((r) => setTimeout(r, 1_600));
+
+        // connect() was invoked once by us and once by scheduleReconnect.
+        expect(connectSpy.mock.calls.length).toBe(2);
+
+        stopAgent(agent);
+        connectSpy.mockRestore();
+    });
+});
+
+describe('PilotAgent opaque proxy 401 (no Sencho reject header)', () => {
+    it('still attempts enroll fallback when a fresh enroll token is present', async () => {
+        const opaque = http.createServer((_req, res) => {
+            res.writeHead(401, { 'Content-Type': 'text/plain' });
+            res.end('Unauthorized');
+        });
+        const opaquePort = await new Promise<number>((resolve, reject) => {
+            opaque.listen(0, '127.0.0.1', () => {
+                const addr = opaque.address();
+                if (!addr || typeof addr === 'string') {
+                    reject(new Error('listen failed'));
+                    return;
+                }
+                resolve(addr.port);
+            });
+        });
+
+        const diskToken = 'opaque-disk-token';
+        persistToken(diskToken);
+        const enroll = jwt.sign(
+            { scope: 'pilot_enroll', nodeId: 1, enrollNonce: crypto.randomUUID() },
+            'irrelevant',
+            { expiresIn: '15m' },
+        );
+
+        const agent = new PilotAgent({
+            primaryUrl: `http://127.0.0.1:${opaquePort}`,
+            loopbackPort: 1,
+            initialToken: diskToken,
+            enrollToken: enroll,
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+        await waitForToken(agent, (t) => t === enroll);
+
+        expect(readPersistedToken()).toBe(diskToken);
+
+        stopAgent(agent);
+        await new Promise<void>((resolve) => opaque.close(() => resolve()));
     });
 });

backend/src/__tests__/reject-upgrade.test.ts

@@ -0,0 +1,46 @@
+/**
+ * Byte-level contract for rejectUpgrade: omitted headers stay historically
+ * identical; optional headers are written before the terminating blank line.
+ */
+import { describe, it, expect, vi } from 'vitest';
+import type { Duplex } from 'stream';
+import { rejectUpgrade } from '../websocket/reject';
+
+function captureWrite(): { socket: Duplex; chunks: string[] } {
+  const chunks: string[] = [];
+  const socket = {
+    write: vi.fn((data: string | Buffer) => {
+      chunks.push(typeof data === 'string' ? data : data.toString('utf8'));
+      return true;
+    }),
+    destroy: vi.fn(),
+  } as unknown as Duplex;
+  return { socket, chunks };
+}
+
+describe('rejectUpgrade', () => {
+  it('emits the historical byte sequence when headers are omitted', () => {
+    const { socket, chunks } = captureWrite();
+    rejectUpgrade(socket, 401, 'Unauthorized');
+    expect(chunks.join('')).toBe('HTTP/1.1 401 Unauthorized\r\n\r\n');
+    expect(socket.destroy).toHaveBeenCalledOnce();
+  });
+
+  it('writes optional headers before the terminating blank line', () => {
+    const { socket, chunks } = captureWrite();
+    rejectUpgrade(socket, 401, 'Unauthorized', {
+      'X-Sencho-Pilot-Reject': 'invalid_token',
+      Connection: 'close',
+    });
+    const body = chunks.join('');
+    expect(body.startsWith('HTTP/1.1 401 Unauthorized\r\n')).toBe(true);
+    expect(body).toContain('X-Sencho-Pilot-Reject: invalid_token\r\n');
+    expect(body).toContain('Connection: close\r\n');
+    expect(body.endsWith('\r\n\r\n')).toBe(true);
+    const blankIndex = body.lastIndexOf('\r\n\r\n');
+    const headerBlock = body.slice(0, blankIndex);
+    expect(headerBlock).toContain('X-Sencho-Pilot-Reject: invalid_token');
+    expect(headerBlock).toContain('Connection: close');
+    expect(socket.destroy).toHaveBeenCalledOnce();
+  });
+});

backend/src/pilot/agent.ts

@@ -34,8 +34,35 @@ const RECONNECT_MAX_MS = 60_000;
 const LOOPBACK_TOKEN_TTL_SECONDS = 300;
 const LOOPBACK_TOKEN_REFRESH_SECONDS = 240;
 const PING_INTERVAL_MS = 30_000;
+/** Deliberate floor for enroll exp checks; gross host skew surfaces via the degraded operator log. */
+const ENROLL_EXP_SKEW_MS = 60_000;
 const TOKEN_PATH = path.join(process.env.DATA_DIR || '/app/data', 'pilot.jwt');
 
+interface UpgradeRejectInfo {
+    status: number;
+    reason?: string;
+}
+
+function headerValue(value: string | string[] | undefined): string | undefined {
+    return Array.isArray(value) ? value[0] : value;
+}
+
+/**
+ * True when the enroll JWT payload is still within its expiry window
+ * (signature is not checked; the hub verifies on use).
+ */
+export function isFreshEnrollToken(token: string, nowMs: number = Date.now()): boolean {
+    try {
+        const payload = jwt.decode(token);
+        if (!payload || typeof payload === 'string') return false;
+        if (payload.scope !== 'pilot_enroll') return false;
+        if (typeof payload.exp !== 'number') return false;
+        return payload.exp * 1000 > nowMs - ENROLL_EXP_SKEW_MS;
+    } catch {
+        return false;
+    }
+}
+
 /**
  * Pilot agent: dials the primary via outbound WebSocket and tunnels every
  * inbound frame to the agent's own loopback HTTP server (the fully-booted
@@ -90,9 +117,17 @@ export class PilotAgent {
     private token: string;
     /** Fallback credential from SENCHO_ENROLL_TOKEN when the persisted token is rejected. */
     private readonly enrollToken: string | null;
+    /**
+     * Last-known-good tunnel JWT held in memory so a failed enroll fallback
+     * can restore without depending on a readable pilot.jwt (persist may have
+     * failed or the volume may be absent).
+     */
+    private tunnelToken: string | null = null;
+    /** At most one enroll-fallback attempt per credential generation. */
+    private enrollFallbackAttempted = false;
+    /** True after we have logged the degraded "re-enrollment required" state. */
+    private degradedLogged = false;
     private backoff = RECONNECT_MIN_MS;
-    /** Set when the WS upgrade is rejected with 401 or 404 before the handshake completes. */
-    private upgradeRejected = false;
     private ws: WebSocket | null = null;
     private pingTimer?: NodeJS.Timeout;
     private reconnectTimer?: NodeJS.Timeout;
@@ -120,6 +155,9 @@ export class PilotAgent {
         this.options = options;
         this.token = options.initialToken;
         this.enrollToken = options.enrollToken;
+        if (!options.enrolling) {
+            this.tunnelToken = options.initialToken;
+        }
         this.agentVersion = getSenchoVersion() || '0.0.0';
         this.customCa = readPilotCaBundle();
     }
@@ -177,8 +215,6 @@ export class PilotAgent {
     private connect(): void {
         if (this.shuttingDown) return;
 
-        this.upgradeRejected = false;
-
         const wsUrl = httpUrlToWs(this.options.primaryUrl) + '/api/pilot/tunnel';
         const ws = new WebSocket(wsUrl, {
             headers: {
@@ -196,20 +232,42 @@ export class PilotAgent {
             ...(this.customCa ? { ca: this.customCa } : {}),
         });
         this.ws = ws;
-        let opened = false;
-        let lastHandshakeError: string | null = null;
 
-        // Do NOT register an 'unexpected-response' listener here. The ws library
-        // skips abortHandshake (and therefore never emits 'error' or 'close')
-        // when a listener exists for that event. Auth rejection is detected via
-        // the error message abortHandshake emits: "Unexpected server response: N".
+        // Per-attempt locals so a late close cannot schedule a second reconnect.
+        let disconnectHandled = false;
+        let rejectInfo: UpgradeRejectInfo | null = null;
+
+        const handleDisconnect = (code?: number, reason?: Buffer): void => {
+            if (disconnectHandled) return;
+            disconnectHandled = true;
+            console.log('[Pilot] Tunnel closed:', code ?? '', reason?.toString?.() ?? '');
+            this.cleanupAfterDisconnect();
+            this.maybeAttemptEnrollFallback(rejectInfo);
+            this.scheduleReconnect();
+        };
+
+        // Primary path for HTTP upgrade rejects: this listener suppresses
+        // ws abortHandshake, so error/close alone would never fire. Capture
+        // status (+ optional X-Sencho-Pilot-Reject), drain, terminate, then
+        // run the single guarded disconnect path (later close is a no-op).
+        ws.on('unexpected-response', (_req, res) => {
+            const status = res.statusCode ?? 0;
+            rejectInfo = {
+                status,
+                reason: headerValue(res.headers['x-sencho-pilot-reject']),
+            };
+            try { res.resume(); } catch { /* ignore */ }
+            try { ws.terminate(); } catch { /* ignore */ }
+            handleDisconnect(status);
+        });
+
+        // Fallback if abortHandshake still emits without unexpected-response.
         ws.on('error', (err) => {
             console.warn('[Pilot] Tunnel error:', err.message);
-            lastHandshakeError = err.message;
-            if (/Unexpected server response: (401|404)/.test(err.message)) {
-                this.upgradeRejected = true;
+            const match = /Unexpected server response: (401|404)/.exec(err.message);
+            if (match && !rejectInfo) {
+                rejectInfo = { status: Number(match[1]) };
             }
-            // 'close' will follow; reconnect is scheduled there.
         });
 
         this.switchboard = attachTcpStreamSwitchboard({
@@ -220,7 +278,6 @@ export class PilotAgent {
         });
 
         ws.on('open', () => {
-            opened = true;
             // 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
@@ -245,22 +302,66 @@ export class PilotAgent {
         });
 
         ws.on('message', (data, isBinary) => this.handleFrame(data, isBinary));
-        ws.on('close', (code, reason) => {
-            console.log('[Pilot] Tunnel closed:', code, reason?.toString?.() ?? '');
-            this.cleanupAfterDisconnect();
+        ws.on('close', handleDisconnect);
+    }
 
-            const authRejected = this.upgradeRejected
-                || (!opened && lastHandshakeError != null && /Unexpected server response: (401|404)/.test(lastHandshakeError));
-            if (authRejected && this.enrollToken && this.token !== this.enrollToken) {
-                clearPersistedToken();
-                console.log('[Pilot] Persisted token rejected; falling back to enroll token');
-                this.token = this.enrollToken;
-                this.backoff = RECONNECT_MIN_MS;
-            }
-            this.upgradeRejected = false;
+    /**
+     * Attempt an in-memory swap to SENCHO_ENROLL_TOKEN after an upgrade 401/404.
+     * Never clears pilot.jwt here; disk is overwritten only on enroll_ack.
+     * Non-401/404 disconnects never start a fallback; a failed enroll dial
+     * restores the last-known tunnel token on any disconnect.
+     */
+    private maybeAttemptEnrollFallback(rejectInfo: UpgradeRejectInfo | null): void {
+        // Enroll dial failed (any disconnect): restore the last-known tunnel token.
+        if (this.enrollToken && this.token === this.enrollToken && this.enrollFallbackAttempted) {
+            this.restoreTunnelTokenAfterFailedEnroll();
+            return;
+        }
 
-            this.scheduleReconnect();
-        });
+        const status = rejectInfo?.status;
+        if (status !== 401 && status !== 404) return;
+
+        const reason = rejectInfo?.reason;
+        const enroll = this.enrollToken;
+        if (this.enrollFallbackAttempted || !enroll || enroll === this.token || !isFreshEnrollToken(enroll)) {
+            this.logDegraded(reason);
+            return;
+        }
+
+        console.log(
+            '[Pilot] Tunnel upgrade rejected'
+            + (reason ? ' (' + sanitizeForLog(reason) + ')' : '')
+            + '; trying fresh enroll token from SENCHO_ENROLL_TOKEN (disk unchanged until enroll_ack)',
+        );
+        this.enrollFallbackAttempted = true;
+        this.token = enroll;
+        this.backoff = RECONNECT_MIN_MS;
+    }
+
+    private restoreTunnelTokenAfterFailedEnroll(): void {
+        const restored = this.tunnelToken ?? readPersistedToken();
+        if (restored) {
+            this.token = restored;
+        }
+        this.logDegraded();
+    }
+
+    private logDegraded(reason?: string): void {
+        if (this.degradedLogged) return;
+        this.degradedLogged = true;
+        const detail = reason ? ` (reason: ${sanitizeForLog(reason)})` : '';
+        console.warn(
+            `[Pilot] Tunnel upgrade rejected${detail}; re-enrollment required.`
+            + ' Regenerate enrollment on the control instance, update the agent compose'
+            + ' SENCHO_ENROLL_TOKEN, and restart the agent.',
+        );
+    }
+
+    private clearDegradedState(): void {
+        if (this.degradedLogged) {
+            console.log('[Pilot] Tunnel credential recovered; leaving degraded state');
+        }
+        this.degradedLogged = false;
     }
 
     private cleanupAfterDisconnect(): void {
@@ -369,12 +470,16 @@ export class PilotAgent {
                 // that always rejects the handshake drive us into a tight
                 // reconnect loop.
                 this.backoff = RECONNECT_MIN_MS;
+                this.clearDegradedState();
                 break;
             }
             case 'ctrl': {
                 if (frame.op === 'enroll_ack' && frame.payload && typeof frame.payload.token === 'string') {
                     this.token = frame.payload.token;
+                    this.tunnelToken = frame.payload.token;
+                    this.enrollFallbackAttempted = false;
                     persistToken(this.token);
+                    this.clearDegradedState();
                     console.log('[Pilot] Enrollment complete; long-lived token persisted.');
                 }
                 break;
@@ -646,10 +751,10 @@ export function persistToken(token: string): void {
 }
 
 /**
- * Remove the persisted long-lived tunnel token from disk. Used when the
- * control instance rejects the stored credential at upgrade so the agent
- * can fall back to SENCHO_ENROLL_TOKEN without re-poisoning on restart.
- * ENOENT is normal when no file was written yet.
+ * Remove the persisted long-lived tunnel token from disk.
+ * Not used by the reconnect path (failed upgrades leave pilot.jwt in place
+ * until a successful enroll_ack overwrites it via persistToken). Kept for
+ * tests and explicit operator cleanup. ENOENT is normal when no file exists.
  *
  * Exposed for unit tests.
  */

backend/src/websocket/pilotTunnel.ts

@@ -10,6 +10,32 @@ import { encodeJsonFrame as encodePilotJsonFrame, PROTOCOL_VERSION as PILOT_PROT
 import { getErrorMessage } from '../utils/errors';
 import { rejectUpgrade as rejectSocket } from './reject';
 
+/** Diagnostic reject reason for Pilot agents (never required for enroll fallback). */
+type PilotRejectReason =
+  | 'missing_token'
+  | 'invalid_token'
+  | 'bad_scope'
+  | 'bad_node'
+  | 'unknown_node'
+  | 'enrollment_used'
+  | 'server_error';
+
+function firstHeader(value: string | string[] | undefined): string | undefined {
+  return Array.isArray(value) ? value[0] : value;
+}
+
+function rejectPilot(
+  socket: Duplex,
+  status: number,
+  message: string,
+  reason: PilotRejectReason,
+): void {
+  rejectSocket(socket, status, message, {
+    'X-Sencho-Pilot-Reject': reason,
+    Connection: 'close',
+  });
+}
+
 /**
  * Handle an inbound pilot-agent tunnel upgrade. Accepts either:
  *   - pilot_enroll (15m, one-time): consume the enrollment row, mint a
@@ -26,30 +52,29 @@ export async function handlePilotTunnel(
   head: Buffer,
   pilotTunnelWss: WebSocketServer,
 ): Promise<void> {
-  const authHeader = req.headers['authorization'];
-  const header = Array.isArray(authHeader) ? authHeader[0] : authHeader;
-  const token = header?.startsWith('Bearer ') ? header.slice(7) : null;
-  if (!token) return rejectSocket(socket, 401, 'Unauthorized');
+  const authHeader = firstHeader(req.headers['authorization']);
+  const token = authHeader?.startsWith('Bearer ') ? authHeader.slice(7) : null;
+  if (!token) return rejectPilot(socket, 401, 'Unauthorized', 'missing_token');
 
   const db = DatabaseService.getInstance();
   const jwtSecret = db.getGlobalSettings().auth_jwt_secret;
-  if (!jwtSecret) return rejectSocket(socket, 500, 'Internal Server Error');
+  if (!jwtSecret) return rejectPilot(socket, 500, 'Internal Server Error', 'server_error');
 
   let decoded: { scope?: string; nodeId?: number; enrollNonce?: string };
   try {
     decoded = jwt.verify(token, jwtSecret) as typeof decoded;
   } catch {
-    return rejectSocket(socket, 401, 'Unauthorized');
+    return rejectPilot(socket, 401, 'Unauthorized', 'invalid_token');
   }
 
   if (decoded.scope !== 'pilot_enroll' && decoded.scope !== 'pilot_tunnel') {
-    return rejectSocket(socket, 403, 'Forbidden');
+    return rejectPilot(socket, 403, 'Forbidden', 'bad_scope');
   }
-  if (typeof decoded.nodeId !== 'number') return rejectSocket(socket, 400, 'Bad Request');
+  if (typeof decoded.nodeId !== 'number') return rejectPilot(socket, 400, 'Bad Request', 'bad_node');
 
   const node = db.getNode(decoded.nodeId);
   if (!node || node.type !== 'remote' || node.mode !== 'pilot_agent') {
-    return rejectSocket(socket, 404, 'Not Found');
+    return rejectPilot(socket, 404, 'Not Found', 'unknown_node');
   }
 
   let mintedTunnelToken: string | null = null;
@@ -57,7 +82,7 @@ export async function handlePilotTunnel(
     const tokenHash = crypto.createHash('sha256').update(token).digest('hex');
     const row = db.consumePilotEnrollment(tokenHash);
     if (!row || row.node_id !== decoded.nodeId) {
-      return rejectSocket(socket, 401, 'Unauthorized');
+      return rejectPilot(socket, 401, 'Unauthorized', 'enrollment_used');
     }
     mintedTunnelToken = jwt.sign(
       { scope: 'pilot_tunnel', nodeId: decoded.nodeId },
@@ -67,8 +92,7 @@ export async function handlePilotTunnel(
     PilotMetrics.increment('enroll_acks');
   }
 
-  const agentVersionHeader = req.headers['x-sencho-agent-version'];
-  const agentVersion = Array.isArray(agentVersionHeader) ? agentVersionHeader[0] : agentVersionHeader;
+  const agentVersion = firstHeader(req.headers['x-sencho-agent-version']);
 
   pilotTunnelWss.handleUpgrade(req, socket, head, async (ws) => {
     try {

backend/src/websocket/reject.ts

@@ -5,8 +5,22 @@ import type { Duplex } from 'stream';
  * handler to reject an upgrade before a successful handshake. Errors during
  * write/destroy are intentionally swallowed: the socket is already being
  * torn down and nothing downstream can recover.
+ *
+ * Optional `headers` are written before the terminating blank line. When
+ * omitted, the response is exactly `HTTP/1.1 ${status} ${message}\r\n\r\n`
+ * so non-Pilot callers stay byte-identical to the historical shape.
  */
-export function rejectUpgrade(socket: Duplex, status: number, message: string): void {
-  try { socket.write(`HTTP/1.1 ${status} ${message}\r\n\r\n`); } catch { /* ignore */ }
+export function rejectUpgrade(
+  socket: Duplex,
+  status: number,
+  message: string,
+  headers?: Record<string, string>,
+): void {
+  try {
+    const extra = headers
+      ? Object.entries(headers).map(([name, value]) => `${name}: ${value}\r\n`).join('')
+      : '';
+    socket.write(`HTTP/1.1 ${status} ${message}\r\n${extra}\r\n`);
+  } catch { /* ignore */ }
   try { socket.destroy(); } catch { /* ignore */ }
 }