fix(mesh): address audit follow-ups (early-data, recompose, admiral gate, error codes, docs) (#1126)

* fix(mesh): buffer early TcpData on reverse-relay path The reverse-relay code dropped the local-shaped reservation and any TcpData frames buffered in it before acceptReverseRelay had wired up the target stream. A peer that sent a request body immediately after tcp_open_reverse lost those bytes when the target lived on a third node, since reverse_local buffers them but reverse_relay did not. Carry the reservation through acceptReverseRelay: transplant its pendingData and pendingBytes into the new reverse_relay state, gate TcpData writes on targetOpen, and flush buffered frames into the target stream before sending tcp_open_ack. Mirrors the reverse_local pattern. Regression test fires tcp_open_reverse plus an immediate TcpData while ensureBridge is in flight and verifies the bytes arrive intact, in order, before the ack. * fix(mesh): recompose affected stacks when node-level mesh is disabled disableForNode used to clear DB rows and override files but leave the running containers attached to sencho_mesh with stale /etc/hosts alias entries until an operator redeployed every stack by hand. Mirror optOutStack: after the existing alias/forwarder cleanup, call regenerateOverridesAcrossFleet, cascadeRecomposeAcrossFleet, and triggerRedeploy for each previously meshed stack on the disabled node so containers detach from sencho_mesh and shed the alias entries they owned. The disabled node's mesh_stacks rows are deleted before the cascade so listMeshStacks returns the right set with no skip tuple required. Route threads the actor through actorFor(req) for parity with optInStack/optOutStack. Tests cover the redeploy fan-out, the cascade no-skip-tuple invariant, and the default actor fallback for non-route callers. * fix(mesh): require Admiral on the WS proxy-tunnel upgrade HTTP mesh routes in routes/mesh.ts all enforce requireAdmiral, but the /api/mesh/proxy-tunnel WS upgrade accepted any node_proxy or full-admin api_token regardless of the receiver's license. A node downgraded from Admiral kept serving mesh data-plane traffic to a sibling central while refusing every mesh management call. Read the receiver's local LicenseService at the upgrade and 403 when the tier is not paid+admiral. The check sits after the existing credential gate and uses LicenseService directly rather than effectiveTier (which trusts forwarded proxy headers); a remote peer dialing in cannot be trusted to assert our entitlement. Dialer and node_proxy token format are unchanged. Three regression tests cover community-tier node_proxy, skipper-tier node_proxy, and community-tier full-admin api_token all rejected with 403. * fix(mesh): handle no_target alongside push_failed in inspectStackServices proxyFetch throws MeshError('no_target') when getProxyTarget returns null (pilot tunnel offline, proxy bridge unreachable), but the inspectStackServices catch branch only matched push_failed. Offline remotes fell through to the generic 'remote unreachable' error log, which the Routing tab surfaces as an unexpected fault. Match both error codes and emit the operator-friendly warn message that names the unreachable node and the error code. Regression test spies on console.warn/console.error to pin the branch. * docs(mesh): align env defaults and forwarder comments with current architecture SENCHO_MESH_PROXY_TUNNEL_IDLE_MS in .env.example carried the old five-minute idle-close value (=300000), but the code default is DEFAULT_IDLE_TTL_MS=0 (persistent tunnel). Copying the example silently reintroduced the idle-close behavior the dialer removed. MeshForwarder.ts's leading docblock and inline listen comment still described host-network mode plus extra_hosts: host-gateway as required for forwarder reachability. Sencho runs in standard bridge mode and attaches to the shared sencho_mesh network at a stable IP; meshed user containers reach the forwarder by that IP directly. Flip the env default to 0, rewrite the env comment to describe the persistent behavior and the opt-in for idle teardown, and rewrite both forwarder comments to match the bridge-network reality. * fix(mesh): trust forwarded tier on proxy-tunnel WS and remove remote overrides on disable Admiral entitlement on the WS data plane now follows the same trust model as the HTTP mesh routes: the central asserts its tier via x-sencho-tier and x-sencho-variant on the WS handshake, and the receiver trusts those headers only when the upgrade carries a node_proxy credential. When no headers are present or the credential is a full-admin api_token, the receiver falls back to its own local license. Without this an Admiral central could be rejected by a Community remote and a Community central could dial a locally-Admiral remote. disableForNode now routes through removeOverrideFromNode so override files pushed earlier via applyLocalOverride are removed on remote nodes via DELETE /api/mesh/local-override/:stack. Sequential awaits are wrapped in Promise.allSettled to match regenerateOverridesForNode's parallel push pattern. Other changes: - Cover the post-state-swap buffering window in the reverse-relay test (TcpData arriving after openTcpStream returns but before target open). - Refresh stale MeshService comments that still referenced the removed sidecar layer and host-network listener model. * test(mesh): pin removeOverrideFromNode remote HTTP shape The disableForNode regression test mocks removeOverrideFromNode itself, so a regression inside the helper would not be caught. Add a narrow contract test that spies on global fetch and asserts the request shape: DELETE /api/mesh/local-override/:stack against the resolved proxy target, with Authorization Bearer plus the x-sencho-tier and x-sencho-variant headers. Also covers the encodeURIComponent path and the swallowed-network-error behavior the disable cascade relies on. * test(mesh): use createTestApiToken helper in proxy-tunnel api_token gate test The full-admin api_token branch of the WS Admiral gate test inlined the canonical generateApiToken + sha256 + addApiToken triple that already lives in the createTestApiToken helper. Switching to the helper removes the duplicated insertion logic and aligns this test with the helper used by the other api_token call sites.
2026-07-26 11:49:16 +00:00 · 2026-05-20 12:53:29 -04:00
parent 08caa914ce
commit 982c7830b1
12 changed files with 658 additions and 57 deletions
@@ -120,9 +120,9 @@ SSO_CALLBACK_URL=
 # live regardless. This flag controls UI discovery only.
 SENCHO_EXPERIMENTAL=false

-# Idle teardown (in ms) for on-demand mesh tunnels that central opens
-# to Distributed API (proxy-mode) remotes. After this many milliseconds
-# of zero active mesh streams, central closes the WebSocket; the next
-# mesh dial re-opens it. Defaults to 5 minutes. Set to 0 to keep
-# tunnels open until the remote drops them or central shuts down.
-SENCHO_MESH_PROXY_TUNNEL_IDLE_MS=300000
+# Idle teardown for the persistent mesh proxy tunnel central opens to
+# Distributed API (proxy-mode) remotes. The tunnel stays open for the
+# life of the WebSocket. Set a positive number of milliseconds to
+# enable idle teardown after that many ms of zero active mesh streams;
+# the next mesh dial re-opens it. Default 0 (persistent).
+SENCHO_MESH_PROXY_TUNNEL_IDLE_MS=0
@@ -148,4 +148,44 @@ describe('MeshService.inspectStackServices dispatch (C-3 fix)', () => {
        expect(fetchSpy).not.toHaveBeenCalled();
        db.deleteNode(remoteNodeId);
    });
+
+    it('logs an offline-shaped warn (not a generic error) when proxyFetch throws MeshError no_target', async () => {
+        const svc = MeshService.getInstance();
+        const db = DatabaseService.getInstance();
+        const remoteNodeId = db.addNode({
+            name: 'inspect-remote-no-target',
+            type: 'remote',
+            mode: 'pilot_agent',
+            compose_dir: '/tmp',
+            is_default: false,
+            api_url: '',
+            api_token: '',
+        });
+
+        // getProxyTarget returns null -> proxyFetch raises MeshError('no_target').
+        // The catch branch must recognise no_target alongside push_failed and
+        // emit console.warn (operator-friendly), not console.error (which the
+        // Routing tab surfaces as an unexpected fault).
+        vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue(null);
+        const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => { /* swallow */ });
+        const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => { /* swallow */ });
+
+        const out = await (svc as unknown as { inspectStackServices: (n: number, s: string) => Promise<unknown> })
+            .inspectStackServices(remoteNodeId, 'audit-mesh-pilot');
+
+        expect(out).toEqual([]);
+        const warnedAboutNoTarget = warnSpy.mock.calls.some((args) =>
+            String(args[0] ?? '').includes('inspectStackServices: unreachable') &&
+            String(args[0] ?? '').includes('no_target'),
+        );
+        expect(warnedAboutNoTarget).toBe(true);
+        // No "remote unreachable" error log: that path is reserved for
+        // unexpected exceptions, not the known no_target / push_failed pair.
+        const erroredAsUnreachable = errorSpy.mock.calls.some((args) =>
+            String(args[0] ?? '').includes('inspectStackServices remote unreachable'),
+        );
+        expect(erroredAsUnreachable).toBe(false);
+
+        db.deleteNode(remoteNodeId);
+    });
 });
