feat(ble_log): mirror local compression headers

components/bt/common/ble_log/extension/log_compression/CMakeLists.txt

@@ -185,18 +185,6 @@ if(LOG_COMPRESSED_MODULE)
         endforeach()
         string(REPLACE "|" ";" LOG_COMPRESSED_MODULE_CODE_PATH "${MODULE_CODE_PATH}")
 
-        # Some header files of NIMBLE are not added to include_dirs,
-        # but rely on relative path searches. This will cause the header
-        # files to be found due to the change in the source code location
-        # after using the log compression scheme.
-        # Therefore, these paths are added to include_dirs here to avoid
-        # unfinished compilation errors.
-        if(CONFIG_BT_NIMBLE_ENABLED)
-            list(APPEND include_dirs
-                "host/nimble/nimble/nimble/host/src"
-                "host/nimble/nimble/nimble/host/store/config/src")
-        endif()
-
         if(BLE_COMPRESSED_LIB_LOG_BUILD)
             execute_process(COMMAND ${BLE_PYTHON_EXECUTABLE} ${PYTHON_SCRIPT}
                                 compress

components/bt/common/ble_log/extension/log_compression/scripts/ble_log_compress.py

@@ -20,6 +20,7 @@ This script processes Bluetooth source files to compress logging statements.
 import argparse
 import enum
 import importlib.util
+import json
 import logging
 import os
 import re
@@ -130,6 +131,10 @@ LINE_MACROS = {
     '__LINE__',
 }
 
+MIRRORED_INCLUDE_SUFFIXES = frozenset({'.h', '.inc'})
+MIRRORED_INCLUDE_MANIFEST = '.mirrored_includes.json'
+MIRRORED_INCLUDE_MANIFEST_VERSION = 1
+
 
 class ARG_SIZE_TYPE(enum.IntEnum):
     U32 = 0
@@ -687,6 +692,151 @@ class LogCompressor:
 
         return generated_macros
 
+    @staticmethod
+    def _require_path_within(path: Path, root: Path, label: str) -> None:
+        try:
+            path.relative_to(root)
+        except ValueError as error:
+            raise ValueError(f'{label} escapes allowed root {root}: {path}') from error
+
+    def _validated_mirror_destination(self, relative: Path) -> Path:
+        if relative.is_absolute() or '..' in relative.parts or relative.suffix not in MIRRORED_INCLUDE_SUFFIXES:
+            raise ValueError(f'Invalid mirrored include path: {relative}')
+
+        mirror_root = self.bt_compressed_srcs_path.resolve(strict=True)
+        destination = self.bt_compressed_srcs_path / relative
+        resolved_destination = destination.resolve(strict=False)
+        self._require_path_within(resolved_destination, mirror_root, 'Mirrored include destination')
+        return destination
+
+    def _discover_local_includes(self) -> dict[Path, Path]:
+        try:
+            code_base = self.code_base_path.resolve(strict=True)
+        except (FileNotFoundError, NotADirectoryError) as error:
+            raise ValueError(f'Invalid CODE_BASE_PATH: {self.code_base_path}') from error
+
+        if not code_base.is_dir():
+            raise ValueError(f'Invalid CODE_BASE_PATH: {code_base}')
+
+        includes: dict[Path, Path] = {}
+        for info in self.module_info.values():
+            for configured_path in info['code_path']:
+                candidate = Path(configured_path)
+                if not candidate.is_absolute():
+                    candidate = code_base / candidate
+                try:
+                    module_root = candidate.resolve(strict=True)
+                except (FileNotFoundError, NotADirectoryError) as error:
+                    raise ValueError(f'Invalid module code path: {configured_path}') from error
+                if not module_root.is_dir():
+                    raise ValueError(f'Invalid module code path: {configured_path}')
+                try:
+                    module_root.relative_to(code_base)
+                except ValueError as error:
+                    raise ValueError(f'Module code path escapes CODE_BASE_PATH: {module_root}') from error
+
+                for source in module_root.rglob('*'):
+                    if source.suffix not in MIRRORED_INCLUDE_SUFFIXES or not source.is_file():
+                        continue
+                    resolved_source = source.resolve(strict=True)
+                    self._require_path_within(resolved_source, module_root, 'Local include source')
+                    relative = source.relative_to(code_base)
+                    includes[relative] = source
+        return includes
+
+    def _load_mirrored_include_manifest(self) -> set[Path]:
+        manifest_path = self.bt_compressed_srcs_path / MIRRORED_INCLUDE_MANIFEST
+        if manifest_path.is_symlink():
+            raise ValueError(f'Invalid mirrored include manifest: {manifest_path}')
+        if not manifest_path.exists():
+            return set()
+        if not manifest_path.is_file():
+            raise ValueError(f'Invalid mirrored include manifest: {manifest_path}')
+
+        try:
+            payload = json.loads(manifest_path.read_text(encoding='utf-8'))
+        except (OSError, json.JSONDecodeError) as error:
+            raise ValueError(f'Invalid mirrored include manifest: {manifest_path}') from error
+        if (
+            not isinstance(payload, dict)
+            or payload.get('version') != MIRRORED_INCLUDE_MANIFEST_VERSION
+            or not isinstance(payload.get('files'), list)
+        ):
+            raise ValueError(f'Invalid mirrored include manifest: {manifest_path}')
+
+        relative_paths: set[Path] = set()
+        for value in payload['files']:
+            if not isinstance(value, str):
+                raise ValueError(f'Invalid mirrored include manifest entry: {value!r}')
+            relative = Path(value)
+            self._validated_mirror_destination(relative)
+            relative_paths.add(relative)
+        return relative_paths
+
+    def _remove_empty_mirror_parents(self, destination: Path) -> None:
+        mirror_root = self.bt_compressed_srcs_path.resolve(strict=True)
+        parent = destination.parent
+        while parent != mirror_root:
+            try:
+                parent.rmdir()
+            except OSError:
+                break
+            parent = parent.parent
+
+    def _remove_stale_mirrored_includes(self, stale_paths: set[Path]) -> None:
+        for relative in sorted(stale_paths, key=lambda path: len(path.parts), reverse=True):
+            destination = self._validated_mirror_destination(relative)
+            if destination.is_dir():
+                raise ValueError(f'Managed mirrored include is a directory: {destination}')
+            if destination.exists() or destination.is_symlink():
+                destination.unlink()
+                LOGGER.info(f'Removed stale mirrored include: {destination}')
+            self._remove_empty_mirror_parents(destination)
+
+    def _write_mirrored_include_manifest(self, relative_paths: set[Path]) -> None:
+        manifest_path = self.bt_compressed_srcs_path / MIRRORED_INCLUDE_MANIFEST
+        temporary_path = manifest_path.with_name(f'{manifest_path.name}.tmp')
+        if manifest_path.is_symlink() or (manifest_path.exists() and not manifest_path.is_file()):
+            raise ValueError(f'Invalid mirrored include manifest path: {manifest_path}')
+        if temporary_path.is_symlink() or temporary_path.exists():
+            if temporary_path.is_dir():
+                raise ValueError(f'Invalid manifest temporary path: {temporary_path}')
+            temporary_path.unlink()
+
+        payload = {
+            'version': MIRRORED_INCLUDE_MANIFEST_VERSION,
+            'files': sorted(path.as_posix() for path in relative_paths),
+        }
+        try:
+            temporary_path.write_text(f'{json.dumps(payload, indent=2)}\n', encoding='utf-8')
+            os.replace(temporary_path, manifest_path)
+        finally:
+            if temporary_path.exists() or temporary_path.is_symlink():
+                temporary_path.unlink()
+
+    def mirror_local_includes(self) -> None:
+        """Mirror unchanged local headers for generated C source include lookup."""
+        self.bt_compressed_srcs_path.mkdir(parents=True, exist_ok=True)
+        previous_paths = self._load_mirrored_include_manifest()
+        includes = self._discover_local_includes()
+        current_paths = set(includes)
+        newly_created_paths: set[Path] = set()
+        try:
+            for relative, source in sorted(includes.items(), key=lambda item: item[0].as_posix()):
+                destination = self._validated_mirror_destination(relative)
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                if not destination.exists() and not destination.is_symlink():
+                    newly_created_paths.add(relative)
+                shutil.copy2(source, destination)
+                LOGGER.info(f'Mirrored local include: {source} -> {destination}')
+            self._remove_stale_mirrored_includes(previous_paths - current_paths)
+            self._write_mirrored_include_manifest(current_paths)
+        except Exception:
+            # Do not leave newly introduced headers unowned by the preserved
+            # manifest when copying, cleanup, or manifest replacement fails.
+            self._remove_stale_mirrored_includes(newly_created_paths)
+            raise
+
     def prepare_source_files(self, srcs: list[str]) -> None:
         """
         Prepare source files for processing.
@@ -884,6 +1034,9 @@ class LogCompressor:
         config_path = self.build_dir / 'ble_log/module_info.yml'
         self.load_config(str(config_path), modules)
 
+        # Preserve local quoted-include semantics for generated C sources.
+        self.mirror_local_includes()
+
         # Initialize database
         db_path = self.build_dir / self.config.get('db_path', 'log_db')
         db_manager = LogDBManager(

components/bt/common/ble_log/extension/log_compression/tests/README.md

@@ -19,6 +19,7 @@ tests/
 ├── test_macro_generation.py        # Bluedroid and Mesh module macro generators
 ├── test_pipeline_e2e.py            # End-to-end .c -> .h with golden file comparison
 ├── test_incremental.py             # Incremental compression scenarios
+├── test_header_mirroring.py        # Local .h/.inc mirror, manifest safety, pipeline isolation
 ├── update_golden.py                # Script to regenerate golden expected files
 └── fixtures/
     ├── c_sources/                  # Test input C files
@@ -145,6 +146,14 @@ Tests caching, re-runs, and config change behavior.
 | `TestAddNewFile` | New file IDs continue from `max_id + 1` |
 | `TestConfigChange` | Config change triggers `SOURCE_LOG_UPDATE_FULL` |
 
+### test_header_mirroring.py — Local Header Mirroring
+
+Validates that enabled compression-module `.h` and `.inc` files are copied
+unchanged into `.compressed_srcs` with paths relative to `CODE_BASE_PATH`.
+Coverage includes overlapping module roots, modified and deleted headers,
+manifest validation, path-escape rejection, production pipeline wiring, and
+removal of the NimBLE global include-directory workaround.
+
 ## Golden File Workflow
 
 The `fixtures/expected/` directory contains known-good header outputs. End-to-end tests compare generated output against these files (with copyright year normalized).

components/bt/common/ble_log/extension/log_compression/tests/test_header_mirroring.py

@@ -0,0 +1,252 @@
+# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for unchanged local-header mirroring into .compressed_srcs."""
+
+import json
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import test_utils  # noqa: F401 — must be first to set up sys.path
+from ble_log_compress import LogCompressor
+from test_utils import BLUEDROID_SCRIPT
+from test_utils import BLUEDROID_TAGS
+from test_utils import write_yaml_config
+
+
+class _HeaderMirrorTestCase(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp())
+        self.code_base = self.tmp / 'code_base'
+        self.mirror = self.tmp / 'build' / 'ble_log' / '.compressed_srcs'
+        self.code_base.mkdir(parents=True)
+        self.mirror.mkdir(parents=True)
+
+        self.compressor = LogCompressor()
+        self.compressor.code_base_path = self.code_base
+        self.compressor.bt_compressed_srcs_path = self.mirror
+        self.compressor.module_info = {'BLE_HOST': {'code_path': ['module']}}
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def write_source(self, relative: str, content: str) -> Path:
+        path = self.code_base / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content)
+        return path
+
+
+class TestHeaderDiscoveryAndCopy(_HeaderMirrorTestCase):
+    def test_recursively_mirrors_h_and_inc_only(self) -> None:
+        private_h = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n')
+        config_inc = self.write_source('module/shared/config.inc', '#define CONFIG_VALUE 9\n')
+        self.write_source('module/src/ignored.c', 'int ignored;\n')
+        self.write_source('module/CMakeLists.txt', 'set(ignored yes)\n')
+
+        self.compressor.mirror_local_includes()
+
+        mirrored_h = self.mirror / private_h.relative_to(self.code_base)
+        mirrored_inc = self.mirror / config_inc.relative_to(self.code_base)
+        self.assertEqual(mirrored_h.read_bytes(), private_h.read_bytes())
+        self.assertEqual(mirrored_inc.read_bytes(), config_inc.read_bytes())
+        self.assertEqual(mirrored_h.stat().st_mtime_ns, private_h.stat().st_mtime_ns)
+        self.assertFalse((self.mirror / 'module/src/ignored.c').exists())
+        self.assertFalse((self.mirror / 'module/CMakeLists.txt').exists())
+
+        manifest = json.loads((self.mirror / '.mirrored_includes.json').read_text())
+        self.assertEqual(
+            manifest,
+            {
+                'version': 1,
+                'files': ['module/shared/config.inc', 'module/src/private.h'],
+            },
+        )
+
+    def test_overlapping_module_roots_copy_each_header_once(self) -> None:
+        self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n')
+        self.compressor.module_info = {
+            'BLE_HOST': {'code_path': ['module']},
+            'BLE_ISO': {'code_path': ['module/src']},
+        }
+
+        with mock.patch('ble_log_compress.shutil.copy2', wraps=shutil.copy2) as copy2:
+            self.compressor.mirror_local_includes()
+
+        self.assertEqual(copy2.call_count, 1)
+
+    def test_rejects_missing_and_non_directory_module_roots(self) -> None:
+        cases = ('missing', 'not_a_directory')
+        self.write_source('not_a_directory', 'not a directory\n')
+
+        for code_path in cases:
+            with self.subTest(code_path=code_path):
+                self.compressor.module_info = {'BLE_HOST': {'code_path': [code_path]}}
+                with self.assertRaisesRegex(ValueError, 'Invalid module code path'):
+                    self.compressor.mirror_local_includes()
+
+    def test_rejects_module_root_outside_code_base(self) -> None:
+        outside = self.tmp / 'outside'
+        outside.mkdir()
+        link = self.code_base / 'module'
+        try:
+            link.symlink_to(outside, target_is_directory=True)
+        except OSError as error:
+            self.skipTest(f'Directory symlinks unavailable: {error}')
+
+        with self.assertRaisesRegex(ValueError, 'escapes CODE_BASE_PATH'):
+            self.compressor.mirror_local_includes()
+
+
+class TestHeaderMirrorSynchronization(_HeaderMirrorTestCase):
+    def test_modified_header_updates_content_and_timestamp(self) -> None:
+        source = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n')
+        self.compressor.mirror_local_includes()
+        destination = self.mirror / 'module/src/private.h'
+
+        updated_mtime = source.stat().st_mtime_ns + 2_000_000_000
+        source.write_text('#define PRIVATE_VALUE 11\n')
+        os.utime(source, ns=(updated_mtime, updated_mtime))
+        self.compressor.mirror_local_includes()
+
+        self.assertEqual(destination.read_bytes(), source.read_bytes())
+        self.assertEqual(destination.stat().st_mtime_ns, source.stat().st_mtime_ns)
+
+    def test_deleted_header_is_removed_without_touching_unmanaged_outputs(self) -> None:
+        source = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n')
+        self.compressor.mirror_local_includes()
+        generated_c = self.mirror / 'module/src/generated.c'
+        unmanaged_h = self.mirror / 'module/src/unmanaged.h'
+        generated_c.write_text('int generated;\n')
+        unmanaged_h.write_text('#define UNMANAGED 1\n')
+
+        source.unlink()
+        self.compressor.mirror_local_includes()
+
+        self.assertFalse((self.mirror / 'module/src/private.h').exists())
+        self.assertEqual(generated_c.read_text(), 'int generated;\n')
+        self.assertEqual(unmanaged_h.read_text(), '#define UNMANAGED 1\n')
+        manifest = json.loads((self.mirror / '.mirrored_includes.json').read_text())
+        self.assertEqual(manifest, {'version': 1, 'files': []})
+
+    def test_rejects_malformed_manifest_before_copying(self) -> None:
+        source = self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n')
+        (self.mirror / '.mirrored_includes.json').write_text('{not-json')
+
+        with self.assertRaisesRegex(ValueError, 'Invalid mirrored include manifest'):
+            self.compressor.mirror_local_includes()
+
+        self.assertFalse((self.mirror / source.relative_to(self.code_base)).exists())
+
+    def test_rejects_unsafe_manifest_path_without_deleting_outside_file(self) -> None:
+        outside = self.mirror.parent / 'outside.h'
+        outside.write_text('#define OUTSIDE 1\n')
+        manifest = {
+            'version': 1,
+            'files': ['../outside.h'],
+        }
+        (self.mirror / '.mirrored_includes.json').write_text(json.dumps(manifest))
+
+        with self.assertRaisesRegex(ValueError, 'Invalid mirrored include path'):
+            self.compressor.mirror_local_includes()
+
+        self.assertEqual(outside.read_text(), '#define OUTSIDE 1\n')
+
+    def test_manifest_replace_failure_preserves_previous_manifest(self) -> None:
+        self.write_source('module/src/private.h', '#define PRIVATE_VALUE 7\n')
+        self.compressor.mirror_local_includes()
+        manifest_path = self.mirror / '.mirrored_includes.json'
+        previous_manifest = manifest_path.read_bytes()
+        second = self.write_source('module/src/second.h', '#define SECOND_VALUE 2\n')
+
+        with mock.patch('ble_log_compress.os.replace', side_effect=OSError('replace failed')):
+            with self.assertRaisesRegex(OSError, 'replace failed'):
+                self.compressor.mirror_local_includes()
+
+        self.assertEqual(manifest_path.read_bytes(), previous_manifest)
+        second.unlink()
+        self.compressor.mirror_local_includes()
+        self.assertFalse((self.mirror / second.relative_to(self.code_base)).exists())
+
+    def test_rejects_header_symlink_leaving_module_root(self) -> None:
+        outside = self.write_source('outside.h', '#define OUTSIDE 1\n')
+        link = self.code_base / 'module/src/external.h'
+        link.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            link.symlink_to(outside)
+        except OSError as error:
+            self.skipTest(f'File symlinks unavailable: {error}')
+
+        with self.assertRaisesRegex(ValueError, 'Local include source escapes allowed root'):
+            self.compressor.mirror_local_includes()
+
+
+class TestHeaderMirrorPipeline(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp())
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_main_mirrors_same_directory_and_parent_relative_includes(self) -> None:
+        code_base = self.tmp / 'code_base'
+        build_dir = self.tmp / 'build'
+        compressed_srcs = build_dir / 'ble_log' / '.compressed_srcs'
+        source_dir = code_base / 'test_src' / 'src'
+        shared_dir = code_base / 'test_src' / 'shared'
+        source_dir.mkdir(parents=True)
+        shared_dir.mkdir(parents=True)
+        source = source_dir / 'local_include.c'
+        private_h = source_dir / 'private.h'
+        config_inc = shared_dir / 'config.inc'
+        source.write_text(
+            '#include "private.h"\n'
+            '#include "../shared/config.inc"\n'
+            'void local_include(void) { APPL_TRACE_DEBUG("value %d", PRIVATE_VALUE); }\n'
+        )
+        private_h.write_text('#define PRIVATE_VALUE 7\n')
+        config_inc.write_text('#define CONFIG_VALUE 9\n')
+        write_yaml_config(
+            str(self.tmp),
+            tags=BLUEDROID_TAGS,
+            script_path=BLUEDROID_SCRIPT,
+        )
+
+        argv = [
+            'ble_log_compress.py',
+            'compress',
+            '--srcs',
+            'test_src/src/local_include.c',
+            '--code_base_path',
+            str(code_base),
+            '--module',
+            'BLE_HOST',
+            '--build_path',
+            str(build_dir),
+            '--compressed_srcs_path',
+            str(compressed_srcs),
+        ]
+        with mock.patch.object(sys, 'argv', argv):
+            result = LogCompressor().main()
+
+        self.assertEqual(result, 0)
+        generated_c = compressed_srcs / 'test_src/src/local_include.c'
+        mirrored_h = generated_c.parent / 'private.h'
+        mirrored_inc = generated_c.parent / '../shared/config.inc'
+        self.assertTrue(generated_c.exists())
+        self.assertEqual(mirrored_h.read_bytes(), private_h.read_bytes())
+        self.assertEqual(mirrored_inc.resolve().read_bytes(), config_inc.read_bytes())
+
+    def test_cmake_does_not_add_nimble_local_header_directories(self) -> None:
+        cmake_path = Path(__file__).resolve().parent.parent / 'CMakeLists.txt'
+        cmake = cmake_path.read_text()
+        self.assertNotIn('"host/nimble/nimble/nimble/host/src"', cmake)
+        self.assertNotIn('"host/nimble/nimble/nimble/host/store/config/src"', cmake)
+
+
+if __name__ == '__main__':
+    unittest.main()

components/bt/common/ble_log/extension/log_compression/tests/test_utils.py

@@ -141,7 +141,8 @@ class PipelineContext:
         return str(rel_path)
 
     def run_compression(self, src_list: list[str]) -> dict[str, list[tuple[int, str]]]:
-        """Run prepare + compress + header generation. Returns generated macros."""
+        """Run mirror + prepare + compress + header generation. Returns generated macros."""
+        self.compressor.mirror_local_includes()
         self.compressor.prepare_source_files(src_list)
 
         files_to_process = []