diff --git a/pyproject.toml b/pyproject.toml
index ce82353..270de4d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "java-functional-lsp"
-version = "0.7.3"
+version = "0.7.4"
 description = "Java LSP server enforcing functional programming best practices — null safety, immutability, no exceptions"
 readme = "README.md"
 license = { text = "MIT" }
diff --git a/src/java_functional_lsp/__init__.py b/src/java_functional_lsp/__init__.py
index d4fd9aa..f1a7966 100644
--- a/src/java_functional_lsp/__init__.py
+++ b/src/java_functional_lsp/__init__.py
@@ -1,3 +1,3 @@
 """java-functional-lsp: A Java LSP server enforcing functional programming best practices."""
 
-__version__ = "0.7.3"
+__version__ = "0.7.4"
diff --git a/src/java_functional_lsp/proxy.py b/src/java_functional_lsp/proxy.py
index 23a3117..d5967a5 100644
--- a/src/java_functional_lsp/proxy.py
+++ b/src/java_functional_lsp/proxy.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import hashlib
 import json
 import logging
@@ -12,6 +13,7 @@
 import shutil
 import subprocess
 from collections.abc import Callable, Mapping
+from functools import lru_cache
 from pathlib import Path
 from typing import Any
 
@@ -311,6 +313,47 @@ async def read_message(reader: asyncio.StreamReader) -> dict[str, Any] | None:
         return None
 
 
+_BUILD_FILES = ("pom.xml", "build.gradle", "build.gradle.kts")
+_WORKSPACE_DID_CHANGE_FOLDERS = "workspace/didChangeWorkspaceFolders"
+_MAX_QUEUED_NOTIFICATIONS = 200
+
+
+@lru_cache(maxsize=256)
+def _cached_module_root(dir_path: str) -> str | None:
+    """Cached walk up from *dir_path* to find nearest directory with a build file."""
+    current = Path(dir_path)
+    while True:
+        if any((current / bf).is_file() for bf in _BUILD_FILES):
+            return str(current)
+        parent = current.parent
+        if parent == current:
+            return None
+        current = parent
+
+
+def find_module_root(file_path: str) -> str | None:
+    """Walk up from *file_path* to find the nearest directory containing a build file.
+
+    Returns the directory path, or ``None`` if no build file is found before
+    reaching the filesystem root. Results are cached by parent directory.
+    """
+    return _cached_module_root(str(Path(file_path).parent))
+
+
+def _resolve_module_uri(file_uri: str) -> str | None:
+    """Convert a file URI to the URI of its nearest module root, or None."""
+    from pygls.uris import from_fs_path, to_fs_path
+
+    file_path = to_fs_path(file_uri)
+    if not file_path:
+        return None
+    module_root = find_module_root(file_path)
+    if module_root is None:
+        return None
+    module_uri = from_fs_path(module_root)
+    return module_uri or None
+
+
 class JdtlsProxy:
     """Manages a jdtls subprocess and provides async request/notification forwarding."""
 