@@ -0,0 +1,128 @@
+/**
+ * Narrow contract test for `MeshService.removeOverrideFromNode` against a
+ * remote node. Pins the HTTP shape (method, path, headers) so a regression
+ * inside `removeOverrideFromNode` itself is caught even when callers like
+ * `disableForNode` are tested against a mock of the helper.
+ *
+ * Same fetch-spy pattern as `mesh-inspect-remote.test.ts`.
+ */
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest';
+import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
+
+let tmpDir: string;
+let MeshService: typeof import('../services/MeshService').MeshService;
+let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
+let NodeRegistry: typeof import('../services/NodeRegistry').NodeRegistry;
+
+beforeAll(async () => {
+    tmpDir = await setupTestDb();
+    ({ MeshService } = await import('../services/MeshService'));
+    ({ DatabaseService } = await import('../services/DatabaseService'));
+    ({ NodeRegistry } = await import('../services/NodeRegistry'));
+});
+
+afterAll(() => {
+    vi.restoreAllMocks();
+    cleanupTestDb(tmpDir);
+});
+
+afterEach(() => {
+    vi.restoreAllMocks();
+});
+
+describe('MeshService.removeOverrideFromNode (remote dispatch)', () => {
+    it('issues DELETE /api/mesh/local-override/<stack> with Authorization and tier headers', async () => {
+        const svc = MeshService.getInstance();
+        const db = DatabaseService.getInstance();
+        const remoteNodeId = db.addNode({
+            name: 'remove-override-remote-test',
+            type: 'remote',
+            mode: 'proxy',
+            compose_dir: '/tmp',
+            is_default: false,
+            api_url: 'https://remote.example.com:1852',
+            api_token: 'remote-tok',
+        });
+
+        vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue({
+            apiUrl: 'https://remote.example.com:1852',
+            apiToken: 'remote-tok',
+        });
+
+        const fetchMock = vi
+            .spyOn(globalThis, 'fetch')
+            .mockResolvedValue(new Response('ok', { status: 200 }));
+
+        await svc.removeOverrideFromNode(remoteNodeId, 'sample-stack');
+
+        expect(fetchMock).toHaveBeenCalledTimes(1);
+        const call = fetchMock.mock.calls[0];
+        expect(String(call[0])).toBe('https://remote.example.com:1852/api/mesh/local-override/sample-stack');
+        const init = call[1] as { method: string; headers: Record<string, string> };
+        expect(init.method).toBe('DELETE');
+        expect(init.headers['Authorization']).toBe('Bearer remote-tok');
+        expect(init.headers).toHaveProperty('x-sencho-tier');
+        expect(init.headers).toHaveProperty('x-sencho-variant');
+
+        db.deleteNode(remoteNodeId);
+    });
+
+    it('url-encodes the stack name in the path', async () => {
+        const svc = MeshService.getInstance();
+        const db = DatabaseService.getInstance();
+        const remoteNodeId = db.addNode({
+            name: 'remove-override-encode-test',
+            type: 'remote',
+            mode: 'proxy',
+            compose_dir: '/tmp',
+            is_default: false,
+            api_url: 'https://remote.example.com:1852',
+            api_token: 'remote-tok',
+        });
+
+        vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue({
+            apiUrl: 'https://remote.example.com:1852',
+            apiToken: 'remote-tok',
+        });
+
+        const fetchMock = vi
+            .spyOn(globalThis, 'fetch')
+            .mockResolvedValue(new Response('ok', { status: 200 }));
+
+        await svc.removeOverrideFromNode(remoteNodeId, 'stack with space');
+
+        const call = fetchMock.mock.calls[0];
+        expect(String(call[0])).toBe('https://remote.example.com:1852/api/mesh/local-override/stack%20with%20space');
+
+        db.deleteNode(remoteNodeId);
+    });
+
+    it('swallows network errors from the remote so the disable cascade can continue', async () => {
+        const svc = MeshService.getInstance();
+        const db = DatabaseService.getInstance();
+        const remoteNodeId = db.addNode({
+            name: 'remove-override-error-test',
+            type: 'remote',
+            mode: 'proxy',
+            compose_dir: '/tmp',
+            is_default: false,
+            api_url: 'https://remote.example.com:1852',
+            api_token: 'remote-tok',
+        });
+
+        vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue({
+            apiUrl: 'https://remote.example.com:1852',
+            apiToken: 'remote-tok',
+        });
+
+        vi.spyOn(globalThis, 'fetch').mockRejectedValue(new Error('ECONNREFUSED'));
+        const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => { /* silence */ });
+
+        // Must not throw: the remote being offline is a tolerable condition;
+        // the cascade upstream uses Promise.allSettled and continues.
+        await expect(svc.removeOverrideFromNode(remoteNodeId, 'sample-stack')).resolves.toBeUndefined();
+        expect(warnSpy).toHaveBeenCalled();
+
+        db.deleteNode(remoteNodeId);
+    });
+});
@@ -339,6 +339,118 @@ describe('MeshService.optInStack', () => {
    });
 });

