diff --git a/pkg-py/src/commons/_measures.py b/pkg-py/src/commons/_measures.py
index 6554986..6e2c14d 100644
--- a/pkg-py/src/commons/_measures.py
+++ b/pkg-py/src/commons/_measures.py
@@ -11,9 +11,17 @@
 
 from __future__ import annotations
 
+import hashlib
+import importlib.util
 import inspect
+import os
+import re
+import sys
+import threading
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass
+from pathlib import Path
+from types import MappingProxyType, ModuleType
 from typing import (
     Annotated,
     Any,
@@ -249,6 +257,18 @@ def decorate(func: Callable[..., Any]) -> Callable[..., Any]:
     return decorate
 
 
+def as_measure(obj: Any) -> Measure | None:
+    """Recognize a measure, whether decorated function or bare record."""
+    if isinstance(obj, Measure):
+        return obj
+    record = getattr(obj, MEASURE_ATTRIBUTE, None)
+    return record if isinstance(record, Measure) else None
+
+
+def _humanize(name: str) -> str:
+    return name.replace("_", " ")
+
+
 def measure_schema_text(
     record: Measure,
     source_names: Sequence[str] = (),
@@ -338,13 +358,343 @@ def _resolve_ref(node: dict[str, Any], defs: dict[str, Any]) -> dict[str, Any]:
     return defs[ref.removeprefix("#/$defs/")]
 
 
-def as_measure(obj: Any) -> Measure | None:
-    """Recognize a measure, whether decorated function or bare record."""
-    if isinstance(obj, Measure):
-        return obj
-    record = getattr(obj, MEASURE_ATTRIBUTE, None)
-    return record if isinstance(record, Measure) else None
+@dataclass(frozen=True)
+class SemanticLayer:
+    """The trusted calculations an agent can run.
 
+    ``source_text`` holds the source of the measures and the module-level
+    helpers they call, keyed by Python name. Only text is kept: the agent's
+    worker session reads measure definitions but never receives a callable.
+    """
 
-def _humanize(name: str) -> str:
-    return name.replace("_", " ")
+    measures: Mapping[str, Measure]
+    source_text: Mapping[str, str]
+
+    def __len__(self) -> int:
+        return len(self.measures)
+
+    def __repr__(self) -> str:
+        count = len(self.measures)
+        plural = "" if count == 1 else "s"
+        return f"A commons semantic layer with {count} measure{plural}."
+
+
+def semantic_layer(*items: Any) -> SemanticLayer:
+    """Collect measures into a semantic layer.
+
+    Each item is a measure, a list of measures, a module, or a path to a
+    Python file or a directory of them. Directory searches are not recursive.
+
+    A sibling file is imported by plain absolute import; its directory is on
+    sys.path only while the file loads. A file whose name collides with the
+    standard library, or with a module already imported from elsewhere, is a
+    construction error.
+    """
+    measures: dict[str, Measure] = {}
+    source_text: dict[str, str] = {}
+    duplicates: list[str] = []
+
+    for item in items:
+        found, sources = _collect(item)
+        for record in found:
+            if record.name in measures:
+                duplicates.append(record.name)
+            measures[record.name] = record
+        _merge_sources(source_text, sources)
+
+    if duplicates:
+        raise ValueError(
+            f"Measure names must be unique; duplicated: "
+            f"{', '.join(sorted(set(duplicates)))}.\n"
+            f"Give one of the colliding measures a distinct name with "
+            f"@measure(name=...)."
+        )
+
+    return SemanticLayer(
+        measures=MappingProxyType(measures),
+        source_text=MappingProxyType(source_text),
+    )
+
+
+def _collect(item: Any) -> tuple[list[Measure], dict[str, str]]:
+    if isinstance(item, (list, tuple)):
+        measures: list[Measure] = []
+        sources: dict[str, str] = {}
+        for entry in item:
+            found, text = _collect(entry)
+            measures.extend(found)
+            _merge_sources(sources, text)
+        return measures, sources
+
+    if isinstance(item, ModuleType):
+        return _from_module(item)
+
+    if isinstance(item, (str, os.PathLike)):
+        return _from_path(Path(item))
+
+    record = as_measure(item)
+    if record is None:
+        raise TypeError(
+            f"Every item in semantic_layer() must be a measure, a list of "
+            f"measures, a module, or a path; got {item!r}.\n"
+            f"Decorate the function with @measure to make it one."
+        )
+    return [record], {record.func.__name__: _source_text(record.func)}
+
+
+def _from_path(path: Path) -> tuple[list[Measure], dict[str, str]]:
+    if not path.exists():
+        raise ValueError(
+            f"Path does not exist: {path}.\n"
+            f"semantic_layer() takes measures, modules, Python files, or "
+            f"directories of them."
+        )
+
+    # Not recursive, and __init__.py is skipped: a directory of measure files
+    # is a directory, not a package.
+    files = (
+        sorted(
+            entry
+            for entry in path.iterdir()
+            if entry.suffix == ".py" and entry.name != "__init__.py"
+        )
+        if path.is_dir()
+        else [path]
+    )
+
+    measures: list[Measure] = []
+    sources: dict[str, str] = {}
+    for file in files:
+        found, text = _from_module(_load_module_from_path(file))
+        measures.extend(found)
+        _merge_sources(sources, text)
+    return measures, sources
+
+
+def _merge_sources(target: dict[str, str], found: Mapping[str, str]) -> None:
+    """Merge harvested source text; the first definition of a name wins.
+
+    Every place source text is combined across items uses this, so a new
+    merge point cannot quietly pick the wrong precedence.
+    """
+    for name, text in found.items():
+        target.setdefault(name, text)
+
+
+def _from_module(module: ModuleType) -> tuple[list[Measure], dict[str, str]]:
+    """Harvest a module's measures and the source of every function it defines.
+
+    Helpers are harvested too, so the worker session can show the reasoning a
+    measure delegates to. Imported names are skipped: they belong to the
+    module they were defined in.
+    """
+    measures: list[Measure] = []
+    sources: dict[str, str] = {}
+    for name, value in vars(module).items():
+        if not inspect.isfunction(value) or value.__module__ != module.__name__:
+            continue
+        sources[name] = _source_text(value)
+        record = as_measure(value)
+        if record is not None:
+            measures.append(record)
+    return measures, sources
+
+
+# The import machinery (sys.path, sys.modules) is process-global state, not
+# owned by any one SemanticLayer, so concurrent construction must serialize
+# around it rather than around the layer itself. Reentrant, not a plain
+# Lock: the lock is held across exec_module(), which runs a measure file's
+# top-level code, and that code can itself call semantic_layer() on another
+# path, re-entering this same function on the same thread.
+_IMPORT_LOCK = threading.RLock()
+
+# Anything that is not a plain identifier character, including the dots in a
+# name like `sales.q3.py`: left alone, a dotted stem would still produce a
+# dotted module name, defeating the single-segment name below.
+_UNSAFE_NAME_CHARS = re.compile(r"[^0-9a-zA-Z_]")
+
+# The mtime, in nanoseconds, a path's cached module was loaded at, keyed by
+# module name. st_mtime_ns, not st_mtime: a float mtime can lose enough
+# filesystem timestamp precision that two rapid edits look identical and a
+# stale module stays cached. Compared against the file's current mtime on
+# every load so a module is reused only while its source is unchanged;
+# sys.modules alone cannot tell a fresh load from a stale one.
+_load_mtimes: dict[str, int] = {}
+
+
+def _load_module_from_path(path: Path) -> ModuleType:
+    resolved = path.resolve()
+    stem = _UNSAFE_NAME_CHARS.sub("_", path.stem)
+    # The digest, not a load counter: two semantic_layer() calls on the same
+    # path should reuse the same module when its source is unchanged, rather
+    # than each minting a new sys.modules entry the old one is never removed
+    # from -- an application constructing an agent per session leaked one
+    # module, and everything it held onto, per session. The cost is that a
+    # file edited mid-process reloads under the same name, so an earlier
+    # layer's Measure.func.__module__ then resolves to the newer module
+    # object; a development-time scenario, not a session-count-scaling leak.
+    #
+    # A single path segment, not `commons._measure_sources.<stem>`: a dotted
+    # name makes `commons._measure_sources` the loaded file's __package__,
+    # so a relative import in the user's own file would resolve into
+    # commons' internals instead of failing with Python's own "no known
+    # parent package" error.
+    digest = hashlib.sha256(str(resolved).encode()).hexdigest()[:8]
+    name = f"_commons_measure_source_{stem}_{digest}"
+
+    with _IMPORT_LOCK:
+        mtime_ns = resolved.stat().st_mtime_ns
+        cached = sys.modules.get(name)
+        if (
+            cached is not None
+            and _load_mtimes.get(name) == mtime_ns
+            and _cached_module_path(cached) == resolved
+        ):
+            return cached
+
+        # Unconditional, not just on a detected reload: an earlier process
+        # can have already written this file's .pyc, and a same-second,
+        # same-size edit since then leaves it looking valid to
+        # SourceFileLoader on this process's first load too, which has no
+        # _load_mtimes entry to have noticed the edit itself. See
+        # _invalidate_bytecode_cache for why this step is required at all.
+        _invalidate_bytecode_cache(path)
+
+        _check_directory_importable(path.parent, requested=path)
+
+        spec = importlib.util.spec_from_file_location(name, path)
+        if spec is None or spec.loader is None:
+            raise ValueError(
+                f"Cannot read measures from {path}: not a Python file.\n"
+                f"Pass a .py file, a directory of them, or a module object."
+            )
+
+        # A miss here can mean a stale or wrong-file entry already occupies
+        # `name`: the digest is only 32 bits, so two distinct files with the
+        # same sanitized stem can collide on it. Replacing the entry, rather
+        # than erroring, is safe because nothing depends on sys.modules[name]
+        # continuing to point at the other file's module -- a Measure holds
+        # its function directly, not a lookup through this name -- so the
+        # collision degrades to that other file re-executing on its own next
+        # load (the identity check above will miss for it too), never to
+        # this load returning its measures.
+        module = importlib.util.module_from_spec(spec)
+        sys.modules[name] = module
+        _load_mtimes[name] = mtime_ns
+
+        # Appended, not inserted at the front: a sibling file can then
+        # import another sibling by plain absolute import, but a sibling
+        # named like a stdlib module must not shadow it for the rest of the
+        # process.
+        directory = str(path.parent)
+        added_to_path = directory not in sys.path
+        if added_to_path:
+            sys.path.append(directory)
+        try:
+            spec.loader.exec_module(module)
+        except BaseException:
+            if sys.modules.get(name) is module:
+                sys.modules.pop(name, None)
+            _load_mtimes.pop(name, None)
+            raise
+        finally:
+            if added_to_path and directory in sys.path:
+                sys.path.remove(directory)
+        return module
+
+
+def _cached_module_path(module: ModuleType) -> Path | None:
+    """Resolve a cached module's own file, to confirm a name match is
+    actually the same file and not a truncated-digest collision between two
+    different ones.
+    """
+    file = getattr(module, "__file__", None)
+    return Path(file).resolve() if file else None
+
+
+def _invalidate_bytecode_cache(path: Path) -> None:
+    """Remove one file's compiled cache before executing it.
+
+    Called on every execution, not only a detected reload: a stale,
+    timestamp-valid .pyc can predate this process entirely, written by an
+    earlier one. SourceFileLoader validates its own .pyc by whole-second
+    mtime and size, coarser than the nanosecond mtime this module's cache
+    compares against; an edit within the same second that leaves the file's
+    size unchanged (changing one digit, say) is exactly the case
+    SourceFileLoader cannot detect, whether this is a reload this process
+    already knows about or a first load of a file some other process
+    touched. Left uninvalidated, it hands back the stale compiled code and
+    the load silently runs the old version. Best-effort and scoped to this
+    one file: the cache directory may be read-only or already gone, and
+    __pycache__ is disposable by design, but only this file's entry is
+    touched.
+    """
+    try:
+        Path(importlib.util.cache_from_source(str(path))).unlink(missing_ok=True)
+    except (OSError, ValueError, NotImplementedError):
+        pass
+
+
+def _check_directory_importable(directory: Path, requested: Path) -> None:
+    """Fail before a directory goes on sys.path if a file in it would shadow
+    an importable module or collide with one already loaded from elsewhere.
+
+    Every .py file in the directory is checked, not only ``requested``: the
+    sys.path entry makes all of them importable, so an unloaded file with a
+    colliding name is exactly as dangerous. ``requested`` is named in every
+    message so a sibling file's collision is not reported with nothing
+    connecting it to the file the caller actually asked to load.
+    """
+    for entry in sorted(directory.glob("*.py")):
+        if entry.name == "__init__.py":
+            continue
+        stem = entry.stem
+
+        # Must run before the directory joins sys.path: added first, the
+        # file would resolve to itself and every directory would look
+        # shadowed. A stdlib name is always findable, so it is folded into
+        # this check rather than tested separately, keeping exactly one
+        # raise per entry. find_spec() also catches a module already cached
+        # in sys.modules under this name (e.g. by an earlier measure
+        # directory's sibling import), except when that cached entry has no
+        # discoverable spec, which find_spec() reports by raising instead of
+        # returning one; the sys.modules check below catches that case.
+        try:
+            spec = importlib.util.find_spec(stem)
+        except (ImportError, ValueError):
+            spec = None
+        if spec is not None and (
+            spec.origin is None or Path(spec.origin).resolve() != entry.resolve()
+        ):
+            if stem in sys.stdlib_module_names:
+                raise ValueError(
+                    f"While loading {requested}, {entry} collides with the "
+                    f"standard library module {stem!r}; one of the two will "
+                    f"be unreachable.\n"
+                    f"Rename {entry}."
+                )
+            origin_note = f" ({spec.origin})" if spec.origin else ""
+            raise ValueError(
+                f"While loading {requested}, {entry} would be shadowed by "
+                f"the already-importable module {stem!r}{origin_note}.\n"
+                f"Rename {entry}."
+            )
+
+        existing = sys.modules.get(stem)
+        existing_file = getattr(existing, "__file__", None) if existing else None
+        if existing is not None and (
+            existing_file is None or Path(existing_file).resolve() != entry.resolve()
+        ):
+            raise ValueError(
+                f"While loading {requested}, {entry} collides with {stem!r}, "
+                f"already imported from "
+                f"{existing_file or 'a module with no file'}.\n"
+                f"Rename {entry}."
+            )
+
+
+def _source_text(func: Callable[..., Any]) -> str:
+    try:
+        return inspect.getsource(func)
+    except (OSError, TypeError):
+        return f"# source unavailable for {func.__name__}"
diff --git a/pkg-py/tests/measure_sources/README.md b/pkg-py/tests/measure_sources/README.md
new file mode 100644
index 0000000..8dd169f
--- /dev/null
+++ b/pkg-py/tests/measure_sources/README.md
@@ -0,0 +1,5 @@
+Two traps for the next person editing this directory: the top level is
+itself a fixture case, so adding any `.py` file here changes the expected
+measure list in `test_semantic_layer_reads_a_directory_without_recursing`;
+and the collision check scans every sibling file, so adding a top-level file
+named after any importable module breaks every path-loading test at once.
diff --git a/pkg-py/tests/measure_sources/broken/broken_import.py b/pkg-py/tests/measure_sources/broken/broken_import.py
new file mode 100644
index 0000000..653a110
--- /dev/null
+++ b/pkg-py/tests/measure_sources/broken/broken_import.py
@@ -0,0 +1,6 @@
+"""Raises at import time, to test that a failed load does not dirty sys.modules.
+
+Lives in a subdirectory so a non-recursive directory scan never reaches it.
+"""
+
+raise RuntimeError("boom")
diff --git a/pkg-py/tests/measure_sources/broken/self_removing_import.py b/pkg-py/tests/measure_sources/broken/self_removing_import.py
new file mode 100644
index 0000000..c1ba07a
--- /dev/null
+++ b/pkg-py/tests/measure_sources/broken/self_removing_import.py
@@ -0,0 +1,11 @@
+"""Deletes its own sys.modules entry, then raises.
+
+Regression fixture for _load_module_from_path's cleanup: it must not turn
+this into a KeyError and swallow the real import error.
+"""
+
+import sys
+
+del sys.modules[__name__]
+
+raise RuntimeError("boom")
diff --git a/pkg-py/tests/measure_sources/collision_a/shared_lib.py b/pkg-py/tests/measure_sources/collision_a/shared_lib.py
new file mode 100644
index 0000000..2ef481e
--- /dev/null
+++ b/pkg-py/tests/measure_sources/collision_a/shared_lib.py
@@ -0,0 +1,5 @@
+"""Shares a file name with collision_b/shared_lib.py on purpose."""
+
+
+def value() -> int:
+    return 1
diff --git a/pkg-py/tests/measure_sources/collision_a/uses_shared.py b/pkg-py/tests/measure_sources/collision_a/uses_shared.py
new file mode 100644
index 0000000..6f48f21
--- /dev/null
+++ b/pkg-py/tests/measure_sources/collision_a/uses_shared.py
@@ -0,0 +1,12 @@
+"""Imports shared_lib by its bare name, registering it in sys.modules under
+that name for the rest of the process.
+"""
+
+from shared_lib import value  # type: ignore[missing-import]
+
+from commons._measures import measure
+
+
+@measure(description="From directory a.")
+def a_measure() -> int:
+    return value()
diff --git a/pkg-py/tests/measure_sources/collision_b/shared_lib.py b/pkg-py/tests/measure_sources/collision_b/shared_lib.py
new file mode 100644
index 0000000..2ca54b4
--- /dev/null
+++ b/pkg-py/tests/measure_sources/collision_b/shared_lib.py
@@ -0,0 +1,9 @@
+"""Shares a file name with collision_a/shared_lib.py on purpose.
+
+Loading anything from this directory must fail once collision_a's
+shared_lib.py has already been imported under the bare name "shared_lib".
+"""
+
+
+def value() -> int:
+    return 2
diff --git a/pkg-py/tests/measure_sources/duplicate_helpers/a_file.py b/pkg-py/tests/measure_sources/duplicate_helpers/a_file.py
new file mode 100644
index 0000000..a6358dc
--- /dev/null
+++ b/pkg-py/tests/measure_sources/duplicate_helpers/a_file.py
@@ -0,0 +1,13 @@
+"""First file, sorted before b_file.py in this directory."""
+
+from commons._measures import measure
+
+
+def helper() -> int:
+    """A helper this file's measure calls."""
+    return 1
+
+
+@measure(description="Measure a.")
+def measure_a() -> int:
+    return helper()
diff --git a/pkg-py/tests/measure_sources/duplicate_helpers/b_file.py b/pkg-py/tests/measure_sources/duplicate_helpers/b_file.py
new file mode 100644
index 0000000..6734b4c
--- /dev/null
+++ b/pkg-py/tests/measure_sources/duplicate_helpers/b_file.py
@@ -0,0 +1,17 @@
+"""Second file, sorted after a_file.py; defines a same-named helper.
+
+Proves directory scanning keeps the first file's source for a colliding
+helper name.
+"""
+
+from commons._measures import measure
+
+
+def helper() -> int:
+    """A colliding helper name; this definition must lose to a_file's."""
+    return 2
+
+
+@measure(description="Measure b.")
+def measure_b() -> int:
+    return helper()
diff --git a/pkg-py/tests/measure_sources/nested/orders.py b/pkg-py/tests/measure_sources/nested/orders.py
new file mode 100644
index 0000000..a17d26b
--- /dev/null
+++ b/pkg-py/tests/measure_sources/nested/orders.py
@@ -0,0 +1,12 @@
+"""Shares a file name with the parent directory's orders.py on purpose.
+
+Directory loading must not reach it, and loading it explicitly must not
+collide with the other orders.py in sys.modules.
+"""
+
+from commons._measures import measure
+
+
+@measure(description="Count of nested orders.")
+def nested_order_count() -> int:
+    return 1
diff --git a/pkg-py/tests/measure_sources/orders.py b/pkg-py/tests/measure_sources/orders.py
new file mode 100644
index 0000000..614fa97
--- /dev/null
+++ b/pkg-py/tests/measure_sources/orders.py
@@ -0,0 +1,20 @@
+"""Measures loaded from a path by the test suite."""
+
+from typing import Annotated, Any
+
+from pydantic import Field
+
+from commons._measures import Injected, measure
+
+
+def double(x: int) -> int:
+    """A helper the measure calls. Not a measure itself."""
+    return x * 2
+
+
+@measure(description="Count of orders.")
+def order_count(
+    region: Annotated[str, Field(description="The sales region.")],
+    warehouse: Injected[Any],
+) -> int:
+    return double(1)
diff --git a/pkg-py/tests/measure_sources/reentrant/composes_a_sibling.py b/pkg-py/tests/measure_sources/reentrant/composes_a_sibling.py
new file mode 100644
index 0000000..0fd7e0a
--- /dev/null
+++ b/pkg-py/tests/measure_sources/reentrant/composes_a_sibling.py
@@ -0,0 +1,17 @@
+"""Calls semantic_layer() on another path during its own import.
+
+A non-reentrant lock around the import machinery would deadlock here: this
+module's own load already holds _IMPORT_LOCK when the line below tries to
+acquire it again on the same thread.
+"""
+
+from pathlib import Path
+
+from commons._measures import measure, semantic_layer
+
+NESTED_LAYER = semantic_layer(Path(__file__).parent.parent / "nested" / "orders.py")
+
+
+@measure(description="Outer measure.")
+def outer_measure() -> int:
+    return 1
diff --git a/pkg-py/tests/measure_sources/revenue.py b/pkg-py/tests/measure_sources/revenue.py
new file mode 100644
index 0000000..78375b0
--- /dev/null
+++ b/pkg-py/tests/measure_sources/revenue.py
@@ -0,0 +1,8 @@
+"""A second file in the same directory, to prove directory loading."""
+
+from commons._measures import measure
+
+
+@measure(description="Total revenue.")
+def total_revenue() -> int:
+    return 100
diff --git a/pkg-py/tests/measure_sources/sibling_imports/helper_lib.py b/pkg-py/tests/measure_sources/sibling_imports/helper_lib.py
new file mode 100644
index 0000000..ba7b0a3
--- /dev/null
+++ b/pkg-py/tests/measure_sources/sibling_imports/helper_lib.py
@@ -0,0 +1,6 @@
+"""A helper file a sibling measure file imports directly."""
+
+
+def double(x: int) -> int:
+    """Doubles a value; imported by a sibling file, not a measure itself."""
+    return x * 2
diff --git a/pkg-py/tests/measure_sources/sibling_imports/uses_helper.py b/pkg-py/tests/measure_sources/sibling_imports/uses_helper.py
new file mode 100644
index 0000000..5ca6fe5
--- /dev/null
+++ b/pkg-py/tests/measure_sources/sibling_imports/uses_helper.py
@@ -0,0 +1,12 @@
+"""Imports a sibling file by plain absolute import, the way an ordinary
+Python module does.
+"""
+
+from helper_lib import double  # type: ignore[missing-import]
+
+from commons._measures import measure
+
+
+@measure(description="Doubled count.")
+def doubled_count() -> int:
+    return double(21)
diff --git a/pkg-py/tests/measure_sources/stdlib_collision/json.py b/pkg-py/tests/measure_sources/stdlib_collision/json.py
new file mode 100644
index 0000000..6fa34b5
--- /dev/null
+++ b/pkg-py/tests/measure_sources/stdlib_collision/json.py
@@ -0,0 +1,10 @@
+"""Named after a standard library module on purpose: loading this must fail
+before the directory ever goes on sys.path.
+"""
+
+from commons._measures import measure
+
+
+@measure(description="Should never load.")
+def unreachable_measure() -> int:
+    return 1
diff --git a/pkg-py/tests/test_measures.py b/pkg-py/tests/test_measures.py
index 471827b..12d2e98 100644
--- a/pkg-py/tests/test_measures.py
+++ b/pkg-py/tests/test_measures.py
@@ -1,8 +1,14 @@
 """The semantic layer: measures, their schemas, and injected arguments."""
 
 import enum
+import importlib
+import os
+import sys
+import threading
 from collections.abc import AsyncIterator
 from dataclasses import FrozenInstanceError
+from pathlib import Path
+from types import ModuleType
 from typing import Annotated, Any, Literal, get_args, get_origin
 
 import pytest
@@ -12,10 +18,13 @@
     INJECTED,
     Injected,
     Measure,
+    _load_module_from_path,
+    _load_mtimes,
     _split_parameters,
     as_measure,
     measure,
     measure_schema_text,
+    semantic_layer,
 )
 
 from ._shared import load_shared_fixture
@@ -536,3 +545,412 @@ def regional_orders(
     rendered = measure_schema_text(_as_measure(regional_orders))
 
     assert "regions (array of {EMEA, AMER}, optional) Enum array, nullable." in rendered
+
+
+def test_semantic_layer_keys_measures_by_name() -> None:
+    @measure(description="Count of orders.")
+    def order_count() -> int:
+        return 1
+
+    layer = semantic_layer(order_count)
+
+    assert list(layer.measures) == ["order_count"]
+    assert layer.measures["order_count"].description == "Count of orders."
+
+
+def test_semantic_layer_accepts_a_list_of_measures() -> None:
+    @measure(description="Count of orders.")
+    def order_count() -> int:
+        return 1
+
+    @measure(description="Total revenue.")
+    def total_revenue() -> int:
+        return 2
+
+    layer = semantic_layer([order_count, total_revenue])
+
+    assert list(layer.measures) == ["order_count", "total_revenue"]
+
+
+def test_semantic_layer_accepts_a_bare_measure_record() -> None:
+    layer = semantic_layer(_count_measure())
+
+    assert list(layer.measures) == ["order_count"]
+
+
+def test_semantic_layer_is_empty_with_no_arguments() -> None:
+    layer = semantic_layer()
+
+    assert len(layer) == 0
+    assert layer.measures == {}
+
+
+def test_semantic_layer_rejects_a_non_measure() -> None:
+    with pytest.raises(TypeError, match="2026"):
+        semantic_layer(2026)
+
+
+def test_semantic_layer_rejects_an_undecorated_function() -> None:
+    def helper() -> int:
+        return 1
+
+    with pytest.raises(TypeError, match="helper"):
+        semantic_layer(helper)
+
+
+def test_semantic_layer_rejects_duplicate_names() -> None:
+    @measure(description="Count of orders.")
+    def order_count() -> int:
+        return 1
+
+    with pytest.raises(ValueError, match="order_count"):
+        semantic_layer(order_count, order_count)
+
+
+def test_semantic_layer_harvests_inline_measure_source() -> None:
+    @measure(description="Count of orders.")
+    def order_count() -> int:
+        return 1
+
+    layer = semantic_layer(order_count)
+
+    assert "def order_count()" in layer.source_text["order_count"]
+
+
+def test_collect_nested_list_keeps_first_definition_wins() -> None:
+    # Both functions are named `calc`, so they collide in `source_text`
+    # (keyed by Python name) without colliding in `measures` (keyed by the
+    # distinct `name=` given to each).
+    def make_first() -> Any:
+        @measure(description="First.", name="first")
+        def calc() -> int:
+            return 1
+
+        return as_measure(calc)
+
+    def make_second() -> Any:
+        @measure(description="Second.", name="second")
+        def calc() -> int:
+            return 2
+
+        return as_measure(calc)
+
+    first, second = make_first(), make_second()
+
+    top_level = semantic_layer(first, second)
+    nested = semantic_layer([first, second])
+
+    assert nested.source_text["calc"] == top_level.source_text["calc"]
+    assert "return 1" in nested.source_text["calc"]
+
+
+def test_semantic_layer_reports_its_size() -> None:
+    layer = semantic_layer(_count_measure())
+
+    assert len(layer) == 1
+    assert "1 measure" in repr(layer)
+
+
+MEASURE_FILES = Path(__file__).parent / "measure_sources"
+
+
+def test_semantic_layer_reads_a_file_path() -> None:
+    layer = semantic_layer(MEASURE_FILES / "orders.py")
+
+    assert list(layer.measures) == ["order_count"]
+
+
+def test_semantic_layer_accepts_a_string_path() -> None:
+    layer = semantic_layer(str(MEASURE_FILES / "orders.py"))
+
+    assert list(layer.measures) == ["order_count"]
+
+
+def test_semantic_layer_reads_a_directory_without_recursing() -> None:
+    layer = semantic_layer(MEASURE_FILES)
+
+    assert list(layer.measures) == ["order_count", "total_revenue"]
+
+
+def test_semantic_layer_reads_a_module_object() -> None:
+    module = importlib.import_module("commons._measures")
+
+    layer = semantic_layer(module)
+
+    assert layer.measures == {}
+
+
+def test_semantic_layer_mixes_files_and_inline_measures() -> None:
+    @measure(description="Inline.")
+    def inline_measure() -> int:
+        return 1
+
+    layer = semantic_layer(MEASURE_FILES / "orders.py", inline_measure)
+
+    assert list(layer.measures) == ["order_count", "inline_measure"]
+
+
+def test_semantic_layer_harvests_helper_source_alongside_measures() -> None:
+    layer = semantic_layer(MEASURE_FILES / "orders.py")
+
+    assert set(layer.source_text) >= {"double", "order_count"}
+    assert "x * 2" in layer.source_text["double"]
+    assert "@measure(" in layer.source_text["order_count"]
+
+
+def test_harvested_source_excludes_imported_names() -> None:
+    layer = semantic_layer(MEASURE_FILES / "orders.py")
+
+    assert "measure" not in layer.source_text
+    assert "Field" not in layer.source_text
+
+
+def test_only_text_leaves_the_semantic_layer() -> None:
+    layer = semantic_layer(MEASURE_FILES / "orders.py")
+
+    assert all(isinstance(text, str) for text in layer.source_text.values())
+
+
+def test_same_file_name_in_two_directories_both_load() -> None:
+    layer = semantic_layer(
+        MEASURE_FILES / "orders.py", MEASURE_FILES / "nested" / "orders.py"
+    )
+
+    assert list(layer.measures) == ["order_count", "nested_order_count"]
+
+
+def test_missing_path_is_an_error() -> None:
+    with pytest.raises(ValueError, match="not a measure"):
+        semantic_layer("not a measure")
+
+
+def test_missing_path_error_names_the_path() -> None:
+    with pytest.raises(ValueError, match="nowhere.py"):
+        semantic_layer(MEASURE_FILES / "nowhere.py")
+
+
+def test_directory_scan_keeps_the_first_files_helper_source() -> None:
+    # a_file.py sorts before b_file.py; both define a `helper` function, and
+    # the first one scanned must win.
+    layer = semantic_layer(MEASURE_FILES / "duplicate_helpers")
+
+    assert list(layer.measures) == ["measure_a", "measure_b"]
+    assert "return 1" in layer.source_text["helper"]
+    assert "return 2" not in layer.source_text["helper"]
+
+
+def test_failed_import_does_not_dirty_sys_modules() -> None:
+    path = MEASURE_FILES / "broken" / "broken_import.py"
+
+    with pytest.raises(RuntimeError, match="boom"):
+        semantic_layer(path)
+
+    assert not any("broken_import" in name for name in sys.modules)
+
+
+def test_failed_import_that_deletes_its_own_module_entry_still_raises() -> None:
+    # If the module removes its sys.modules entry before raising, cleanup
+    # must not turn the real error into a KeyError.
+    path = MEASURE_FILES / "broken" / "self_removing_import.py"
+
+    with pytest.raises(RuntimeError, match="boom"):
+        semantic_layer(path)
+
+
+def test_measure_file_imports_a_sibling_file_directly() -> None:
+    layer = semantic_layer(MEASURE_FILES / "sibling_imports" / "uses_helper.py")
+
+    assert list(layer.measures) == ["doubled_count"]
+    assert layer.measures["doubled_count"].func() == 42
+
+
+def test_dotted_filename_does_not_get_a_dotted_module_name(
+    tmp_path: Path,
+) -> None:
+    # path.stem for "sales.q3.py" is "sales.q3": left unsanitized, the
+    # generated module name would still be dotted, giving the loaded file a
+    # non-empty __package__ and undoing the single-segment name fix.
+    dotted = tmp_path / "sales.q3.py"
+    dotted.write_text("from ..nope import thing\n")
+
+    with pytest.raises(
+        ImportError, match="attempted relative import with no known parent package"
+    ):
+        semantic_layer(dotted)
+
+
+def test_load_module_from_path_reuses_an_unchanged_file(tmp_path: Path) -> None:
+    source = tmp_path / "m.py"
+    source.write_text("VALUE = 1\n")
+
+    first = _load_module_from_path(source)
+    second = _load_module_from_path(source)
+
+    assert first is second
+
+
+def test_load_module_from_path_reloads_an_edited_file(tmp_path: Path) -> None:
+    # An edit within the same whole second that leaves the file's size
+    # unchanged ("VALUE = 1" -> "VALUE = 2"): the case SourceFileLoader's own
+    # bytecode cache cannot detect, since it validates by whole-second mtime
+    # and size, coarser than the nanosecond mtime our cache compares
+    # against. Both mtimes are pinned explicitly, not read off the file
+    # naturally and nudged, so the test cannot straddle a real second
+    # boundary and become flaky.
+    source = tmp_path / "m.py"
+    base_ns = 1_700_000_000 * 1_000_000_000
+    source.write_text("VALUE = 1\n")
+    os.utime(source, ns=(base_ns, base_ns))
+    first = _load_module_from_path(source)
+
+    source.write_text("VALUE = 2\n")
+    os.utime(source, ns=(base_ns, base_ns + 500_000_000))
+
+    second = _load_module_from_path(source)
+
+    assert second is not first
+    assert second.VALUE == 2
+
+
+def test_load_module_from_path_invalidates_a_pre_existing_bytecode_cache(
+    tmp_path: Path,
+) -> None:
+    # A stale, timestamp-valid .pyc can predate this process's own record of
+    # having loaded the file at all -- written by an earlier process, then
+    # the file edited same-second, same-size before this process's first
+    # load of it. Simulated here without spawning a real second process: load
+    # once to produce the .pyc via SourceFileLoader, edit the file, then
+    # clear this process's own sys.modules and _load_mtimes entries for it
+    # so the next load has no in-memory record either -- indistinguishable,
+    # from _load_module_from_path's point of view, from a fresh process's
+    # first load of an already-edited file.
+    source = tmp_path / "m.py"
+    base_ns = 1_700_000_000 * 1_000_000_000
+    source.write_text("VALUE = 1\n")
+    os.utime(source, ns=(base_ns, base_ns))
+    first = _load_module_from_path(source)
+    name = first.__name__
+
+    source.write_text("VALUE = 2\n")
+    os.utime(source, ns=(base_ns, base_ns + 500_000_000))
+    sys.modules.pop(name, None)
+    _load_mtimes.pop(name, None)
+
+    second = _load_module_from_path(source)
+
+    assert second.VALUE == 2
+
+
+def test_load_module_from_path_ignores_a_same_named_module_from_elsewhere(
+    tmp_path: Path,
+) -> None:
+    # Exercises the identity check directly rather than forcing a genuine
+    # 32-bit digest collision between two distinct filenames: plant a module
+    # under the exact sys.modules name this path would use, with the same
+    # recorded mtime but a __file__ pointing elsewhere, and confirm the real
+    # file is (re-)loaded rather than the stand-in being returned.
+    source = tmp_path / "m.py"
+    source.write_text("VALUE = 1\n")
+    real = _load_module_from_path(source)
+    name = real.__name__
+
+    imposter = ModuleType(name)
+    imposter.__file__ = str(tmp_path / "elsewhere.py")
+    sys.modules[name] = imposter
+
+    loaded = _load_module_from_path(source)
+
+    assert loaded is not imposter
+    assert loaded.VALUE == 1
+
+
+def test_sys_path_is_restored_after_a_successful_load() -> None:
+    directory = str(MEASURE_FILES / "sibling_imports")
+
+    semantic_layer(MEASURE_FILES / "sibling_imports" / "uses_helper.py")
+
+    assert directory not in sys.path
+
+
+def test_sys_path_is_restored_after_a_failing_load() -> None:
+    directory = str(MEASURE_FILES / "broken")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        semantic_layer(MEASURE_FILES / "broken" / "broken_import.py")
+
+    assert directory not in sys.path
+
+
+def test_stdlib_name_collision_is_a_construction_error() -> None:
+    path = MEASURE_FILES / "stdlib_collision" / "json.py"
+
+    with pytest.raises(ValueError, match="json.py") as excinfo:
+        semantic_layer(path)
+
+    assert "standard library" in str(excinfo.value)
+
+
+def test_same_named_helper_in_two_directories_is_a_construction_error() -> None:
+    try:
+        semantic_layer(MEASURE_FILES / "collision_a" / "uses_shared.py")
+
+        with pytest.raises(ValueError, match="shared_lib"):
+            semantic_layer(MEASURE_FILES / "collision_b" / "shared_lib.py")
+    finally:
+        sys.modules.pop("shared_lib", None)
+
+
+def test_installed_but_unimported_module_is_a_construction_error(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # A directory on sys.path stands in for an installed package: find_spec()
+    # can resolve it without anything having imported it yet.
+    site_packages = tmp_path / "site-packages"
+    site_packages.mkdir()
+    (site_packages / "certainly_not_a_measure.py").write_text("VALUE = 1\n")
+    monkeypatch.syspath_prepend(str(site_packages))
+    sys.modules.pop("certainly_not_a_measure", None)
+
+    measures_dir = tmp_path / "measures"
+    measures_dir.mkdir()
+    colliding = measures_dir / "certainly_not_a_measure.py"
+    colliding.write_text(
+        "from commons._measures import measure\n\n\n"
+        "@measure(description='d')\n"
+        "def m() -> int:\n"
+        "    return 1\n"
+    )
+
+    with pytest.raises(ValueError, match="certainly_not_a_measure.py") as excinfo:
+        semantic_layer(colliding)
+
+    message = str(excinfo.value)
+    assert "certainly_not_a_measure" in message
+    assert "already-importable" in message
+
+
+def test_semantic_layer_reenters_during_a_measure_files_import() -> None:
+    # A non-reentrant lock deadlocks here rather than raising, so this runs
+    # on a daemon thread with a timeout: a regression fails the test instead
+    # of hanging the suite.
+    result: dict[str, Any] = {}
+
+    def target() -> None:
+        result["layer"] = semantic_layer(
+            MEASURE_FILES / "reentrant" / "composes_a_sibling.py"
+        )
+
+    thread = threading.Thread(target=target, daemon=True)
+    thread.start()
+    thread.join(timeout=5)
+
+    assert not thread.is_alive(), (
+        "semantic_layer() deadlocked re-entering during a measure file's import"
+    )
+
+    outer_layer = result["layer"]
+    assert list(outer_layer.measures) == ["outer_measure"]
+
+    module_name = outer_layer.measures["outer_measure"].func.__module__
+    fixture_module = sys.modules[module_name]
+    assert list(fixture_module.NESTED_LAYER.measures) == ["nested_order_count"]