diff --git a/pkg-py/src/commons/_measures.py b/pkg-py/src/commons/_measures.py
new file mode 100644
index 00000000..192b4d18
--- /dev/null
+++ b/pkg-py/src/commons/_measures.py
@@ -0,0 +1,260 @@
+"""The semantic layer: trusted calculations an agent can run.
+
+A measure's signature carries both kinds of argument. Parameters annotated
+``Annotated[T, Field(description=...)]`` are supplied by the model; parameters
+annotated ``Injected[T]`` are supplied by commons at call time and never reach
+the model. A parameter that is neither is an error, so an argument cannot be
+hidden from the model by forgetting to describe it.
+
+``pkg-r/R/measures.R`` implements the same semantic layer for R.
+"""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from typing import (
+    Annotated,
+    Any,
+    Final,
+    TypeVar,
+    cast,
+    get_args,
+    get_origin,
+    get_type_hints,
+)
+
+from pydantic import BaseModel, ConfigDict, create_model
+from pydantic.fields import FieldInfo
+from pydantic_core import PydanticUndefined
+
+
+class _InjectedMarker:
+    """Sentinel distinguishing a commons-supplied parameter from a described one."""
+
+    def __repr__(self) -> str:
+        return "Injected"
+
+
+INJECTED: Final = _InjectedMarker()
+
+_T = TypeVar("_T")
+
+# `Injected[Engine]` expands to `Annotated[Engine, INJECTED]`, so the marker
+# survives `get_type_hints(include_extras=True)` and stays out of the schema.
+Injected = Annotated[_T, INJECTED]
+
+MEASURE_ATTRIBUTE: Final = "__commons_measure__"
+
+
+@dataclass(frozen=True)
+class Measure:
+    """A trusted calculation the agent can run.
+
+    ``params`` describes only the arguments the model supplies; ``injected``
+    names the arguments commons supplies, which the model never sees.
+    """
+
+    name: str
+    title: str
+    description: str
+    func: Callable[..., Any]
+    params: type[BaseModel]
+    injected: tuple[str, ...] = ()
+    provenance: tuple[str, ...] = ()
+
+    def validate_args(self, args: Mapping[str, Any] | None) -> dict[str, Any]:
+        """Check and coerce the model's arguments against the schema.
+
+        The provider only ever sees ``call_measure``, so a measure's own
+        arguments arrive unchecked and are validated here.
+        """
+        validated = self.params.model_validate(dict(args or {}))
+        return validated.model_dump(exclude_unset=True)
+
+
+def _split_parameters(
+    func: Callable[..., Any],
+) -> tuple[dict[str, tuple[Any, FieldInfo]], tuple[str, ...]]:
+    if inspect.iscoroutinefunction(func) or inspect.isasyncgenfunction(func):
+        raise TypeError(_async_message(func))
+
+    try:
+        hints = get_type_hints(func, include_extras=True)
+    except NameError as error:
+        raise TypeError(_unresolvable_hint_message(func, error)) from error
+
+    fields: dict[str, tuple[Any, FieldInfo]] = {}
+    injected: list[str] = []
+
+    for name, param in inspect.signature(func).parameters.items():
+        if param.kind is inspect.Parameter.VAR_POSITIONAL:
+            raise TypeError(_unsupported_message(func, f"*{name}"))
+        if param.kind is inspect.Parameter.VAR_KEYWORD:
+            raise TypeError(_unsupported_message(func, f"**{name}"))
+        if param.kind is inspect.Parameter.POSITIONAL_ONLY:
+            raise TypeError(
+                _unsupported_message(func, f"positional-only parameter {name!r}")
+            )
+
+        annotation = hints.get(name, inspect.Parameter.empty)
+        field_meta = _described_field(annotation)
+        if _is_injected(annotation):
+            if field_meta is not None:
+                raise TypeError(_dual_marker_message(func, name))
+            injected.append(name)
+            continue
+
+        if field_meta is None:
+            raise TypeError(_undeclared_message(func, name))
+
+        if param.default is inspect.Parameter.empty:
+            if _annotation_declares_default(annotation):
+                raise TypeError(_annotation_default_message(func, name))
+            default = PydanticUndefined
+        else:
+            default = param.default
+
+        field = FieldInfo.from_annotated_attribute(annotation, default)
+        fields[name] = (get_args(annotation)[0], field)
+
+    return fields, tuple(injected)
+
+
+def _is_injected(annotation: Any) -> bool:
+    return get_origin(annotation) is Annotated and any(
+        metadata is INJECTED for metadata in get_args(annotation)[1:]
+    )
+
+
+def _described_field(annotation: Any) -> FieldInfo | None:
+    if get_origin(annotation) is not Annotated:
+        return None
+    for metadata in get_args(annotation)[1:]:
+        if isinstance(metadata, FieldInfo) and (metadata.description or "").strip():
+            return metadata
+    return None
+
+
+def _annotation_declares_default(annotation: Any) -> bool:
+    # An annotation can carry more than one Field(...); the default can hide
+    # in any of them, not just the one _described_field() returns.
+    if get_origin(annotation) is not Annotated:
+        return False
+    return any(
+        isinstance(metadata, FieldInfo)
+        and (
+            metadata.default is not PydanticUndefined
+            or metadata.default_factory is not None
+        )
+        for metadata in get_args(annotation)[1:]
+    )
+
+
+def _undeclared_message(func: Callable[..., Any], name: str) -> str:
+    return (
+        f"Parameter {name!r} of measure {func.__name__!r} is neither described "
+        f"nor injected.\n"
+        f"Annotate it Annotated[T, Field(description=...)] for the model to "
+        f"supply it, or Injected[T] for commons to supply it."
+    )
+
+
+def _dual_marker_message(func: Callable[..., Any], name: str) -> str:
+    return (
+        f"Parameter {name!r} of measure {func.__name__!r} is marked both "
+        f"Injected and Field(description=...), so commons cannot tell "
+        f"whether the model or commons supplies it.\n"
+        f"Remove one of the two markers: Injected[T] if commons supplies "
+        f"it, or Annotated[T, Field(description=...)] if the model does."
+    )
+
+
+def _unresolvable_hint_message(func: Callable[..., Any], error: NameError) -> str:
+    name = error.name or str(error)
+    return (
+        f"Measure {func.__name__!r} has an annotation that could not be "
+        f"resolved: {name!r} is not defined.\n"
+        f"This usually means {name!r} is only imported under "
+        f"`if TYPE_CHECKING:`. Import it unconditionally, or annotate the "
+        f"parameter Injected[Any] instead."
+    )
+
+
+def _async_message(func: Callable[..., Any]) -> str:
+    return (
+        f"Measure {func.__name__!r} is defined with async def, but a "
+        f"measure must be a synchronous function.\n"
+        f"Define it with def instead of async def."
+    )
+
+
+def _annotation_default_message(func: Callable[..., Any], name: str) -> str:
+    return (
+        f"Parameter {name!r} of measure {func.__name__!r} declares a default "
+        f"inside Field(...), but the signature has no default.\n"
+        f"Python never applies the Field default, so a call that omits "
+        f"{name!r} would fail. Put the default in the signature instead: "
+        f"{name}: Annotated[T, Field(description=...)] = <default>."
+    )
+
+
+def _unsupported_message(func: Callable[..., Any], name: str) -> str:
+    return (
+        f"Measure {func.__name__!r} takes {name}, which has no schema.\n"
+        f"Declare each argument explicitly."
+    )
+
+
+def measure(
+    *,
+    description: str | None = None,
+    name: str | None = None,
+    title: str | None = None,
+    provenance: Sequence[str] = (),
+) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
+    """Mark a function as a measure.
+
+    The decorated function is returned unchanged, so measures and the helpers
+    they call stay ordinary callables.
+    """
+
+    def decorate(func: Callable[..., Any]) -> Callable[..., Any]:
+        text = description or inspect.getdoc(func) or ""
+        if not text.strip():
+            raise ValueError(
+                f"Measure {func.__name__!r} has no description.\n"
+                f"Pass description= to @measure, or give the function a docstring."
+            )
+        fields, injected = _split_parameters(func)
+        resolved_name = name or func.__name__
+        record = Measure(
+            name=resolved_name,
+            title=title or _humanize(resolved_name),
+            description=text.strip(),
+            func=func,
+            params=create_model(
+                resolved_name,
+                __config__=ConfigDict(extra="forbid"),
+                **cast(dict[str, Any], fields),
+            ),
+            injected=injected,
+            provenance=tuple(provenance),
+        )
+        setattr(func, MEASURE_ATTRIBUTE, record)
+        return func
+
+    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("_", " ")
diff --git a/pkg-py/tests/test_measures.py b/pkg-py/tests/test_measures.py
new file mode 100644
index 00000000..52a36953
--- /dev/null
+++ b/pkg-py/tests/test_measures.py
@@ -0,0 +1,370 @@
+"""The semantic layer: measures, their schemas, and injected arguments."""
+
+from collections.abc import AsyncIterator
+from dataclasses import FrozenInstanceError
+from typing import Annotated, Any, Literal, get_args, get_origin
+
+import pytest
+from pydantic import Field, ValidationError
+
+from commons._measures import (
+    INJECTED,
+    Injected,
+    Measure,
+    _split_parameters,
+    as_measure,
+    measure,
+)
+
+
+def test_injected_alias_carries_the_marker() -> None:
+    alias = Injected[int]
+    assert get_origin(alias) is Annotated
+    assert get_args(alias) == (int, INJECTED)
+
+
+def test_split_parameters_separates_described_from_injected() -> None:
+    def revenue(
+        region: Annotated[str, Field(description="The sales region.")],
+        warehouse: Injected[Any],
+    ) -> int:
+        return 0
+
+    fields, injected = _split_parameters(revenue)
+
+    assert list(fields) == ["region"]
+    assert injected == ("warehouse",)
+
+
+def test_split_parameters_keeps_declaration_order() -> None:
+    def m(
+        b: Annotated[str, Field(description="B.")],
+        a: Annotated[str, Field(description="A.")],
+        con: Injected[Any] = None,
+    ) -> None: ...
+
+    fields, injected = _split_parameters(m)
+
+    assert list(fields) == ["b", "a"]
+    assert injected == ("con",)
+
+
+def test_split_parameters_carries_defaults_into_the_field() -> None:
+    def m(limit: Annotated[int, Field(description="Cap.")] = 10) -> None: ...
+
+    fields, _ = _split_parameters(m)
+
+    assert fields["limit"][1].default == 10
+
+
+def test_field_default_without_signature_default_is_an_error() -> None:
+    def m(limit: Annotated[int, Field(default=10, description="Cap.")]) -> None: ...
+
+    with pytest.raises(TypeError) as excinfo:
+        _split_parameters(m)
+
+    message = str(excinfo.value)
+    assert "limit" in message
+    assert "m" in message
+    assert "= <default>" in message
+
+
+def test_field_default_factory_without_signature_default_is_an_error() -> None:
+    def m(
+        tags: Annotated[list[str], Field(default_factory=list, description="Tags.")],
+    ) -> None: ...
+
+    with pytest.raises(TypeError, match="tags"):
+        _split_parameters(m)
+
+
+def test_field_default_in_a_separate_metadata_item_is_an_error() -> None:
+    def m(
+        limit: Annotated[int, Field(default=10), Field(description="Cap.")],
+    ) -> None: ...
+
+    with pytest.raises(TypeError) as excinfo:
+        _split_parameters(m)
+
+    message = str(excinfo.value)
+    assert "limit" in message
+    assert "= <default>" in message
+
+
+def test_field_default_factory_in_a_separate_metadata_item_is_an_error() -> None:
+    def m(
+        tags: Annotated[
+            list[str], Field(default_factory=list), Field(description="Tags.")
+        ],
+    ) -> None: ...
+
+    with pytest.raises(TypeError, match="tags"):
+        _split_parameters(m)
+
+
+def test_unannotated_parameter_is_an_error() -> None:
+    def m(region) -> None: ...
+
+    with pytest.raises(TypeError, match="region"):
+        _split_parameters(m)
+
+
+def test_bare_annotation_without_a_description_is_an_error() -> None:
+    def m(region: str) -> None: ...
+
+    with pytest.raises(TypeError) as excinfo:
+        _split_parameters(m)
+
+    message = str(excinfo.value)
+    assert "region" in message
+    assert "Field(description=" in message
+    assert "Injected[" in message
+
+
+def test_annotated_without_a_description_is_an_error() -> None:
+    def m(region: Annotated[str, Field()]) -> None: ...
+
+    with pytest.raises(TypeError, match="region"):
+        _split_parameters(m)
+
+
+def test_empty_description_is_an_error() -> None:
+    def m(region: Annotated[str, Field(description="  ")]) -> None: ...
+
+    with pytest.raises(TypeError, match="region"):
+        _split_parameters(m)
+
+
+def test_var_args_are_an_error() -> None:
+    def m(*args: str) -> None: ...
+
+    with pytest.raises(TypeError, match=r"\*args"):
+        _split_parameters(m)
+
+
+def test_var_kwargs_are_an_error() -> None:
+    def m(**kwargs: str) -> None: ...
+
+    with pytest.raises(TypeError, match=r"\*\*kwargs"):
+        _split_parameters(m)
+
+
+def test_positional_only_parameter_is_an_error() -> None:
+    def m(region: Annotated[str, Field(description="The region.")], /) -> None: ...
+
+    with pytest.raises(TypeError, match="region"):
+        _split_parameters(m)
+
+
+def test_async_def_measure_is_an_error() -> None:
+    async def m(x: Annotated[str, Field(description="d")]) -> None: ...
+
+    with pytest.raises(TypeError, match="async") as excinfo:
+        _split_parameters(m)
+
+    assert "m" in str(excinfo.value)
+
+
+def test_async_generator_measure_is_an_error() -> None:
+    # iscoroutinefunction() alone misses this: a `yield` inside an async def
+    # makes it an async generator function, a different kind entirely.
+    async def m(x: Annotated[str, Field(description="d")]) -> AsyncIterator[str]:
+        yield x
+
+    with pytest.raises(TypeError, match="async") as excinfo:
+        _split_parameters(m)
+
+    assert "m" in str(excinfo.value)
+
+
+def test_unresolvable_annotation_names_the_measure_and_the_missing_name() -> None:
+    # Mimics a TYPE_CHECKING-only import: the annotation is a forward
+    # reference get_type_hints() cannot resolve at runtime. Built via exec so
+    # the missing name is never visible to static analysis of this file.
+    namespace: dict[str, Any] = {"Injected": Injected}
+    exec("def m(conn: 'Injected[NoSuchConnection]') -> None: ...", namespace)  # noqa: S102
+    m = namespace["m"]
+
+    with pytest.raises(TypeError) as excinfo:
+        _split_parameters(m)
+
+    message = str(excinfo.value)
+    assert "m" in message
+    assert "NoSuchConnection" in message
+    assert "TYPE_CHECKING" in message
+    assert "Injected[Any]" in message
+
+
+def test_parameter_marked_both_injected_and_described_is_an_error() -> None:
+    def m(x: Injected[Annotated[str, Field(description="d")]]) -> None: ...
+
+    with pytest.raises(TypeError, match="x") as excinfo:
+        _split_parameters(m)
+
+    assert "Injected" in str(excinfo.value)
+
+
+def test_split_parameters_merges_field_constraints() -> None:
+    def m(
+        value: Annotated[int, Field(gt=0), Field(description="Positive.")],
+    ) -> None: ...
+
+    fields, _ = _split_parameters(m)
+
+    field_info = fields["value"][1]
+    assert field_info.description == "Positive."
+    # Verify that the gt=0 constraint is in the metadata
+    assert len(field_info.metadata) > 0
+    assert any(str(m).startswith("Gt") for m in field_info.metadata)
+
+
+def _as_measure(obj: Any) -> Measure:
+    """Narrow as_measure()'s result for tests that require it to succeed."""
+    record = as_measure(obj)
+    assert record is not None
+    return record
+
+
+def _count_measure() -> Measure:
+    """The running example, matching count_measure_tool() in the R suite."""
+
+    @measure(description="Count of orders.")
+    def order_count(
+        region: Annotated[
+            Literal["EMEA", "AMER"], Field(description="The sales region.")
+        ],
+        revenue_under: Annotated[float, Field(description="Cap.")] = 0.0,
+    ) -> int:
+        return 1
+
+    return _as_measure(order_count)
+
+
+def test_measure_defaults_name_and_title_from_the_function() -> None:
+    @measure(description="Count of orders.")
+    def order_count() -> int:
+        return 1
+
+    m = as_measure(order_count)
+    assert m is not None
+    assert m.name == "order_count"
+    assert m.title == "order count"
+    assert m.description == "Count of orders."
+
+
+def test_measure_leaves_the_function_callable() -> None:
+    @measure(description="Count of orders.")
+    def order_count() -> int:
+        return 7
+
+    assert order_count() == 7
+
+
+def test_measure_takes_its_description_from_the_docstring() -> None:
+    @measure()
+    def order_count() -> int:
+        """Count of orders."""
+        return 1
+
+    assert _as_measure(order_count).description == "Count of orders."
+
+
+def test_measure_prefers_an_explicit_description_over_the_docstring() -> None:
+    @measure(description="Explicit.")
+    def order_count() -> int:
+        """Docstring."""
+        return 1
+
+    assert _as_measure(order_count).description == "Explicit."
+
+
+def test_measure_without_any_description_is_an_error() -> None:
+    with pytest.raises(ValueError, match="order_count"):
+
+        @measure()
+        def order_count() -> int:
+            return 1
+
+
+def test_measure_accepts_an_explicit_name_and_title() -> None:
+    @measure(description="d", name="orders", title="Orders placed")
+    def order_count() -> int:
+        return 1
+
+    m = _as_measure(order_count)
+    assert m.name == "orders"
+    assert m.title == "Orders placed"
+
+
+def test_measure_records_provenance_links() -> None:
+    @measure(description="d", provenance=["https://example.com/spec"])
+    def order_count() -> int:
+        return 1
+
+    assert _as_measure(order_count).provenance == ("https://example.com/spec",)
+
+
+def test_measure_hides_injected_parameters_from_the_schema() -> None:
+    @measure(description="Revenue for a region.")
+    def region_revenue(
+        region: Annotated[str, Field(description="The sales region.")],
+        warehouse: Injected[Any],
+    ) -> int:
+        return 0
+
+    m = _as_measure(region_revenue)
+    assert list(m.params.model_fields) == ["region"]
+    assert m.injected == ("warehouse",)
+
+
+def test_as_measure_returns_none_for_a_plain_function() -> None:
+    def helper() -> int:
+        return 1
+
+    assert as_measure(helper) is None
+
+
+def test_as_measure_passes_a_measure_through() -> None:
+    m = _count_measure()
+    assert as_measure(m) is m
+
+
+def test_validate_args_coerces_valid_arguments() -> None:
+    args = _count_measure().validate_args({"region": "EMEA", "revenue_under": "1000"})
+
+    assert args == {"region": "EMEA", "revenue_under": 1000.0}
+
+
+def test_validate_args_rejects_out_of_vocabulary_enum_values() -> None:
+    with pytest.raises(ValidationError, match="LATAM"):
+        _count_measure().validate_args({"region": "LATAM"})
+
+
+def test_validate_args_rejects_unknown_arguments() -> None:
+    with pytest.raises(ValidationError, match="nope"):
+        _count_measure().validate_args({"region": "EMEA", "nope": 1})
+
+
+def test_validate_args_enforces_required_arguments() -> None:
+    with pytest.raises(ValidationError, match="region"):
+        _count_measure().validate_args({})
+
+
+def test_validate_args_omits_arguments_the_model_did_not_send() -> None:
+    args = _count_measure().validate_args({"region": "EMEA"})
+
+    assert args == {"region": "EMEA"}
+
+
+def test_validate_args_treats_none_as_no_arguments() -> None:
+    @measure(description="d")
+    def no_args() -> int:
+        return 1
+
+    assert _as_measure(no_args).validate_args(None) == {}
+
+
+def test_measure_is_frozen() -> None:
+    m = _count_measure()
+    with pytest.raises(FrozenInstanceError):
+        m.name = "other"  # type: ignore[misc]