@@ -324,6 +367,17 @@ def __init__(self, on_diagnostics: Callable[[str, list[Any]], None] | None = Non
         self._on_diagnostics = on_diagnostics
         self._available = False
         self._jdtls_capabilities: dict[str, Any] = {}
+        # Lazy-start state
+        self._start_lock = asyncio.Lock()
+        self._starting = False
+        self._start_failed = False
+        self._jdtls_on_path = False
+        self._lazy_start_fired = False
+        self._queued_notifications: list[tuple[str, Any]] = []
+        self._original_root_uri: str | None = None
+        self._initial_module_uri: str | None = None
+        self._added_module_uris: set[str] = set()
+        self._workspace_expanded = False
 
     @property
     def is_available(self) -> bool:
@@ -339,31 +393,55 @@ def get_cached_diagnostics(self, uri: str) -> list[Any]:
         """Get the latest jdtls diagnostics for a URI."""
         return list(self._diagnostics_cache.get(uri, []))
 
-    async def start(self, init_params: dict[str, Any]) -> bool:
-        """Start jdtls subprocess and initialize it."""
+    def check_available(self) -> bool:
+        """Check if jdtls is on PATH (lightweight, no subprocess started)."""
+        self._jdtls_on_path = shutil.which("jdtls") is not None
+        if not self._jdtls_on_path:
+            logger.warning("jdtls not found on PATH — running in standalone mode (custom rules only)")
+        return self._jdtls_on_path
+
+    async def start(self, init_params: dict[str, Any], *, module_root_uri: str | None = None) -> bool:
+        """Start jdtls subprocess and initialize it.
+
+        If *module_root_uri* is provided, jdtls is scoped to that module for
+        fast startup. The data-directory hash is always based on the original
+        workspace root (from init_params) so the index persists across restarts.
+        """
         jdtls_path = shutil.which("jdtls")
         if not jdtls_path:
-            logger.warning("jdtls not found on PATH — running in standalone mode (custom rules only)")
             return False
 
-        # jdtls requires a -data directory for workspace metadata (index, classpath, build state).
-        # Use ~/.cache/jdtls-data/<hash> so it persists across reboots and LSP restarts.
-        # Fallback order mirrors LSP spec: rootUri → rootPath → cwd.
-        root_uri = init_params.get("rootUri") or init_params.get("rootPath") or str(Path.cwd())
-        workspace_hash = hashlib.sha256(root_uri.encode()).hexdigest()[:12]
+        # Data-dir hash based on original workspace root (stable across module changes).
+        original_root: str = init_params.get("rootUri") or init_params.get("rootPath") or str(Path.cwd())
+        self._original_root_uri = original_root
+        workspace_hash = hashlib.sha256(original_root.encode()).hexdigest()[:12]
         data_dir = Path.home() / ".cache" / "jdtls-data" / workspace_hash
         data_dir.mkdir(parents=True, exist_ok=True)
 
-        # Build a clean environment for jdtls: detect Java 21+ and set JAVA_HOME
-        # explicitly, or strip JAVA_HOME if the inherited value points at an older
-        # Java (e.g. an IDE launched us with a project SDK of Java 8). Without this,
-        # jdtls 1.57+ fails with "jdtls requires at least Java 21" during its
-        # Python launcher's version check.
-        #
-        # build_jdtls_env() issues several blocking subprocess calls (java -version,
-        # /usr/libexec/java_home) to detect a suitable JDK. Run it in a thread pool
-        # so those calls don't block the asyncio event loop — the IDE's LSP handshake
-        # messages would otherwise stall for up to a few seconds during startup.
+        # Deep copy to avoid mutating server._init_params.
+        effective_params = copy.deepcopy(init_params)
+        effective_root_uri = module_root_uri or original_root
+        if module_root_uri:
+            effective_params["rootUri"] = module_root_uri
+            from pygls.uris import to_fs_path
+
+            effective_params["rootPath"] = to_fs_path(module_root_uri)
+            logger.info(
+                "jdtls: scoping to module %s (full root: %s)",
+                _redact_path(module_root_uri),
+                _redact_path(original_root),
+            )
+
+        # Inject workspaceFolders capability for later expansion.
+        caps = effective_params.setdefault("capabilities", {})
+        ws = caps.setdefault("workspace", {})
+        ws["workspaceFolders"] = True
+
+        # Track the initial module as already loaded.
+        self._initial_module_uri = module_root_uri
+        self._added_module_uris.add(effective_root_uri)
+
+        # Build a clean environment for jdtls.
         loop = asyncio.get_running_loop()
         jdtls_env = await loop.run_in_executor(None, build_jdtls_env)
 
@@ -385,14 +463,12 @@ async def start(self, init_params: dict[str, Any]) -> bool:
                 _redact_path(jdtls_env.get("JAVA_HOME")),
             )
 
-            # Start background readers for stdout (JSON-RPC) and stderr (diagnostics/errors)
             assert self._process.stdout is not None
             self._reader_task = asyncio.create_task(self._reader_loop(self._process.stdout))
             if self._process.stderr is not None:
                 self._stderr_task = asyncio.create_task(self._stderr_reader(self._process.stderr))
 
-            # Send initialize request
-            result = await self.send_request("initialize", init_params)
+            result = await self.send_request("initialize", effective_params)
             if result is None:
                 logger.error("jdtls initialize request failed or timed out")
                 await self.stop()
@@ -401,7 +477,6 @@ async def start(self, init_params: dict[str, Any]) -> bool:
             self._jdtls_capabilities = result.get("capabilities", {})
             logger.info("jdtls initialized (capabilities: %s)", list(self._jdtls_capabilities.keys()))
 