+describe('MeshService.disableForNode', () => {
+    it('redeploys affected stacks and cascades across the fleet on node disable', async () => {
+        const svc = MeshService.getInstance();
+        const db = DatabaseService.getInstance();
+        const localNodeId = db.getNodes()[0].id;
+        const remoteNodeId = db.addNode({
+            name: 'remote-pilot', type: 'remote', mode: 'pilot_agent',
+            compose_dir: '/tmp', is_default: false, api_url: '', api_token: '',
+        });
+        db.setNodeMeshEnabled(localNodeId, true);
+        db.insertMeshStack(localNodeId, 'alpha', 'setup');
+        db.insertMeshStack(localNodeId, 'beta', 'setup');
+        db.insertMeshStack(remoteNodeId, 'gamma', 'setup');
+
+        const regenSpy = vi.spyOn(
+            svc as unknown as { regenerateOverridesAcrossFleet: (n?: number, s?: string) => Promise<void> },
+            'regenerateOverridesAcrossFleet',
+        ).mockResolvedValue(undefined);
+        const cascadeSpy = vi.spyOn(
+            svc as unknown as { cascadeRecomposeAcrossFleet: (n: number | undefined, s: string | undefined, a: string) => void },
+            'cascadeRecomposeAcrossFleet',
+        ).mockImplementation(() => { /* noop */ });
+        const triggerSpy = vi.spyOn(svc, 'triggerRedeploy').mockImplementation(() => { /* noop */ });
+        const removeSpy = vi.spyOn(svc, 'removeOverrideFromNode').mockResolvedValue(undefined);
+
+        await svc.disableForNode(localNodeId, 'tester');
+
+        expect(db.getNodeMeshEnabled(localNodeId)).toBe(false);
+        // Disabled-node rows are deleted before the cascade so listMeshStacks
+        // does not return them. The other node's row stays.
+        expect(db.listMeshStacks(localNodeId)).toEqual([]);
+        expect(db.listMeshStacks(remoteNodeId).map((s) => s.stack_name)).toEqual(['gamma']);
+
+        expect(regenSpy).toHaveBeenCalledTimes(1);
+        // Cascade walks every remaining mesh_stacks row, no skip tuple.
+        expect(cascadeSpy).toHaveBeenCalledWith(undefined, undefined, 'tester');
+        // Each disabled-node stack triggers a direct redeploy so its
+        // container detaches from sencho_mesh. Cascade is mocked, so the
+        // only triggerRedeploy calls come from the disable loop itself
+        // (twice, not three: gamma stays put on the remote node).
+        expect(triggerSpy).toHaveBeenCalledTimes(2);
+        expect(triggerSpy).toHaveBeenCalledWith(localNodeId, 'alpha', 'tester');
+        expect(triggerSpy).toHaveBeenCalledWith(localNodeId, 'beta', 'tester');
+        // disableForNode routes through removeOverrideFromNode (not the
+        // local-only removeStackOverride) so remote nodes also have their
+        // pushed override files deleted via DELETE /api/mesh/local-override.
+        expect(removeSpy).toHaveBeenCalledTimes(2);
+        expect(removeSpy).toHaveBeenCalledWith(localNodeId, 'alpha');
+        expect(removeSpy).toHaveBeenCalledWith(localNodeId, 'beta');
+
+        db.deleteNode(remoteNodeId);
+    });
+
+    it('dispatches removeOverrideFromNode for each stack when the disabled node is remote', async () => {
+        const svc = MeshService.getInstance();
+        const db = DatabaseService.getInstance();
+        const remoteNodeId = db.addNode({
+            name: 'remote-pilot-disable', type: 'remote', mode: 'pilot_agent',
+            compose_dir: '/tmp', is_default: false, api_url: '', api_token: '',
+        });
+        db.setNodeMeshEnabled(remoteNodeId, true);
+        db.insertMeshStack(remoteNodeId, 'delta', 'setup');
+        db.insertMeshStack(remoteNodeId, 'epsilon', 'setup');
+
+        vi.spyOn(
+            svc as unknown as { regenerateOverridesAcrossFleet: (n?: number, s?: string) => Promise<void> },
+            'regenerateOverridesAcrossFleet',
+        ).mockResolvedValue(undefined);
+        vi.spyOn(
+            svc as unknown as { cascadeRecomposeAcrossFleet: (n: number | undefined, s: string | undefined, a: string) => void },
+            'cascadeRecomposeAcrossFleet',
+        ).mockImplementation(() => { /* noop */ });
+        vi.spyOn(svc, 'triggerRedeploy').mockImplementation(() => { /* noop */ });
+        const removeSpy = vi.spyOn(svc, 'removeOverrideFromNode').mockResolvedValue(undefined);
+
+        await svc.disableForNode(remoteNodeId, 'tester');
+
+        expect(db.getNodeMeshEnabled(remoteNodeId)).toBe(false);
+        // Remote-node disable must dispatch the remote-aware helper so the
+        // override file pushed earlier via applyLocalOverride is deleted
+        // on the peer via DELETE /api/mesh/local-override/:stack.
+        expect(removeSpy).toHaveBeenCalledTimes(2);
+        expect(removeSpy).toHaveBeenCalledWith(remoteNodeId, 'delta');
+        expect(removeSpy).toHaveBeenCalledWith(remoteNodeId, 'epsilon');
+
+        db.deleteNode(remoteNodeId);
+    });
+
+    it('defaults the actor when none is supplied so legacy callers still log a non-empty actor', async () => {
+        const svc = MeshService.getInstance();
+        const db = DatabaseService.getInstance();
+        const localNodeId = db.getNodes()[0].id;
+        db.setNodeMeshEnabled(localNodeId, true);
+        db.insertMeshStack(localNodeId, 'solo', 'setup');
+
+        vi.spyOn(svc as unknown as { regenerateOverridesAcrossFleet: () => Promise<void> }, 'regenerateOverridesAcrossFleet')
+            .mockResolvedValue(undefined);
+        const cascadeSpy = vi.spyOn(
+            svc as unknown as { cascadeRecomposeAcrossFleet: (n: number | undefined, s: string | undefined, a: string) => void },
+            'cascadeRecomposeAcrossFleet',
+        ).mockImplementation(() => { /* noop */ });
+        const triggerSpy = vi.spyOn(svc, 'triggerRedeploy').mockImplementation(() => { /* noop */ });
+        vi.spyOn(svc, 'removeOverrideFromNode').mockResolvedValue(undefined);
+
+        await svc.disableForNode(localNodeId);
+
+        expect(db.getNodeMeshEnabled(localNodeId)).toBe(false);
+        expect(cascadeSpy.mock.calls[0][2]).toBe('system:mesh.disable');
+        expect(triggerSpy.mock.calls[0][2]).toBe('system:mesh.disable');
+    });
+});
+
 describe('MeshService activity log', () => {
    it('keeps the most recent events under the 1000-cap', () => {
        const svc = MeshService.getInstance();
@@ -212,4 +212,84 @@ describe('PilotTunnelBridge handles tcp_open_reverse (Phase B)', () => {
        bridge.close();
        vi.restoreAllMocks();
    });
+
+    it('buffers TcpData arriving during ensureBridge and flushes it on target open (reverse-relay early-data fix)', async () => {
+        const mockWs = makeMockTunnelWs();
+        const bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket);
+        await bridge.start();
+
+        // Fake target stream: an EventEmitter with a write() method that
+        // captures bytes. We control when 'open' fires to widen the
+        // race window the fix guards.
+        const writes: Buffer[] = [];
+        const fakeTargetStream = new EventEmitter() as EventEmitter & {
+            write: (b: Buffer) => void;
+        };
+        fakeTargetStream.write = (b: Buffer) => { writes.push(b); };
+
+        const fakeTargetBridge = {
+            openTcpStream: vi.fn(() => fakeTargetStream),
+            getBufferedAmount: () => 0,
+        };
+
+        const { PilotTunnelManager } = await import('../services/PilotTunnelManager');
+        // Delay ensureBridge so a TcpData frame fired back-to-back with
+        // tcp_open_reverse lands while acceptReverseRelay is still
+        // waiting on the target bridge. Without the fix the bytes are
+        // dropped at the lookup-miss path in handleBinaryFrame.
+        vi.spyOn(PilotTunnelManager.getInstance(), 'ensureBridge')
+            .mockImplementation(() => new Promise((r) => setTimeout(() => r(fakeTargetBridge as never), 50)));
+
+        const { NodeRegistry } = await import('../services/NodeRegistry');
+        const localNodeId = NodeRegistry.getInstance().getDefaultNodeId();
+        const remoteTargetNodeId = localNodeId + 99;
+
+        const s = AGENT_REVERSE_ID_BASE + 7;
+        mockWs.emit('message', encodeJsonFrame({
+            t: 'tcp_open_reverse', s,
+            targetNodeId: remoteTargetNodeId, stack: 'real', service: 'svc', port: 8080,
+        }), false);
+        // Immediately push the request bytes while ensureBridge is in flight.
+        mockWs.emit('message',
+            encodeBinaryFrame(BinaryFrameType.TcpData, s, Buffer.from('POST /hook HTTP/1.1\r\n\r\n')),
+            true);
+
+        // Wait until acceptReverseRelay has actually swapped the reservation
+        // to kind:'reverse_relay' (and targetOpen is still false because the
+        // target hasn't emitted 'open'). Polling the bridge's stream state
+        // directly is more deterministic than polling openTcpStream.mock
+        // calls, since the state swap is synchronous after openTcpStream
+        // returns but the JS scheduler decides when the test sees it.
+        const bridgeStreams = (bridge as unknown as { streams: Map<number, { kind: string; targetOpen?: boolean }> }).streams;
+        await waitFor(() => {
+            const cur = bridgeStreams.get(s);
+            return (cur?.kind === 'reverse_relay' && cur.targetOpen === false) ? true : undefined;
+        });
+        // Second TcpData frame in the targetOpen===false window after the
+        // state swap. Without the reverse_relay branch's pendingData buffer,
+        // these bytes would be silently dropped.
+        mockWs.emit('message',
+            encodeBinaryFrame(BinaryFrameType.TcpData, s, Buffer.from('mid-window')),
+            true);
+        // Fire 'open' on the fake target so the relay flushes its pending buffer.
+        fakeTargetStream.emit('open');
+
+        const ack = await waitFor(() => findAck(mockWs, s));
+        expect(ack.ok).toBe(true);
+        // Both pre-openTcpStream and post-state-swap buffered bytes were
+        // flushed exactly once, intact, in order.
+        await waitFor(() => (Buffer.concat(writes).toString().includes('mid-window')) ? true : undefined);
+        const combined = Buffer.concat(writes).toString();
+        expect(combined).toBe('POST /hook HTTP/1.1\r\n\r\nmid-window');
+
+        // Subsequent post-open writes also flow.
+        mockWs.emit('message',
+            encodeBinaryFrame(BinaryFrameType.TcpData, s, Buffer.from('trailer')),
+            true);
+        await waitFor(() => (Buffer.concat(writes).toString().endsWith('trailer')) ? true : undefined);
+        expect(Buffer.concat(writes).toString()).toBe('POST /hook HTTP/1.1\r\n\r\nmid-windowtrailer');
+
+        bridge.close();
+        vi.restoreAllMocks();
+    });
 });
@@ -8,13 +8,14 @@
 * product paths. This test pins each position by observing behavior that
 * could only originate from the expected handler.
 */
-import { describe, it, expect, beforeAll, afterAll } from 'vitest';
+import { describe, it, expect, beforeAll, afterAll, afterEach } from 'vitest';
 import WebSocket from 'ws';
 import jwt from 'jsonwebtoken';
 import crypto from 'crypto';
 import type { AddressInfo } from 'net';
 import { setupTestDb, cleanupTestDb, TEST_USERNAME, TEST_JWT_SECRET } from './helpers/setupTestDb';
 import { generateApiToken } from '../utils/apiTokenFormat';
+import { createTestApiToken } from './helpers/apiTokenTestHelper';

 describe('WebSocket upgrade dispatch order', () => {
  let tmpDir: string;
@@ -43,19 +44,35 @@ describe('WebSocket upgrade dispatch order', () => {

    const token = jwt.sign({ username: TEST_USERNAME }, TEST_JWT_SECRET, { expiresIn: '1m' });
    sessionCookie = `sencho_token=${token}`;
+
+    // Existing /api/mesh/proxy-tunnel scope tests assume the receiver
+    // license clears the Admiral check. Set the license to Admiral so
+    // the credential-only assertions below still hold; the dedicated
+    // license-gating describe block flips and restores per-test.
+    DatabaseService.getInstance().setSystemState('license_status', 'active');
+    DatabaseService.getInstance().setSystemState('license_variant_type', 'admiral');
  });

  afterAll(async () => {
+    // Clear the Admiral state beforeAll set so this file does not leak
+    // license context into other tests sharing the same test DB.
+    const { DatabaseService } = await import('../services/DatabaseService');
+    DatabaseService.getInstance().setSystemState('license_status', 'community');
+    DatabaseService.getInstance().setSystemState('license_variant_type', '');
    await new Promise<void>((resolve, reject) => {
      server.close((err) => (err ? reject(err) : resolve()));
    });
    cleanupTestDb(tmpDir);
  });

-  function connect(pathAndQuery: string, opts: { cookie?: string; bearer?: string } = {}): WebSocket {
+  function connect(
+    pathAndQuery: string,
+    opts: { cookie?: string; bearer?: string; extraHeaders?: Record<string, string> } = {},
+  ): WebSocket {
    const headers: Record<string, string> = {};
    if (opts.cookie) headers['cookie'] = opts.cookie;
    if (opts.bearer) headers['authorization'] = `Bearer ${opts.bearer}`;
+    if (opts.extraHeaders) Object.assign(headers, opts.extraHeaders);
    return new WebSocket(`ws://127.0.0.1:${port}${pathAndQuery}`, { headers });
  }

@@ -218,6 +235,113 @@ describe('WebSocket upgrade dispatch order', () => {
    });
  });

+  describe('/api/mesh/proxy-tunnel Admiral entitlement gating', () => {
+    // Admiral entitlement on the WS data plane is decided against the
+    // *central's* asserted tier, matching the HTTP mesh routes
+    // (requireAdmiral in routes/mesh.ts reads req.proxyTier from forwarded
+    // headers off the node_proxy credential). The WS dispatcher trusts
+    // x-sencho-tier / x-sencho-variant only when the upgrade carries a
+    // node_proxy JWT; when no headers are present, or when the credential
+    // is a full-admin api_token (no central is asserting tier), it falls
+    // back to the receiver's own license. These tests pin both branches:
+    // (a) the trusted-header path accepts an Admiral central even on a
+    //     Community receiver, and rejects a Community central on an
+    //     Admiral receiver;
+    // (b) the local-fallback path keeps the receiver-license check intact
+    //     for full-admin api_token upgrades and for header-less node_proxy
+    //     upgrades.
+
+    async function setLicense(status: string, variantType: string | null): Promise<void> {
+      const { DatabaseService } = await import('../services/DatabaseService');
+      DatabaseService.getInstance().setSystemState('license_status', status);
+      if (variantType === null) {
+        DatabaseService.getInstance().setSystemState('license_variant_type', '');
+      } else {
+        DatabaseService.getInstance().setSystemState('license_variant_type', variantType);
+      }
+    }
+
+    afterEach(async () => {
+      // Restore the Admiral state beforeAll established so subsequent
+      // tests in this file (and the proxy-tunnel scope block) keep passing.
+      await setLicense('active', 'admiral');
+    });
+
+    it('rejects a node_proxy Bearer with HTTP 403 when no tier headers and the receiver license is community', async () => {
+      await setLicense('community', null);
+      const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' });
+      const ws = connect('/api/mesh/proxy-tunnel', { bearer: nodeProxyToken });
+      const outcome = await waitForOutcome(ws);
+      expect(outcome.kind).toBe('unexpected');
+      if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403);
+    });
+
+    it('rejects a node_proxy Bearer with HTTP 403 when no tier headers and the receiver license is paid but Skipper (not Admiral)', async () => {
+      await setLicense('active', 'skipper');
+      const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' });
+      const ws = connect('/api/mesh/proxy-tunnel', { bearer: nodeProxyToken });
+      const outcome = await waitForOutcome(ws);
+      expect(outcome.kind).toBe('unexpected');
+      if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403);
+    });
+
+    it('rejects a full-admin api_token with HTTP 403 when the receiver license is community (forwarded headers ignored on api_token path)', async () => {
+      await setLicense('community', null);
+      const { DatabaseService } = await import('../services/DatabaseService');
+      const adminId = DatabaseService.getInstance().getUserByUsername(TEST_USERNAME)!.id;
+      const rawToken = createTestApiToken({
+        db: DatabaseService,
+        scope: 'full-admin',
+        userId: adminId,
+        name: `mesh-license-gate-${Date.now()}`,
+      });
+      // Header is set but must be ignored: the full-admin api_token is a
+      // local-entitlement credential, not a node_proxy forwarder.
+      const ws = connect('/api/mesh/proxy-tunnel', {
+        bearer: rawToken,
+        extraHeaders: { 'x-sencho-tier': 'paid', 'x-sencho-variant': 'admiral' },
+      });
+      const outcome = await waitForOutcome(ws);
+      expect(outcome.kind).toBe('unexpected');
+      if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403);
+    });
+
+    it('accepts a node_proxy Bearer asserting paid+admiral via forwarded headers even when the receiver license is community', async () => {
+      await setLicense('community', null);
+      const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' });
+      const ws = connect('/api/mesh/proxy-tunnel', {
+        bearer: nodeProxyToken,
+        extraHeaders: { 'x-sencho-tier': 'paid', 'x-sencho-variant': 'admiral' },
+      });
+      const outcome = await waitForOutcome(ws);
+      expect(outcome.kind).toBe('open');
+      try { ws.terminate(); } catch { /* ignore */ }
+    });
+
+    it('rejects a node_proxy Bearer asserting community via forwarded headers even when the receiver license is Admiral', async () => {
+      // Receiver state is already Admiral (set by beforeAll / afterEach).
+      const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' });
+      const ws = connect('/api/mesh/proxy-tunnel', {
+        bearer: nodeProxyToken,
+        extraHeaders: { 'x-sencho-tier': 'community' },
+      });
+      const outcome = await waitForOutcome(ws);
+      expect(outcome.kind).toBe('unexpected');
+      if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403);
+    });
+
+    it('rejects a node_proxy Bearer asserting paid+skipper via forwarded headers (paid but not Admiral)', async () => {
+      const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' });
+      const ws = connect('/api/mesh/proxy-tunnel', {
+        bearer: nodeProxyToken,
+        extraHeaders: { 'x-sencho-tier': 'paid', 'x-sencho-variant': 'skipper' },
+      });
+      const outcome = await waitForOutcome(ws);
+      expect(outcome.kind).toBe('unexpected');
+      if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403);
+    });
+  });
+
  it('dispatches /api/pilot/tunnel to the pilot handler (rejects non-pilot bearer before path-based dispatch)', async () => {
    // A plain session cookie is a valid *user* JWT but not a pilot JWT. The
    // pilot handler runs first (before the shared cookie/Bearer auth) and
@@ -85,7 +85,7 @@ meshRouter.post('/nodes/:nodeId/disable', async (req: Request, res: Response): P
    const nodeId = Number.parseInt(req.params.nodeId as string, 10);
    if (!Number.isFinite(nodeId)) { res.status(400).json({ error: 'Invalid node id' }); return; }
    try {
-        await MeshService.getInstance().disableForNode(nodeId);
+        await MeshService.getInstance().disableForNode(nodeId, actorFor(req));
        res.json({ ok: true });
    } catch (err) {
        res.status(500).json({ error: (err as Error).message });
@@ -2,19 +2,17 @@ import net from 'net';
 import { sanitizeForLog } from '../utils/safeLog';

 /**
- * In-process mesh TCP forwarder. Owns per-port `net.Server` listeners on the
- * host network and delegates accepted sockets to the host (MeshService) for
- * resolve + splice. Replaces the separate `saelix/sencho-mesh` sidecar
- * container that previously did this job over a control WebSocket. The
- * resolve step is now a sync map lookup rather than a round-trip, so
- * MeshForwarder is just a thin lifecycle layer; all routing + splicing
- * lives on MeshService.
+ * In-process mesh TCP forwarder. Owns per-port `net.Server` listeners and
+ * delegates accepted sockets to the host (MeshService) for resolve +
+ * splice. All routing and splicing logic lives on MeshService;
+ * MeshForwarder is a thin lifecycle layer that opens and closes ports.
 *
- * Sencho's container must run in `network_mode: host` (Linux) for the
- * listeners to bind on the host's network where meshed containers'
- * `extra_hosts: <alias>:host-gateway` entries point. Without host network
- * mode, `net.createServer().listen(port)` lands inside the container's
- * namespace and inbound traffic from peers never reaches it.
+ * Sencho runs in standard Docker bridge mode and attaches its own
+ * container to the shared `sencho_mesh` network at a stable IP. Meshed
+ * user containers are attached to the same network, so they reach the
+ * forwarder by the Sencho IP without `network_mode: host` or the
+ * `extra_hosts: <alias>:host-gateway` indirection an older sidecar
+ * design required.
 */

 export interface MeshForwarderHost {
@@ -50,9 +48,10 @@ export class MeshForwarder {
                    const onListening = () => { server.removeListener('error', onError); resolve(); };
                    server.once('error', onError);
                    server.once('listening', onListening);
-                    // Bind on all interfaces. Under host network mode this is
-                    // the host's own network; under bridge mode (mesh disabled
-                    // at boot) this would be the container's namespace.
+                    // Bind on all interfaces inside the Sencho container's
+                    // networking namespace. The sencho_mesh bridge attaches
+                    // both Sencho and meshed user containers, so peers reach
+                    // this listener at Sencho's mesh-network IP.
                    server.listen(port, '0.0.0.0');
                });
                this.listeners.set(port, server);
@@ -9,6 +9,8 @@ import { httpUrlToWs } from '../utils/wsUrl';
 import { isDebugEnabled } from '../utils/debug';
 import { PilotMetrics } from './PilotMetrics';
 import type { MeshActivityType } from './MeshService';
+import { LicenseService } from './LicenseService';
+import { PROXY_TIER_HEADER, PROXY_VARIANT_HEADER } from './license-headers';

 /**
 * Central-side dialer for proxy-mode mesh tunnels.
@@ -234,10 +236,23 @@ export class MeshProxyTunnelDialer extends EventEmitter {
        // the peer falls back to its local DB default (always 1) and treats
        // cross-node aliases as same-node.
        const wsUrl = httpUrlToWs(target.apiUrl) + `/api/mesh/proxy-tunnel?nodeId=${nodeId}`;
+        // Forward central's tier and variant so the receiver enforces Admiral
+        // against the *central's* license (matching the HTTP mesh routes,
+        // which all gate on `requireAdmiral` against `req.proxyTier`). Without
+        // these the receiver falls back to its own local license, which would
+        // both reject Admiral centrals talking to Community remotes and let
+        // Community centrals dial locally-Admiral remotes. The headers are
+        // trusted on the receiver only when the WS carries a node_proxy /
+        // pilot_tunnel credential (see middleware/auth.ts:117-135).
+        const proxyHeaders = LicenseService.getInstance().getProxyHeaders();
        let ws: WebSocket;
        try {
            ws = new WebSocket(wsUrl, {
-                headers: { Authorization: `Bearer ${target.apiToken}` },
+                headers: {
+                    Authorization: `Bearer ${target.apiToken}`,
+                    [PROXY_TIER_HEADER]: proxyHeaders.tier,
+                    [PROXY_VARIANT_HEADER]: proxyHeaders.variant || '',
+                },
                handshakeTimeout: HANDSHAKE_TIMEOUT_MS,
                maxPayload: MAX_FRAME_SIZE_BYTES,
            });
@@ -242,9 +242,8 @@ interface ActiveStreamRecord {

 /**
 * Sencho Mesh orchestrator. Owns:
- *   - in-process TCP forwarder (`MeshForwarder`) that binds host-network
- *     listeners on alias ports. Replaces the prior separate sidecar
- *     container; one container per node now.
+ *   - in-process TCP forwarder (`MeshForwarder`) that binds per-alias
+ *     listeners on Sencho's `sencho_mesh` bridge-network IP.
 *   - opt-in / opt-out persistence and cascading override regeneration
 *   - global alias aggregation (across the fleet via the existing HTTP
 *     proxy chain, see `inspectStackServices`)
@@ -258,8 +257,6 @@ interface ActiveStreamRecord {
 *     Docker bridge network. Meshed user services join `sencho_mesh` so
 *     that IP is reachable from inside their containers without any
 *     host-firewall coordination.
- *   - cross-node mesh routing is central → pilot in this phase. Pilot →
- *     central and pilot ↔ pilot via central relay land in Phase B.
 */
 export class MeshService extends EventEmitter implements MeshForwarderHost {
    private static instance: MeshService;
@@ -904,15 +901,38 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
        }
    }

-    public async disableForNode(nodeId: number): Promise<void> {
+    public async disableForNode(
+        nodeId: number,
+        actor: string = 'system:mesh.disable',
+    ): Promise<void> {
        DatabaseService.getInstance().setNodeMeshEnabled(nodeId, false);
        const stacks = DatabaseService.getInstance().listMeshStacks(nodeId);
        for (const s of stacks) {
            DatabaseService.getInstance().deleteMeshStack(nodeId, s.stack_name);
-            await this.removeStackOverride(nodeId, s.stack_name);
        }
+        // Dispatch DELETE /api/mesh/local-override/:stack for remote nodes
+        // (pilot or proxy) so the override file pushed earlier via
+        // applyLocalOverride is removed; falls back to local deletion for
+        // local nodes. Parallelize per the regenerateOverridesForNode
+        // rationale: each remote call is its own HTTP round-trip, so
+        // awaiting sequentially turns N stacks into N serialised DELETEs.
+        // `allSettled` so a single failure does not abort the others
+        // (removeOverrideFromNode already swallows errors internally).
+        await Promise.allSettled(
+            stacks.map((s) => this.removeOverrideFromNode(nodeId, s.stack_name)),
+        );
        await this.refreshAliasCache();
        await this.syncForwarderListeners();
+        // Mirror optOutStack: regenerate every remaining node's override
+        // without the dropped aliases, recompose the rest of the fleet so
+        // their containers shed the stale extra_hosts, and redeploy the
+        // disabled node's own stacks so their containers detach from the
+        // sencho_mesh network and lose the alias entries they owned.
+        await this.regenerateOverridesAcrossFleet();
+        this.cascadeRecomposeAcrossFleet(undefined, undefined, actor);
+        for (const s of stacks) {
+            this.triggerRedeploy(nodeId, s.stack_name, actor);
+        }
        this.logActivity({
            source: 'mesh', level: 'info', type: 'mesh.disable',
            nodeId, message: `mesh disabled on node ${nodeId}`,
@@ -1415,11 +1435,13 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
            const body = await res.json() as { services?: Array<{ service: string; ports: number[] }> };
            return body.services ?? [];
        } catch (err) {
-            // proxyFetch throws MeshError('push_failed') when getProxyTarget
-            // returns null (e.g. pilot tunnel offline). Treat the same as a
-            // non-OK response: empty list.
-            if (err instanceof MeshError && err.code === 'push_failed') {
-                console.warn(`[MeshService] inspectStackServices: no proxy target for node ${nodeId} (${sanitizeForLog(node.name)})`);
+            // proxyFetch throws MeshError('no_target') when getProxyTarget
+            // returns null (pilot tunnel offline, proxy bridge unreachable)
+            // and MeshError('push_failed') on a non-OK HTTP response. Treat
+            // both as a soft "no services to report" so the Routing tab does
+            // not surface a stack trace for a node that is simply offline.
+            if (err instanceof MeshError && (err.code === 'no_target' || err.code === 'push_failed')) {
+                console.warn(`[MeshService] inspectStackServices: unreachable node ${nodeId} (${sanitizeForLog(node.name)}): ${err.code}`);
                return [];
            }
            console.error('[MeshService] inspectStackServices remote unreachable:', sanitizeForLog((err as Error).message));
@@ -1690,12 +1712,12 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {

    /**
     * Same-node forward: dial the target container's bridge IP directly.
-     * Sencho runs in `network_mode: host` so it sees the docker bridge
-     * networks and can reach container IPs without going through any
-     * host-port publish. Looks up the container by Compose's
-     * `<project>-<service>-<index>` naming convention; falls back to a
-     * label-filtered listContainers if the conventional name is absent
-     * (e.g. when the operator overrode the project name).
+     * Sencho joins the `sencho_mesh` Docker bridge network alongside the
+     * meshed user containers, so it can reach their bridge IPs without
+     * going through any host-port publish. Looks up the container by
+     * Compose's `<project>-<service>-<index>` naming convention; falls
+     * back to a label-filtered listContainers if the conventional name
+     * is absent (e.g. when the operator overrode the project name).
     */
    private async openSameNode(target: MeshTarget, src: net.Socket): Promise<void> {
        const ip = await this.resolveContainerIp(target);
@@ -2244,13 +2266,6 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
    private isLocalForwarderActive(): boolean {
        return this.started && this.forwarder.getListenerPorts().length > 0;
    }
-
-    // mintSidecarToken / verifySidecarToken / spawnSidecar / stopSidecar /
-    // isSidecarRunning / handleSidecarResolve / sendSidecar /
-    // attachSidecarSocket are gone: the in-process MeshForwarder replaces
-    // the entire sidecar layer. Routing decisions happen via direct
-    // MeshService calls — no JWT minting, no separate container, no control
-    // WebSocket. See `docs/internal/architecture/mesh.md` for the new flow.
 }

 export type MeshErrorCode =
@@ -91,6 +91,16 @@ interface ReverseLocalTcpStreamState extends StreamMeta {
 interface ReverseRelayTcpStreamState extends StreamMeta {
    kind: 'reverse_relay';
    target: TcpStream;
+    /**
+     * False between `acceptReverseRelay` swapping the reservation for this
+     * state and the target stream emitting `'open'`. `TcpData` frames
+     * arriving in that window are queued in `pendingData` rather than
+     * written into the target stream, whose own remote ack may not yet
+     * have landed.
+     */
+    targetOpen: boolean;
+    pendingData: Buffer[];
+    pendingBytes: number;
    bytesIn: number;
    bytesOut: number;
 }
@@ -639,7 +649,23 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
                } else if (s.kind === 'reverse_relay') {
                    s.bytesIn += frame.payload.length;
                    this.refreshIdleTimer(frame.streamId, s);
-                    s.target.write(frame.payload);
+                    if (s.targetOpen) {
+                        s.target.write(frame.payload);
+                    } else {
+                        // Mirror the reverse_local pending-data path: between
+                        // the state swap in acceptReverseRelay and the target
+                        // stream's 'open' event, buffer up to the cap. Without
+                        // this, a peer that sends body bytes immediately after
+                        // tcp_open_reverse would drop them into a stream that
+                        // is still waiting for its remote ack.
+                        if (s.pendingBytes + frame.payload.length > STREAM_PENDING_DATA_MAX_BYTES) {
+                            this.removeStream(frame.streamId);
+                            this.sendJson({ t: 'tcp_close', s: frame.streamId });
+                            break;
+                        }
+                        s.pendingData.push(Buffer.from(frame.payload));
+                        s.pendingBytes += frame.payload.length;
+                    }
                }
                break;
            }
@@ -825,11 +851,12 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
            if (targetNodeId === localNodeId) {
                await this.acceptReverseLocal(s, { stack, service, port });
            } else {
-                // Relay path does not yet buffer early data (separate
-                // follow-up); drop the local-shaped reservation so the
-                // relay state machine starts from a clean slot. Any frames
-                // buffered up to this point are discarded with it.
-                if (this.streams.get(s) === reservation) this.removeStream(s);
+                // Leave the local-shaped reservation in place so any
+                // TcpData arriving while ensureBridge + openTcpStream
+                // resolve keeps buffering through the reverse_local
+                // branch in handleBinaryFrame. acceptReverseRelay
+                // transplants the buffered frames into the relay state
+                // and flushes them before the open ack.
                await this.acceptReverseRelay(s, targetNodeId, { stack, service, port });
            }
        })().catch((err) => {
@@ -938,25 +965,62 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
    }

    private async acceptReverseRelay(s: number, targetNodeId: number, target: { stack: string; service: string; port: number }): Promise<void> {
+        // Pull the reverse_local-shaped reservation that handleTcpOpenReverse
+        // placed synchronously. Its pendingData buffer holds any TcpData
+        // frames that arrived while we were dispatching, which we transplant
+        // into the relay state so they can be flushed once the target stream
+        // opens.
+        const reservation = this.streams.get(s);
+        if (!reservation || reservation.kind !== 'reverse_local') return;
+
        const { PilotTunnelManager } = await import('./PilotTunnelManager');
        // ensureBridge resolves either an existing pilot tunnel or a fresh
        // proxy-mode tunnel dialed on demand, so proxy <-> proxy relays
        // succeed even when neither remote has a pilot agent.
        const targetBridge = await PilotTunnelManager.getInstance().ensureBridge(targetNodeId);
        if (!targetBridge) {
+            if (this.streams.get(s) === reservation) this.removeStream(s);
            this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' });
            return;
        }
+        // Reservation may have been evicted under the pending-bytes cap
+        // while ensureBridge was resolving. If so, the peer has already
+        // been told via tcp_close to tear down; bail out cleanly.
+        if (this.streams.get(s) !== reservation) return;
+
        const targetStream = targetBridge.openTcpStream(target);
        if (!targetStream) {
+            if (this.streams.get(s) === reservation) this.removeStream(s);
            this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' });
            return;
        }
-        const state: ReverseRelayTcpStreamState = { kind: 'reverse_relay', target: targetStream, bytesIn: 0, bytesOut: 0 };
+        const state: ReverseRelayTcpStreamState = {
+            kind: 'reverse_relay',
+            target: targetStream,
+            targetOpen: false,
+            pendingData: reservation.pendingData,
+            pendingBytes: reservation.pendingBytes,
+            bytesIn: reservation.bytesIn,
+            bytesOut: 0,
+        };
        this.streams.set(s, state);
        this.refreshIdleTimer(s, state);

        targetStream.once('open', () => {
+            const cur = this.streams.get(s);
+            if (!cur || cur.kind !== 'reverse_relay') return;
+            // Flush buffered early data into the target stream BEFORE
+            // acking so the peer sees a clean "ack + data" sequence and
+            // the upstream receives the request body ahead of anything
+            // that arrives post-ack.
+            if (cur.pendingData.length > 0) {
+                for (const buf of cur.pendingData) {
+                    try { cur.target.write(buf); } catch { /* ignore */ }
+                }
+                cur.pendingData = [];
+                cur.pendingBytes = 0;
+            }
+            cur.targetOpen = true;
            this.sendJson({ t: 'tcp_open_ack', s, ok: true });
        });
        targetStream.on('data', (chunk: Buffer) => {
@@ -4,6 +4,7 @@ import type { WebSocketServer } from 'ws';
 import jwt from 'jsonwebtoken';
 import crypto from 'crypto';
 import { DatabaseService, type UserRole } from '../services/DatabaseService';
+import { LicenseService } from '../services/LicenseService';
 import { NodeRegistry } from '../services/NodeRegistry';
 import { COOKIE_NAME } from '../helpers/constants';
 import { handlePilotTunnel } from './pilotTunnel';
@@ -15,6 +16,8 @@ import { handleHostConsoleWs } from './hostConsole';
 import { handleGenericWs, attachGenericConnectionHandlers } from './generic';
 import { rejectUpgrade as reject } from './reject';
 import { looksLikeApiToken, verifyApiTokenChecksum } from '../utils/apiTokenFormat';
+import { PROXY_TIER_HEADER, PROXY_VARIANT_HEADER } from '../services/license-headers';
+import { isLicenseTier, normalizeTier, isLicenseVariant, normalizeVariant } from '../services/license-normalize';

 function parseCookies(req: IncomingMessage): Record<string, string> {
  const header = req.headers.cookie || '';
@@ -138,10 +141,31 @@ export function attachUpgrade(
      // decoded scope is undefined (isProxyToken=false, wsApiTokenScope=null).
      // Restricted api_token scopes (read-only, deploy-only) are blocked
      // earlier by the scope gate above before this branch is reached.
+      //
+      // Admiral entitlement is decided against the *central's* license, not
+      // the receiver's, matching every HTTP mesh route in routes/mesh.ts that
+      // uses `requireAdmiral` / `effectiveTier`. On the node_proxy path the
+      // central forwards `x-sencho-tier` / `x-sencho-variant` and the WS
+      // dispatcher trusts them off the node_proxy credential (same rule as
+      // middleware/auth.ts:117-135 for HTTP). On the full-admin api_token
+      // path no central is asserting tier, so we fall back to the receiver's
+      // own license. Both produce paid+admiral or the upgrade is rejected.
      if (pathname === '/api/mesh/proxy-tunnel') {
        if (!isProxyToken && wsApiTokenScope !== 'full-admin') {
          return reject(socket, 403, 'Forbidden');
        }
+        const license = LicenseService.getInstance();
+        const tunnelTierHeader = req.headers[PROXY_TIER_HEADER] as string | undefined;
+        const tunnelVariantHeader = req.headers[PROXY_VARIANT_HEADER] as string | undefined;
+        const tunnelTier = isProxyToken && isLicenseTier(tunnelTierHeader)
+          ? normalizeTier(tunnelTierHeader)
+          : license.getTier();
+        const tunnelVariant = isProxyToken && tunnelVariantHeader !== undefined && isLicenseVariant(tunnelVariantHeader)
+          ? normalizeVariant(tunnelVariantHeader)
+          : license.getVariant();
+        if (tunnelTier !== 'paid' || tunnelVariant !== 'admiral') {
+          return reject(socket, 403, 'Forbidden');
+        }
        await handleMeshProxyTunnel(req, socket, head);
        return;
      }