-            # Send initialized notification
             await self.send_notification("initialized", {})
             self._available = True
             return True
@@ -410,6 +485,95 @@ async def start(self, init_params: dict[str, Any]) -> bool:
             logger.error("Failed to start jdtls: %s", e)
             return False
 
+    async def ensure_started(self, init_params: dict[str, Any], file_uri: str) -> bool:
+        """Start jdtls lazily, scoped to the module containing *file_uri*.
+
+        Thread-safe: uses asyncio.Lock to prevent double-start from rapid
+        didOpen calls. Sets ``_start_failed`` on failure to prevent retries.
+        """
+        if self._available:
+            return True
+        if self._start_failed or not self._jdtls_on_path:
+            return False
+
+        async with self._start_lock:
+            if self._available:
+                return True
+
+            self._starting = True
+            try:
+                module_uri = _resolve_module_uri(file_uri)
+                started = await self.start(init_params, module_root_uri=module_uri)
+                if not started:
+                    self._start_failed = True
+                    self._queued_notifications.clear()
+                return started
+            finally:
+                self._starting = False
+
+    def queue_notification(self, method: str, params: Any) -> None:
+        """Buffer a notification for replay after jdtls starts.
+
+        Capped at ``_MAX_QUEUED_NOTIFICATIONS`` to prevent unbounded memory
+        growth during long jdtls startup. Oldest entries are dropped on overflow.
+        """
+        if len(self._queued_notifications) >= _MAX_QUEUED_NOTIFICATIONS:
+            self._queued_notifications.pop(0)
+        self._queued_notifications.append((method, params))
+
+    async def flush_queued_notifications(self) -> None:
+        """Send all queued notifications to jdtls."""
+        queue, self._queued_notifications = self._queued_notifications, []
+        for method, params in queue:
+            await self.send_notification(method, params)
+
+    async def add_module_if_new(self, file_uri: str) -> None:
+        """Add the module containing *file_uri* to jdtls if not already added."""
+        if not self._available:
+            return
+        module_uri = _resolve_module_uri(file_uri)
+        if module_uri is None or module_uri in self._added_module_uris:
+            return
+        self._added_module_uris.add(module_uri)
+        from pygls.uris import to_fs_path
+
+        logger.info("jdtls: adding module %s", _redact_path(to_fs_path(module_uri)))
+        mod_name = Path(to_fs_path(module_uri) or module_uri).name
+        await self.send_notification(
+            _WORKSPACE_DID_CHANGE_FOLDERS,
+            {"event": {"added": [{"uri": module_uri, "name": mod_name}], "removed": []}},
+        )
+
+    async def expand_full_workspace(self) -> None:
+        """Expand jdtls workspace to the full monorepo root (background task).
+
+        Removes the initial module-scoped folder and adds the full root to
+        avoid double-indexing.
+        """
+        if self._workspace_expanded or not self._available or not self._original_root_uri:
+            return
+        from pygls.uris import from_fs_path, to_fs_path
+
+        root_path = to_fs_path(self._original_root_uri) or self._original_root_uri
+        root_uri = from_fs_path(root_path) or self._original_root_uri
+        if root_uri in self._added_module_uris:
+            self._workspace_expanded = True
+            return
+        self._added_module_uris.add(root_uri)
+
+        # Remove initial module folder to avoid double-indexing.
+        removed: list[dict[str, str]] = []
+        if self._initial_module_uri and self._initial_module_uri != root_uri:
+            ini_path = to_fs_path(self._initial_module_uri) or self._initial_module_uri
+            removed.append({"uri": self._initial_module_uri, "name": Path(ini_path).name})
+
+        logger.info("jdtls: expanding to full workspace %s", _redact_path(root_path))
+        await self.send_notification(
+            _WORKSPACE_DID_CHANGE_FOLDERS,
+            {"event": {"added": [{"uri": root_uri, "name": Path(root_path).name}], "removed": removed}},
+        )
+        self._workspace_expanded = True
+
     async def stop(self) -> None:
         """Shutdown jdtls subprocess gracefully."""
         self._available = False
diff --git a/src/java_functional_lsp/server.py b/src/java_functional_lsp/server.py
index 648535b..030d3c0 100644
--- a/src/java_functional_lsp/server.py
+++ b/src/java_functional_lsp/server.py
@@ -77,6 +77,8 @@ def _on_jdtls_diagnostics(self, uri: str, diagnostics: list[Any]) -> None:
 
 # Debounce state for didChange events (only affects human typing in IDEs, not agents)
 _pending: dict[str, asyncio.Task[None]] = {}
+# Background tasks (prevent GC of fire-and-forget tasks)
+_bg_tasks: set[asyncio.Task[None]] = set()
 _DEBOUNCE_SECONDS = 0.15
 
 
@@ -242,17 +244,15 @@ def on_initialize(params: lsp.InitializeParams) -> lsp.InitializeResult:
 
 @server.feature(lsp.INITIALIZED)
 async def on_initialized(params: lsp.InitializedParams) -> None:
-    """Start jdtls proxy after initialization."""
+    """Check jdtls availability; actual start deferred to first didOpen."""
     logger.info(
         "java-functional-lsp initialized (rules: %s)",
         list(server._config.get("rules", {}).keys()) or "all defaults",
     )
-    started = await server._proxy.start(server._init_params)
-    if started:
-        logger.info("jdtls proxy active — full Java language support enabled")
-        await _register_jdtls_capabilities()
+    if server._proxy.check_available():
+        logger.info("jdtls found on PATH — will start lazily on first file open")
     else:
-        logger.info("jdtls proxy unavailable — running with custom rules only")
+        logger.info("jdtls not on PATH — running with custom rules only")
 
 
 _JAVA_SELECTOR = [lsp.TextDocumentFilterLanguage(language="java")]
@@ -331,20 +331,50 @@ async def _deferred_validate(uri: str) -> None:
     _analyze_and_publish(uri)
 
 
+def _forward_or_queue(method: str, serialized: Any) -> None:
+    """Forward a notification to jdtls if available, or queue it if starting."""
+    if server._proxy.is_available:
+        task = asyncio.create_task(server._proxy.send_notification(method, serialized))
+        _bg_tasks.add(task)
+        task.add_done_callback(_bg_tasks.discard)
+    elif server._proxy._starting:
+        server._proxy.queue_notification(method, serialized)
+
+
 @server.feature(lsp.TEXT_DOCUMENT_DID_OPEN)
 async def on_did_open(params: lsp.DidOpenTextDocumentParams) -> None:
-    """Forward to jdtls and analyze immediately."""
+    """Forward to jdtls (starting lazily if needed) and analyze immediately.
+
+    Custom diagnostics always publish immediately regardless of jdtls state.
+    jdtls startup is non-blocking — it runs in the background so the first
+    didOpen response isn't delayed by jdtls cold-start.
+    """
+    uri = params.text_document.uri
+    serialized = _serialize_params(params)
+
     if server._proxy.is_available:
-        await server._proxy.send_notification("textDocument/didOpen", _serialize_params(params))
-    _analyze_and_publish(params.text_document.uri)
+        # Fast path: jdtls running. Forward didOpen + add module if new.
+        await server._proxy.send_notification("textDocument/didOpen", serialized)
+        await server._proxy.add_module_if_new(uri)
+    elif server._proxy._jdtls_on_path and not server._proxy._start_failed:
+        # Queue the didOpen (whether this is the first file or a subsequent one during startup).
+        server._proxy.queue_notification("textDocument/didOpen", serialized)
+        if not server._proxy._lazy_start_fired:
+            # First file: kick off lazy start in background.
+            server._proxy._lazy_start_fired = True
+            task = asyncio.create_task(_lazy_start_jdtls(uri))
+            _bg_tasks.add(task)
+            task.add_done_callback(_bg_tasks.discard)
+
+    # Custom diagnostics always publish immediately — never blocked by jdtls.
+    _analyze_and_publish(uri)
 
 
 @server.feature(lsp.TEXT_DOCUMENT_DID_CHANGE)
 async def on_did_change(params: lsp.DidChangeTextDocumentParams) -> None:
     """Forward to jdtls and schedule debounced re-analysis."""
     uri = params.text_document.uri
-    if server._proxy.is_available:
-        await server._proxy.send_notification("textDocument/didChange", _serialize_params(params))
+    _forward_or_queue("textDocument/didChange", _serialize_params(params))
     # Cancel pending validation, schedule new one (150ms debounce for IDE typing)
     if uri in _pending:
         _pending[uri].cancel()
@@ -354,8 +384,7 @@ async def on_did_change(params: lsp.DidChangeTextDocumentParams) -> None:
 @server.feature(lsp.TEXT_DOCUMENT_DID_SAVE)
 async def on_did_save(params: lsp.DidSaveTextDocumentParams) -> None:
     """Forward to jdtls and re-analyze immediately (no debounce on save)."""
-    if server._proxy.is_available:
-        await server._proxy.send_notification("textDocument/didSave", _serialize_params(params))
+    _forward_or_queue("textDocument/didSave", _serialize_params(params))
     _analyze_and_publish(params.text_document.uri)
 
 
@@ -368,8 +397,39 @@ async def on_did_close(params: lsp.DidCloseTextDocumentParams) -> None:
         del _pending[uri]
     # Clear diagnostics for the closed document (LSP best practice)
     server.text_document_publish_diagnostics(lsp.PublishDiagnosticsParams(uri=uri, diagnostics=[]))
-    if server._proxy.is_available:
-        await server._proxy.send_notification("textDocument/didClose", _serialize_params(params))
+    _forward_or_queue("textDocument/didClose", _serialize_params(params))
+
+
+async def _lazy_start_jdtls(file_uri: str) -> None:
+    """Background task: start jdtls scoped to the module containing *file_uri*.
+
+    Runs in the background so ``on_did_open`` returns immediately with custom
+    diagnostics. After jdtls initializes, registers capabilities, flushes
+    queued notifications, and schedules workspace expansion.
+    """
+    try:
+        started = await server._proxy.ensure_started(server._init_params, file_uri)
+        if started:
+            logger.info("jdtls proxy active — full Java language support enabled")
+            await _register_jdtls_capabilities()
+            await server._proxy.flush_queued_notifications()
+            await _expand_workspace_background()
+    except Exception:
+        logger.warning("jdtls lazy start failed", exc_info=True)
+
+
+async def _expand_workspace_background() -> None:
+    """Background task: expand jdtls workspace to full monorepo root.
+
+    Runs after jdtls finishes initializing with the first module scope.
+    The user's actively-opened modules are loaded immediately via
+    ``add_module_if_new()`` in ``on_did_open``; this adds the full root
+    so cross-module references for unopened files also work.
+    """
+    try:
+        await server._proxy.expand_full_workspace()
+    except Exception:
+        logger.warning("Failed to expand jdtls workspace", exc_info=True)
 
 
 # --- jdtls passthrough handlers (registered dynamically, NOT at module level) ---
diff --git a/tests/test_proxy.py b/tests/test_proxy.py
index 446245b..af75ab4 100644
--- a/tests/test_proxy.py
+++ b/tests/test_proxy.py
@@ -857,3 +857,265 @@ async def fake_send_request(*_args: Any, **_kwargs: Any) -> None:
         assert ok is False
         # The crucial assertion: env= was passed through to the subprocess call.
         assert captured["env"] == sentinel_env
+
+
+class TestFindModuleRoot:
+    """Tests for find_module_root — build-file detection for module scoping."""
+
+    def test_finds_pom_xml(self, tmp_path: Any) -> None:
+        from java_functional_lsp.proxy import find_module_root
+
+        (tmp_path / "pom.xml").touch()
+        java_file = tmp_path / "src" / "Main.java"
+        java_file.parent.mkdir()
+        java_file.touch()
+        assert find_module_root(str(java_file)) == str(tmp_path)
+
+    def test_finds_build_gradle(self, tmp_path: Any) -> None:
+        from java_functional_lsp.proxy import find_module_root
+
+        (tmp_path / "build.gradle").touch()
+        java_file = tmp_path / "src" / "Main.java"
+        java_file.parent.mkdir()
+        java_file.touch()
+        assert find_module_root(str(java_file)) == str(tmp_path)
+
+    def test_finds_build_gradle_kts(self, tmp_path: Any) -> None:
+        from java_functional_lsp.proxy import find_module_root
+
+        (tmp_path / "build.gradle.kts").touch()
+        java_file = tmp_path / "src" / "Main.java"
+        java_file.parent.mkdir()
+        java_file.touch()
+        assert find_module_root(str(java_file)) == str(tmp_path)
+
+    def test_finds_nearest_not_parent(self, tmp_path: Any) -> None:
+        """Nested modules: should find the innermost module root."""
+        from java_functional_lsp.proxy import find_module_root
+
+        (tmp_path / "pom.xml").touch()  # parent module
+        child = tmp_path / "child-module"
+        child.mkdir()
+        (child / "pom.xml").touch()  # child module
+        java_file = child / "src" / "Main.java"
+        java_file.parent.mkdir()
+        java_file.touch()
+        assert find_module_root(str(java_file)) == str(child)
+
+    def test_returns_none_when_no_build_file(self, tmp_path: Any) -> None:
+        from java_functional_lsp.proxy import find_module_root
+
+        java_file = tmp_path / "src" / "Main.java"
+        java_file.parent.mkdir()
+        java_file.touch()
+        assert find_module_root(str(java_file)) is None
+
+
+class TestLazyStart:
+    """Tests for lazy-start proxy features."""
+
+    def test_check_available_true(self) -> None:
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        with patch("java_functional_lsp.proxy.shutil.which", return_value="/usr/bin/jdtls"):
+            assert proxy.check_available() is True
+        assert proxy._jdtls_on_path is True
+
+    def test_check_available_false(self) -> None:
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        with patch("java_functional_lsp.proxy.shutil.which", return_value=None):
+            assert proxy.check_available() is False
+        assert proxy._jdtls_on_path is False
+
+    async def test_queue_and_flush(self) -> None:
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy.queue_notification("textDocument/didOpen", {"uri": "a"})
+        proxy.queue_notification("textDocument/didChange", {"uri": "b"})
+        assert len(proxy._queued_notifications) == 2
+
+        flushed: list[tuple[str, Any]] = []
+
+        async def mock_send(method: str, params: Any) -> None:
+            flushed.append((method, params))
+
+        proxy.send_notification = mock_send  # type: ignore[assignment]
+        await proxy.flush_queued_notifications()
+        assert len(flushed) == 2
+        assert flushed[0] == ("textDocument/didOpen", {"uri": "a"})
+        assert flushed[1] == ("textDocument/didChange", {"uri": "b"})
+        assert len(proxy._queued_notifications) == 0
+
+    def test_queue_caps_at_max(self) -> None:
+        from java_functional_lsp.proxy import _MAX_QUEUED_NOTIFICATIONS, JdtlsProxy
+
+        proxy = JdtlsProxy()
+        for i in range(_MAX_QUEUED_NOTIFICATIONS + 50):
+            proxy.queue_notification("textDocument/didChange", {"i": i})
+        assert len(proxy._queued_notifications) == _MAX_QUEUED_NOTIFICATIONS
+        # Oldest entries dropped — last entry should be the most recent
+        assert proxy._queued_notifications[-1] == ("textDocument/didChange", {"i": _MAX_QUEUED_NOTIFICATIONS + 49})
+
+    async def test_ensure_started_no_retry_after_failure(self) -> None:
+        from unittest.mock import AsyncMock
+
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._jdtls_on_path = True
+        proxy.queue_notification("textDocument/didOpen", {"uri": "test"})
+        proxy.start = AsyncMock(return_value=False)  # type: ignore[assignment]
+        result = await proxy.ensure_started({"rootUri": "file:///tmp"}, "file:///tmp/F.java")
+        assert result is False
+        assert proxy._start_failed is True
+        # Queue should be cleared on failure
+        assert len(proxy._queued_notifications) == 0
+        # Second call should return immediately without calling start()
+        proxy.start.reset_mock()  # type: ignore[attr-defined]
+        result2 = await proxy.ensure_started({"rootUri": "file:///tmp"}, "file:///tmp/F.java")
+        assert result2 is False
+        proxy.start.assert_not_called()  # type: ignore[attr-defined]
+
+    async def test_add_module_if_new_sends_notification(self) -> None:
+        from unittest.mock import AsyncMock
+
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._available = True
+        proxy.send_notification = AsyncMock()  # type: ignore[assignment]
+        # Create a tmp dir with pom.xml
+        import tempfile
+        from pathlib import Path
+
+        with tempfile.TemporaryDirectory() as td:
+            (Path(td) / "pom.xml").touch()
+            java_file = Path(td) / "src" / "Main.java"
+            java_file.parent.mkdir()
+            java_file.touch()
+            uri = java_file.as_uri()
+            await proxy.add_module_if_new(uri)
+            proxy.send_notification.assert_called_once()  # type: ignore[attr-defined]
+            call_args = proxy.send_notification.call_args  # type: ignore[attr-defined]
+            assert call_args[0][0] == "workspace/didChangeWorkspaceFolders"
+
+    async def test_add_module_if_new_skips_duplicate(self) -> None:
+        from unittest.mock import AsyncMock
+
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._available = True
+        proxy.send_notification = AsyncMock()  # type: ignore[assignment]
+        import tempfile
+        from pathlib import Path
+
+        with tempfile.TemporaryDirectory() as td:
+            (Path(td) / "pom.xml").touch()
+            java_file = Path(td) / "src" / "Main.java"
+            java_file.parent.mkdir()
+            java_file.touch()
+            uri = java_file.as_uri()
+            await proxy.add_module_if_new(uri)
+            await proxy.add_module_if_new(uri)  # duplicate
+            assert proxy.send_notification.call_count == 1  # type: ignore[attr-defined]
+
+    async def test_expand_full_workspace_sends_notification(self) -> None:
+        from unittest.mock import AsyncMock
+
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._available = True
+        proxy._original_root_uri = "file:///workspace/monorepo"
+        proxy.send_notification = AsyncMock()  # type: ignore[assignment]
+        await proxy.expand_full_workspace()
+        proxy.send_notification.assert_called_once()  # type: ignore[attr-defined]
+        assert proxy._workspace_expanded is True
+
+    async def test_expand_full_workspace_noop_when_not_available(self) -> None:
+        from unittest.mock import AsyncMock
+
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._original_root_uri = "file:///workspace/monorepo"
+        proxy.send_notification = AsyncMock()  # type: ignore[assignment]
+        await proxy.expand_full_workspace()
+        proxy.send_notification.assert_not_called()  # type: ignore[attr-defined]
+        assert proxy._workspace_expanded is False
+
+    async def test_expand_full_workspace_noop_when_already_added(self) -> None:
+        from unittest.mock import AsyncMock
+
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._available = True
+        proxy._original_root_uri = "file:///workspace/monorepo"
+        proxy._added_module_uris.add("file:///workspace/monorepo")
+        proxy.send_notification = AsyncMock()  # type: ignore[assignment]
+        await proxy.expand_full_workspace()
+        proxy.send_notification.assert_not_called()  # type: ignore[attr-defined]
+        assert proxy._workspace_expanded is True
+
+    async def test_ensure_started_no_build_file(self) -> None:
+        """ensure_started with no build file should pass module_root_uri=None."""
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._jdtls_on_path = True
+        captured: dict[str, Any] = {}
+
+        async def capturing_start(params: Any, *, module_root_uri: str | None = None) -> bool:
+            captured["module_root_uri"] = module_root_uri
+            return False
+
+        proxy.start = capturing_start  # type: ignore[assignment]
+        await proxy.ensure_started(
+            {"rootUri": "file:///monorepo", "capabilities": {}},
+            "file:///nonexistent/src/Main.java",
+        )
+        assert captured["module_root_uri"] is None
+
+    async def test_ensure_started_with_build_file(self, tmp_path: Any) -> None:
+        """ensure_started should find module root and pass it to start()."""
+        from java_functional_lsp.proxy import JdtlsProxy
+
+        proxy = JdtlsProxy()
+        proxy._jdtls_on_path = True
+        (tmp_path / "pom.xml").touch()
+        java_file = tmp_path / "src" / "Main.java"
+        java_file.parent.mkdir()
+        java_file.touch()
+
+        captured: dict[str, Any] = {}
+
+        async def capturing_start(params: Any, *, module_root_uri: str | None = None) -> bool:
+            captured["module_root_uri"] = module_root_uri
+            return False
+
+        proxy.start = capturing_start  # type: ignore[assignment]
+        await proxy.ensure_started(
+            {"rootUri": "file:///monorepo", "capabilities": {}},
+            java_file.as_uri(),
+        )
+        assert captured["module_root_uri"] is not None
+        assert str(tmp_path) in captured["module_root_uri"]
+
+    def test_data_dir_hash_uses_original_root(self) -> None:
+        """Data-dir hash should be based on original rootUri, not module root."""
+        import hashlib
+
+        # The hash is computed from the original rootUri, not the module root.
+        # Verify these produce different hashes, confirming start() must use
+        # the original root for stability.
+        root = "file:///workspace/monorepo"
+        expected_hash = hashlib.sha256(root.encode()).hexdigest()[:12]
+        module_root = "file:///workspace/monorepo/module-a"
+        module_hash = hashlib.sha256(module_root.encode()).hexdigest()[:12]
+        assert expected_hash != module_hash
diff --git a/tests/test_server.py b/tests/test_server.py
index 9d2f441..0b5dc5a 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -414,6 +414,68 @@ async def test_register_jdtls_capabilities_idempotent(self) -> None:
             srv_mod._jdtls_capabilities_registered = old_flag
         mock_reg.assert_not_called()
 
+    async def test_lazy_start_jdtls_success(self, caplog: Any) -> None:
+        """_lazy_start_jdtls logs success, flushes queue, and expands workspace."""
+        import logging
+        from unittest.mock import AsyncMock, MagicMock, patch
+
+        import java_functional_lsp.server as srv_mod
+        from java_functional_lsp.server import _lazy_start_jdtls
+        from java_functional_lsp.server import server as srv
+
+        mock_flush = AsyncMock()
+        mock_expand = AsyncMock()
+        old_flag = srv_mod._jdtls_capabilities_registered
+        srv_mod._jdtls_capabilities_registered = False
+        try:
+            with (
+                caplog.at_level(logging.INFO, logger="java_functional_lsp.server"),
+                patch.object(srv._proxy, "ensure_started", AsyncMock(return_value=True)),
+                patch.object(srv._proxy, "flush_queued_notifications", mock_flush),
+                patch.object(srv._proxy, "expand_full_workspace", mock_expand),
+                patch.object(srv, "feature", MagicMock(return_value=lambda fn: fn)),
+                patch.object(srv, "client_register_capability_async", AsyncMock()),
+            ):
+                await _lazy_start_jdtls("file:///test/F.java")
+        finally:
+            srv_mod._jdtls_capabilities_registered = old_flag
+        assert any("jdtls proxy active" in r.getMessage() for r in caplog.records)
+        mock_flush.assert_called_once()
+        mock_expand.assert_called_once()
+
+    async def test_lazy_start_jdtls_failure_logged(self, caplog: Any) -> None:
+        """_lazy_start_jdtls logs warning on exception."""
+        import logging
+        from unittest.mock import AsyncMock, patch
+
+        from java_functional_lsp.server import _lazy_start_jdtls
+        from java_functional_lsp.server import server as srv
+
+        with (
+            caplog.at_level(logging.WARNING, logger="java_functional_lsp.server"),
+            patch.object(srv._proxy, "ensure_started", AsyncMock(side_effect=Exception("boom"))),
+        ):
+            await _lazy_start_jdtls("file:///test/F.java")
+        assert any("lazy start failed" in r.getMessage() for r in caplog.records)
+
+    async def test_lazy_start_jdtls_silent_failure(self) -> None:
+        """When ensure_started returns False, flush/expand are not called."""
+        from unittest.mock import AsyncMock, patch
+
+        from java_functional_lsp.server import _lazy_start_jdtls
+        from java_functional_lsp.server import server as srv
+
+        mock_flush = AsyncMock()
+        mock_expand = AsyncMock()
+        with (
+            patch.object(srv._proxy, "ensure_started", AsyncMock(return_value=False)),
+            patch.object(srv._proxy, "flush_queued_notifications", mock_flush),
+            patch.object(srv._proxy, "expand_full_workspace", mock_expand),
+        ):
+            await _lazy_start_jdtls("file:///test/F.java")
+        mock_flush.assert_not_called()
+        mock_expand.assert_not_called()
+
     def test_serialize_params_camelcase(self) -> None:
         from java_functional_lsp.server import _serialize_params
 
diff --git a/uv.lock b/uv.lock
index 3919fb8..650e5c7 100644
--- a/uv.lock
+++ b/uv.lock
@@ -184,7 +184,7 @@ wheels = [
 
 [[package]]
 name = "java-functional-lsp"
-version = "0.7.3"
+version = "0.7.4"
 source = { editable = "." }
 dependencies = [
     { name = "pygls" },