diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
index 2c52f64..f7b9f36 100644
--- a/.github/workflows/build.yaml
+++ b/.github/workflows/build.yaml
@@ -111,13 +111,13 @@ jobs:
       uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
       with:
         name: test-results-${{ matrix.os }}-${{ matrix.python-version }}
-        path: '**/junit.xml'
+        path: junit.xml
       if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.11'
 
     - name: Publish Unit Test Results
       uses: EnricoMi/publish-unit-test-result-action@c950f6fb443cb5af20a377fd0dfaa78838901040 # v2
       with:
-        files: '**/junit.xml'
+        files: junit.xml
       if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.11'
 
     - name: Upload coverage
diff --git a/Makefile b/Makefile
index 9dd1f1e..d9e6134 100644
--- a/Makefile
+++ b/Makefile
@@ -97,7 +97,7 @@ test-py:  ## run python tests
 tests-py: test-py
 
 coverage-py:  ## run python tests and collect test coverage
-	python -m pytest -v csp_gateway/tests --cov=csp_gateway --cov-report term-missing --cov-report xml
+	python -m pytest -v csp_gateway/tests --junitxml=junit.xml --cov=csp_gateway --cov-report term-missing --cov-report xml
 
 .PHONY: test-js tests-js coverage-js
 test-js:  ## run js tests
diff --git a/csp_gateway/server/demo/omnibus.py b/csp_gateway/server/demo/omnibus.py
index 4354dd3..f4feec8 100644
--- a/csp_gateway/server/demo/omnibus.py
+++ b/csp_gateway/server/demo/omnibus.py
@@ -115,7 +115,14 @@ class ExampleGatewayChannels(GatewayChannels):
     example_list: ts[List[ExampleData]] = None
     never_ticks: ts[ExampleData] = None
 
-    s_example: ts[State[ExampleData]] = None
+    # State fields can be added via annotation or the `set_state` API in the module's `connect` method
+    example_with_state: Annotated[ts[ExampleData], State(("id", "x"))] = None
+    example_with_state_multiple: Annotated[
+        ts[ExampleData],
+        State(("id", "x")),
+        State(("id", "y"), alias="example_with_state_alternative"),
+    ] = None
+    # example: Set in connect with a different schema, to show flexibility of state definition
 
     basket: Dict[ExampleEnum, ts[ExampleData]] = None
     str_basket: Dict[str, ts[ExampleData]] = None
@@ -179,6 +186,8 @@ def connect(self, channels: ExampleGatewayChannels):
         # Channels set via `set_channel`
         channels.set_channel(ExampleGatewayChannels.example, data)
         channels.set_channel(ExampleGatewayChannels.example_list, data_list)
+        channels.set_channel(ExampleGatewayChannels.example_with_state, data)
+        channels.set_channel(ExampleGatewayChannels.example_with_state_multiple, data)
 
         # Generic channel for sending data from non-csp sources
         channels.add_send_channel(ExampleGatewayChannels.example)
@@ -187,8 +196,11 @@ def connect(self, channels: ExampleGatewayChannels):
         channels.add_send_channel(ExampleGatewayChannels.basket, ExampleEnum.C)
         channels.add_send_channel(ExampleGatewayChannels.basket)
 
-        # Rudimentary state accumulation via `set_state`
-        channels.set_state(ExampleGatewayChannels.example, "id")
+        # State accumulation via `set_state`
+        channels.set_state(
+            ExampleGatewayChannels.example,
+            ("id",),
+        )
 
         # Create some data streams for dict baskets
         data_a = self.subscribe(
diff --git a/csp_gateway/server/gateway/csp/channels.py b/csp_gateway/server/gateway/csp/channels.py
index b143b8d..e98896c 100644
--- a/csp_gateway/server/gateway/csp/channels.py
+++ b/csp_gateway/server/gateway/csp/channels.py
@@ -9,6 +9,7 @@
     DefaultDict,
     Dict,
     List,
+    NamedTuple,
     Optional,
     Set,
     Tuple,
@@ -60,6 +61,24 @@
 log = getLogger(__name__)
 
 
+class _StateSpec(NamedTuple):
+    """Describes a state collection attached to a Channels instance.
+
+    ``source_field`` is the channel name whose edge feeds the state, or None
+    when the state was registered via ``set_state`` with a raw edge.
+    """
+
+    source_field: Optional[str]
+    keyby: Union[str, Tuple[str, ...]]
+    indexer: Optional[Union[str, int]] = None
+
+
+def _normalize_keyby(keyby: Union[str, Tuple[str, ...], list]) -> Tuple[str, ...]:
+    if isinstance(keyby, (list, tuple)):
+        return tuple(keyby)
+    return (keyby,)
+
+
 class _SnapshotModelBaseClass(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid", coerce_numbers_to_str=True)
 
@@ -166,6 +185,7 @@ def __new__(mcs: Any, name: Any, bases: Any, namespace: Any, **kwargs: Any) -> A
 
         _add_field_attributes(cls)
         ts_pydantic_field_types = {}
+        declared_states: Dict[str, _StateSpec] = {}
         for field_name, field_type in cls.model_fields.items():
             # Validate that timeseries types contain structs or list of structs
             outer_type = field_type.annotation
@@ -173,6 +193,22 @@ def __new__(mcs: Any, name: Any, bases: Any, namespace: Any, **kwargs: Any) -> A
             if ts_pydantic_field_type is not None:
                 ts_pydantic_field_types[field_name] = ts_pydantic_field_type
 
+            # Collect State(...) annotation markers from Annotated metadata.
+            for meta in getattr(field_type, "metadata", ()):  # pydantic 2 FieldInfo.metadata
+                if isinstance(meta, State):
+                    alias = meta._meta_alias or field_name
+                    if alias in declared_states:
+                        raise ValueError(
+                            f"Duplicate state alias '{alias}' on {name}: already declared on field '{declared_states[alias].source_field}'"
+                        )
+                    declared_states[alias] = _StateSpec(
+                        source_field=field_name,
+                        keyby=_normalize_keyby(meta._meta_keyby),
+                        indexer=meta._meta_indexer,
+                    )
+
+        cls._declared_states = declared_states
+
         ts_pydantic_field_types[_CSP_ENGINE_CYCLE_TIMESTAMP_FIELD] = (Optional[datetime], None)
         dynamic_pydantic_model = create_model("_snapshot_model", __base__=_SnapshotModelBaseClass, **ts_pydantic_field_types)
         cls._snapshot_model = dynamic_pydantic_model
@@ -188,11 +224,17 @@ class Channels(BaseModel, metaclass=ChannelsMetaclass):
     The names of the channels match the names that are used via the different APIs, i.e. REST, WebSockets, Perspective, etc.
     It is expected that developers interact with channels through these APIs (or via get_channel/set_channel in csp).
 
-    Channels that begin with ``s_`` are "state" channels, meaning that they represent a collection of messages, typically
-    the last message grouped by some key (i.e. security id). These are not meant to be interacted with directly, but rather
-    through the "state" part of the REST API.
+    State collections are declared via ``Annotated[ts[X], State(keyby=..., indexer=..., alias=...)]``
+    on a channel field (auto-wired from the channel's edge), or registered at module
+    connect time via ``set_state(field_or_edge, keyby, indexer=None)``. State
+    collections are exposed through the state part of the REST API and
+    ``state``/``query`` helpers.
     """
 
+    # Populated by ChannelsMetaclass from Annotated[ts[X], State(...)] markers.
+    # alias -> _StateSpec(source_field, keyby, indexer)
+    _declared_states: Dict[str, _StateSpec] = {}
+
     model_config = dict(arbitrary_types_allowed=True)  # (for FeedbackOutputDef)
 
     _finalized: bool = PrivateAttr(default=False)
@@ -206,7 +248,19 @@ class Channels(BaseModel, metaclass=ChannelsMetaclass):
     _override_blocks: Dict[Any, Optional[datetime]] = PrivateAttr(default_factory=dict)
     _feedbacks: Dict[int, FeedbackOutputDef] = PrivateAttr(default_factory=dict)
 
+    # alias -> _StateSpec (annotation-declared and set_state-registered combined)
+    _states: Dict[str, _StateSpec] = PrivateAttr(default_factory=dict)
+    # (alias, indexer) -> ConcurrentFutureAdapter trigger
     _state_requests: Dict[Tuple[str, Optional[Union[str, int]]], Any] = PrivateAttr(default_factory=dict)
+    # (alias, indexer) -> bound state Edge
+    _state_edges: Dict[Tuple[str, Optional[Union[str, int]]], Any] = PrivateAttr(default_factory=dict)
+    # (alias, indexer) -> DelayedEdge handed out by get_state for a state that has
+    # been declared (via Module.dynamic_state_channels) but not yet wired by set_state.
+    # The DelayedEdge is bound to the real state Edge once set_state runs, so modules
+    # may call get_state before the owning module's connect() runs.
+    _delayed_state_edges: Dict[Tuple[str, Optional[Union[str, int]]], DelayedEdge] = PrivateAttr(default_factory=dict)
+    # alias -> element type for pre-declared but not-yet-wired dynamic states.
+    _pending_state_element_types: Dict[str, type] = PrivateAttr(default_factory=dict)
     _last_requests: Dict[Tuple[str, Optional[Union[str, int]]], Any] = PrivateAttr(default_factory=dict)
     _next_requests: Dict[Tuple[str, Optional[Union[str, int]]], Any] = PrivateAttr(default_factory=dict)
     _send_channels: Dict[Tuple[str, Optional[Union[str, int]]], Any] = PrivateAttr(default_factory=dict)
@@ -233,6 +287,20 @@ class Channels(BaseModel, metaclass=ChannelsMetaclass):
     # Might be public in the future.
     _null_ts: List[Tuple[str, Optional[str]]] = PrivateAttr(default_factory=list)
 
+    def model_post_init(self, __context: Any) -> None:
+        # Seed instance state registry with class-level declarations from annotations.
+        for alias, spec in self.__class__._declared_states.items():
+            self._states[alias] = spec
+
+    @classmethod
+    def state_aliases(cls) -> List[str]:
+        """Return the list of state aliases declared on the class via annotations."""
+        return list(cls._declared_states.keys())
+
+    def all_state_aliases(self) -> List[str]:
+        """Return all known state aliases (declared + dynamically registered)."""
+        return list(self._states.keys())
+
     def dynamic_keys(self) -> Optional[Dict[str, List[Any]]]:
         """Define dynamic dictionary keys by field, driven by data from the channels."""
         ...
@@ -340,6 +408,9 @@ def _finalize(self) -> None:
                 who_requires_id[requires].add(id(module))
 
         if not self._finalized:
+            # Auto-wire state collections declared via annotations.
+            self._wire_declared_states()
+
             # first ensure everything is provided
             for (
                 field,
@@ -435,6 +506,29 @@ def _finalize(self) -> None:
         self._finalized = True
         log.debug(f"Feedback count: {self._feedback_count}")
 
+    def _wire_declared_states(self) -> None:
+        for alias, spec in list(self._states.items()):
+            if spec.source_field is None:
+                continue  # registered via set_state, already wired
+            if (alias, spec.indexer) in self._state_edges:
+                continue
+
+            # Skip if no module provides this channel (e.g. all setters disabled)
+            if not self._delayed_edge_providers.get(spec.source_field):
+                continue
+
+            tstype = self.get_outer_type(spec.source_field)
+            if is_dict_basket(tstype):
+                if spec.indexer is None:
+                    raise NotImplementedError(
+                        f"Annotation-declared state '{alias}' on dict basket '{spec.source_field}' "
+                        f"requires an indexer (set indexer=... in State(...))"
+                    )
+                edge = self.get_channel(spec.source_field, indexer=spec.indexer)
+            else:
+                edge = self.get_channel(spec.source_field)
+            self._wire_state_edge(alias, edge, spec.keyby, spec.indexer)
+
     def _bind_delayed_channel(self, field, list_of_edges_and_modules, indexer=None):
         tstype = self.get_outer_type(field)
         # Make sure a getter node exists first
@@ -602,10 +696,6 @@ def get_channel(
 
         return getattr(self, field)
 
-    @classmethod
-    def is_state_field(cls, field):
-        return field.startswith("s_")
-
     @classmethod
     def get_outer_type(cls, field):
         return cls.model_fields[field].annotation
@@ -619,9 +709,6 @@ def set_channel(
         # add to graph
         self._add_field_to_graph(field, self._module_being_attached, True, indexer)
 
-        # TODO fix ugly state field stuff
-        is_state_field = self.is_state_field(field)
-
         tstype = self.get_outer_type(field)
         if is_dict_basket(tstype):
             _is_dict_basket = True
@@ -668,7 +755,7 @@ def set_channel(
         else:
             edge_tstypes = [edge.tstype]  # type: ignore[union-attr]
 
-        if not all(edge_tstype == gateway_tstype for edge_tstype in edge_tstypes) and not is_state_field:
+        if not all(edge_tstype == gateway_tstype for edge_tstype in edge_tstypes):
             raise TypeError("Edge type incorrect for {}: should be {}, found {}".format(field, gateway_tstype, edge_tstypes[0]))
 
         module = self._module_being_attached
@@ -687,60 +774,146 @@ def set_channel(
         self._set_last(field)
         self._set_next(field)
 
-    def _ensure_state_field(self, field: str) -> str:
-        if not field.startswith("s_"):
-            return "s_{}".format(field)
-        return field
-
     def set_state(
         self,
-        field: str,
+        field_or_edge: Union[Edge, str],
         keyby: Union[str, Tuple[str, ...]],
-        indexer: Union[str, int] = None,
+        indexer: Optional[Union[str, int]] = None,
     ) -> None:
-        # grab state version of field
-        state_field = self._ensure_state_field(field)
-
-        # Bail if already setup
-        if (state_field, indexer) in self._state_requests:
-            return
+        """Register a state collection on a channel.
 
-        # First ensure edge is constructed
-        edge = self.get_channel(field, indexer=indexer)
+        The first argument may be either:
 
-        # And ensure the state edge is constructed
-        self.get_state(state_field, indexer=indexer)
+        - a channel field name (``str``) — the edge is resolved via
+          :meth:`get_channel`, and the state is exposed under that field name;
+        - a csp ``Edge`` previously registered via :meth:`set_channel` — the
+          field name is recovered by reverse lookup of the registered edge.
 
-        if isinstance(edge, Edge):
-            # instantiate state node
-            if get_origin(edge.tstype.typ) is list:
-                edge_type_name = get_args(edge.tstype.typ)[0].__name__
-                state_edge = build_track_state_node(csp.unroll(edge), keyby)
-            else:
-                edge_type_name = edge.tstype.typ.__name__
-                state_edge = build_track_state_node(edge, keyby)
-
-            state_edge.nodedef.__name__ = "State[{}]".format(edge_type_name)
-
-            # register for use inside other csp nodes
-            self.set_channel(state_field, state_edge, indexer=indexer)
+        ``keyby`` and ``indexer`` describe how ticks are accumulated. If the
+        state is already registered, this is a no-op when ``keyby``/``indexer``
+        match; otherwise a ``ValueError`` is raised.
+        """
+        if isinstance(field_or_edge, str):
+            field = field_or_edge
+            edge = self.get_channel(field, indexer=indexer) if indexer is not None else self.get_channel(field)
+        elif isinstance(field_or_edge, Edge):
+            edge = field_or_edge
+            field = self._find_field_for_edge(edge)
+            if field is None:
+                raise ValueError(
+                    "set_state could not resolve a channel name from the given edge; "
+                    "register it via set_channel first, or pass the channel field name (str)."
+                )
+        else:
+            raise TypeError("set_state expects a channel field name (str) or a csp Edge as the first argument; got {}".format(type(field_or_edge)))
+
+        keyby = _normalize_keyby(keyby)
+        existing = self._states.get(field)
+        if existing is not None:
+            if existing.keyby != keyby or existing.indexer != indexer:
+                raise ValueError(
+                    f"State '{field}' already registered with "
+                    f"keyby={existing.keyby!r}, indexer={existing.indexer!r}; "
+                    f"cannot redefine with keyby={keyby!r}, indexer={indexer!r}"
+                )
+            if (field, indexer) in self._state_edges:
+                return  # already wired
+
+        # If this was pre-declared via dynamic_state_channels, it is no longer pending
+        # once set_state is called for it.
+        self._pending_state_element_types.pop(field, None)
+
+        self._states[field] = _StateSpec(source_field=field, keyby=keyby, indexer=indexer)
+        self._wire_state_edge(field, edge, keyby, indexer)
+
+    def _find_field_for_edge(self, edge: Edge) -> Optional[str]:
+        """Reverse-lookup a channel field name for a previously-set edge."""
+        for field, providers in self._delayed_edge_providers.items():
+            for _module, provided in providers:
+                if provided is edge:
+                    return field
+                if isinstance(provided, dict):
+                    for v in provided.values():
+                        if v is edge:
+                            return field
+        return None
 
-            # setup ad-hoc querying
-            trigger = ConcurrentFutureAdapter(name="RequestState<{}>".format(edge_type_name))
+    def _wire_state_edge(
+        self,
+        field: str,
+        edge: Edge,
+        keyby: Union[str, Tuple[str, ...]],
+        indexer: Optional[Union[str, int]],
+    ) -> None:
+        if get_origin(edge.tstype.typ) is list:
+            edge_type_name = get_args(edge.tstype.typ)[0].__name__
+            state_edge = build_track_state_node(csp.unroll(edge), keyby)
+        else:
+            edge_type_name = edge.tstype.typ.__name__
+            state_edge = build_track_state_node(edge, keyby)
+
+        state_edge.nodedef.__name__ = "State[{}]".format(edge_type_name)
+
+        trigger = ConcurrentFutureAdapter(name="RequestState<{}>".format(edge_type_name))
+        named_on_request_node("QueryState<{}>".format(edge_type_name))(state_edge, trigger.out())
+
+        self._state_requests[field, indexer] = trigger
+        # If a DelayedEdge was previously handed out by get_state (because another module
+        # called get_state before this module's set_state), bind it now so the consumer's
+        # edge resolves to the real state node.
+        delayed = self._delayed_state_edges.get((field, indexer))
+        if delayed is not None:
+            delayed.bind(state_edge)
+            self._state_edges[field, indexer] = delayed
+        else:
+            self._state_edges[field, indexer] = state_edge
 
-            named_on_request_node("QueryState<{}>".format(edge_type_name))(state_edge, trigger.out())
+    def _declare_dynamic_state(self, field: str, element_type: type) -> None:
+        """Pre-register a dynamically-created state channel so that :meth:`get_state`
+        for ``field`` returns a :class:`DelayedEdge` before the owning module's
+        ``connect`` calls :meth:`set_state`.
 
-            # register the trigger
-            self._state_requests[state_field, indexer] = trigger
-        else:
-            # TODO
-            raise NotImplementedError()
+        ``element_type`` is the unwrapped element type of the state (i.e. ``T`` for a
+        channel typed as either ``ts[T]`` or ``ts[List[T]]``). The eventual
+        :meth:`set_state` call provides ``keyby``/``indexer`` and binds the real state
+        edge into the previously-handed-out :class:`DelayedEdge`.
+        """
+        if field in self._states or field in self._pending_state_element_types:
+            return
+        self._pending_state_element_types[field] = element_type
 
     def get_state(self, field: str, indexer: Union[str, int] = None) -> Any:
-        # grab state version of field
-        state_field = self._ensure_state_field(field)
-
-        return self.get_channel(state_field, indexer=indexer)
+        """Return the underlying state Edge for ``field`` (csp-graph use)."""
+        if (field, indexer) in self._state_edges:
+            return self._state_edges[field, indexer]
+        if field in self._pending_state_element_types:
+            # Pre-declared by Module.dynamic_state_channels but the owning module has
+            # not yet called set_state. Hand out a DelayedEdge that will be bound once
+            # set_state runs (order-independent across modules).
+            delayed = self._delayed_state_edges.get((field, indexer))
+            if delayed is None:
+                element_type = self._pending_state_element_types[field]
+                delayed = DelayedEdge(ts[State[element_type]])
+                delayed.__name__ = "s_{}".format(field)
+                self._delayed_state_edges[field, indexer] = delayed
+            return delayed
+        if field not in self._states:
+            raise NoProviderException("Unknown state: {}".format(field))
+        spec = self._states[field]
+        if spec.source_field is None:
+            raise NoProviderException("State '{}' (indexer={}) has not been wired yet".format(field, indexer))
+        # Lazily wire annotation-declared state on first access
+        tstype = self.get_outer_type(spec.source_field)
+        if is_dict_basket(tstype):
+            if spec.indexer is None:
+                raise NotImplementedError(
+                    f"Annotation-declared state '{field}' on dict basket '{spec.source_field}' requires an indexer (set indexer=... in State(...))"
+                )
+            edge = self.get_channel(spec.source_field, indexer=spec.indexer)
+        else:
+            edge = self.get_channel(spec.source_field)
+        self._wire_state_edge(field, edge, spec.keyby, spec.indexer)
+        return self._state_edges[field, indexer]
 
     def _set_last(self, field: str, indexer: Union[str, int] = None) -> None:
         # Bail if already setup
@@ -923,17 +1096,13 @@ def next(self, field: str, indexer: Union[str, int] = None, *, timeout=None) ->
         return result
 
     def state(self, field: str, indexer: Union[str, int] = None, *, timeout=None) -> Any:
-        # grab state version of field
-        state_field = self._ensure_state_field(field)
-
-        self._check(state_field, self._state_requests, "state", indexer=indexer)
+        self._check(field, self._state_requests, "state", indexer=indexer)
 
-        # TODO checks for state tracking
         # TODO make sure not called from inside graph context
-        state_edge = self._state_requests[state_field, indexer]
+        trigger = self._state_requests[field, indexer]
 
         # trigger request for state into graph
-        future = state_edge.push_tick()
+        future = trigger.push_tick()
 
         # wait for result
         return future.result(timeout=timeout)
diff --git a/csp_gateway/server/gateway/csp/factory.py b/csp_gateway/server/gateway/csp/factory.py
index 5ba3fa7..c5017df 100644
--- a/csp_gateway/server/gateway/csp/factory.py
+++ b/csp_gateway/server/gateway/csp/factory.py
@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Generic, List, Optional, Type
+from typing import Generic, List, Optional, Type, get_args, get_origin
 
 from ccflow import BaseModel
 from csp.impl.enum import Enum
@@ -57,6 +57,24 @@ def build(self, channels: ChannelsType) -> ChannelsType:
         if self.block_set_channels_until is not None:
             channels._block_set_channels_until = self.block_set_channels_until
 
+        # Pre-declare dynamically-created state channels so that modules calling get_state
+        # during connect() are independent of the order in which set_state is called by the
+        # owning module. The owning module includes the name in both dynamic_channels()
+        # (which provides the element type) and dynamic_state_channels().
+        for node in enabled_modules:
+            declared = node.dynamic_state_channels() if hasattr(node, "dynamic_state_channels") else None
+            if not declared:
+                continue
+            owned_channels = node.dynamic_channels() or {}
+            for name in declared:
+                if name not in owned_channels:
+                    raise ValueError(
+                        f"Module {type(node).__name__} declared '{name}' in dynamic_state_channels() but did not include it in dynamic_channels()"
+                    )
+                t = owned_channels[name]
+                element_type = get_args(t)[0] if get_origin(t) is list else t
+                channels._declare_dynamic_state(name, element_type)
+
         # Wire in each edge Provider.
         # The implementation of set_channel will handle multiplexing streams
         for node in enabled_modules:
diff --git a/csp_gateway/server/gateway/csp/module.py b/csp_gateway/server/gateway/csp/module.py
index e434573..e9746aa 100644
--- a/csp_gateway/server/gateway/csp/module.py
+++ b/csp_gateway/server/gateway/csp/module.py
@@ -49,8 +49,13 @@ def dynamic_channels(self) -> Optional[Dict[str, Union[Type[GatewayStruct], Type
         ...
 
     def dynamic_state_channels(self) -> Optional[Set[str]]:
-        """
-        The set of dynamic channels that have state.
+        """The subset of :meth:`dynamic_channels` for which this module will also call
+        :meth:`GatewayChannels.set_state` from within :meth:`connect`.
+
+        Declaring a name here lets *other* modules call :meth:`GatewayChannels.get_state`
+        on the channel from their own ``connect`` regardless of the order in which modules
+        are connected; the returned state edge is bound to the real state node once the
+        owning module's ``connect`` calls ``set_state``.
         """
         ...
 
diff --git a/csp_gateway/server/gateway/csp/state.py b/csp_gateway/server/gateway/csp/state.py
index 43dd2c8..557035d 100644
--- a/csp_gateway/server/gateway/csp/state.py
+++ b/csp_gateway/server/gateway/csp/state.py
@@ -7,7 +7,7 @@
 from enum import Enum as PyEnum
 from functools import lru_cache
 from pprint import pformat
-from typing import Any, Deque, Dict, List, Tuple, Union
+from typing import Any, Deque, Dict, List, Optional, Tuple, Union
 
 import csp
 import duckdb
@@ -75,6 +75,16 @@ def disable_duckdb_state() -> None:
     _USE_DUCKDB_STATE = False
 
 
+def _resolve_keyby_attr(record: Any, path: str) -> Any:
+    """Resolve a (possibly dotted) attribute path on ``record``, returning ``None`` if any segment is missing."""
+    obj = record
+    for segment in path.split("."):
+        if obj is None:
+            return None
+        obj = getattr(obj, segment, None)
+    return obj
+
+
 class StateType(CoreEnum):
     UNKNOWN = 0
     DEFAULT = 1
@@ -149,7 +159,7 @@ def insert(self, record: Any) -> None:
 
         for subkey in self._keyby:
             # extract the key from the record
-            subkey_to_use = getattr(record, subkey, None)
+            subkey_to_use = _resolve_keyby_attr(record, subkey)
 
             if subkey == self._keyby[-1]:
                 # Put the element there if last
@@ -408,7 +418,7 @@ def insert(self, record: Struct) -> None:
             obj_id = None
             for subkey in self._keyby:
                 # extract the key from the record
-                subkey_to_use = getattr(record, subkey, None)
+                subkey_to_use = _resolve_keyby_attr(record, subkey)
 
                 if subkey == self._keyby[-1]:
                     if subkey_to_use not in place.keys():
@@ -527,23 +537,45 @@ def get_duckdb_schema_struct(cls: Struct) -> Tuple[Dict, bool]:
     return (new_type_info, use_duckdb)
 
 
-# NOTE: NEVER access State object directly, always access through the __class_getitem__ API
+# NOTE: For runtime state instances, always use the State[<typ>] API.
+# State() called directly (with no parameterized typ) is the annotation
+# marker form used in `Annotated[ts[X], State(keyby=..., indexer=..., alias=...)]`.
 class State(BaseState):
-    def __init__(self, keyby: Union[Tuple[str, ...], str] = ("id",)) -> None:
-        """Switch case between different state specializations based on the type of the records"""
+    # Annotation metadata. Set on every instance; consumed by ChannelsMetaclass
+    # when this State is found in a field's Annotated metadata.
+    _meta_keyby: Union[Tuple[str, ...], str] = ("id",)
+    _meta_indexer: Optional[Union[str, int]] = None
+    _meta_alias: Optional[str] = None
+
+    def __init__(
+        self,
+        keyby: Union[Tuple[str, ...], str] = ("id",),
+        indexer: Optional[Union[str, int]] = None,
+        alias: Optional[str] = None,
+    ) -> None:
+        """Initialize a State.
+
+        When ``self._typ`` is set (via ``State[T](...)``), this constructs a runtime
+        state collection and dispatches to the appropriate backend. Otherwise the
+        instance is treated as an annotation marker and only retains its metadata.
+        """
+        # Always retain annotation metadata.
+        self._meta_keyby = keyby
+        self._meta_indexer = indexer
+        self._meta_alias = alias
+
+        typ = getattr(self, "_typ", None)
+        if typ is None:
+            # Annotation-marker form; no runtime backend needed.
+            return
 
         global _USE_DUCKDB_STATE
-        try:
-            typ = self._typ
-            if _USE_DUCKDB_STATE and isinstance(typ, type) and issubclass(typ, Struct):
-                schema, use_duckdb = get_duckdb_schema_struct(typ)
-                if use_duckdb:
-                    self._state_impl = DuckDBState(typ, keyby, schema)
-                    self._state_type = StateType.DUCKDB
-                    return
-        except AttributeError:
-            log.warning("Do not create object directly from State use the State[<typ>] API instead for performance reasons")
-            log.warning("Using DefaultStateClass")
+        if _USE_DUCKDB_STATE and isinstance(typ, type) and issubclass(typ, Struct):
+            schema, use_duckdb = get_duckdb_schema_struct(typ)
+            if use_duckdb:
+                self._state_impl = DuckDBState(typ, keyby, schema)
+                self._state_type = StateType.DUCKDB
+                return
         self._state_impl = DefaultState(keyby)
         self._state_type = StateType.DEFAULT
 
diff --git a/csp_gateway/server/gateway/gateway.py b/csp_gateway/server/gateway/gateway.py
index 9bab60a..ea7b4eb 100644
--- a/csp_gateway/server/gateway/gateway.py
+++ b/csp_gateway/server/gateway/gateway.py
@@ -6,13 +6,12 @@
 from datetime import datetime, timedelta
 from socket import gethostname
 from time import sleep
-from typing import Any, Callable, List, Optional, Type, Union, get_args, get_origin
+from typing import Any, Callable, List, Optional, Type, Union
 
 import csp
 from csp import ts
 from pydantic import Field, PrivateAttr, create_model, model_validator
 
-from csp_gateway.server.gateway import State
 from csp_gateway.server.settings import Settings
 
 from .csp import Channels, ChannelsFactory, ChannelsType, Module
@@ -113,7 +112,6 @@ def _instantiate_dynamic_channel(self, modules: List[Module[GatewayChannels]], c
         for m in modules:
             module_dynamic_channels = m.dynamic_channels() if hasattr(m, "dynamic_channels") else None
             if module_dynamic_channels:
-                channels_with_state = m.dynamic_state_channels() if hasattr(m, "dynamic_state_channels") else None
                 for n, t in module_dynamic_channels.items():
                     existing_type = dynamic_channels.get(n, None)
                     if existing_type is not None:
@@ -121,11 +119,6 @@ def _instantiate_dynamic_channel(self, modules: List[Module[GatewayChannels]], c
                             raise ValueError(f"Conflicting types for dynamic channel {n}.")
 
                     dynamic_channels[n] = t
-                    if channels_with_state and n in channels_with_state:
-                        if get_origin(t) is list:
-                            t = get_args(t)[0]
-
-                        dynamic_channels[f"s_{n}"] = State[t]
 
         if dynamic_channels:
             dynamic_channel_kwargs = {n: (ts[t], None) for n, t in dynamic_channels.items()}
diff --git a/csp_gateway/server/modules/logging/stdlib.py b/csp_gateway/server/modules/logging/stdlib.py
index ec55b40..59d56af 100644
--- a/csp_gateway/server/modules/logging/stdlib.py
+++ b/csp_gateway/server/modules/logging/stdlib.py
@@ -383,7 +383,7 @@ def _convert_log_level(cls, v: Union[str, int]) -> int:
     def connect(self, channels: ChannelsType):
         logger_to_use = logging.getLogger(self.log_name)
 
-        for field in self.selection.select_from(channels, state_channels=self.log_states):
+        for field in self.selection.select_from(channels):
             data = channels.get_channel(field)
             # list baskets not supported yet
             if isinstance(data, dict):
@@ -393,6 +393,11 @@ def connect(self, channels: ChannelsType):
                 edge = channels.get_channel(field)
                 csp.log(self.log_level, field, edge, logger=logger_to_use)
 
+        if self.log_states:
+            for alias in self.selection.select_from(channels, state_channels=True):
+                edge = channels.get_state(alias)
+                csp.log(self.log_level, f"s_{alias}", edge, logger=logger_to_use)
+
 
 # Backwards compatibility alias
 StdlibLogging = Logging
diff --git a/csp_gateway/server/modules/web/mount.py b/csp_gateway/server/modules/web/mount.py
index a28e37a..dca6a2f 100644
--- a/csp_gateway/server/modules/web/mount.py
+++ b/csp_gateway/server/modules/web/mount.py
@@ -91,7 +91,8 @@ def _mount_send(self, app: GatewayWebApp) -> None:
 
     def _mount_state(self, app: GatewayWebApp) -> None:
         selection = ChannelSelection() if self.force_mount_all else self.mount_state
-        channels_set = set(selection.select_from(app.gateway.channels_model, state_channels=True))
+        # Use the instance so dynamically-registered set_state aliases are included.
+        channels_set = set(selection.select_from(app.gateway.channels, state_channels=True))
         seen_channels = set()
 
         # Bind every wire
@@ -110,7 +111,7 @@ def _mount_state(self, app: GatewayWebApp) -> None:
 
             missing_channels = channels_set - seen_channels
             if missing_channels:
-                log.info(f"Requested channels missing state routes: {list(channel[2:] for channel in missing_channels)}")
+                log.info(f"Requested channels missing state routes: {list(missing_channels)}")
 
         app.add_state_available_channels(seen_channels)
 
diff --git a/csp_gateway/server/shared/channel_selection.py b/csp_gateway/server/shared/channel_selection.py
index 1191f62..89bba44 100644
--- a/csp_gateway/server/shared/channel_selection.py
+++ b/csp_gateway/server/shared/channel_selection.py
@@ -68,21 +68,24 @@ def select_from(
             fields = channels.fields() if self.include is None else self.include
             return list(dict.fromkeys([field for field in fields if field not in self.exclude]))
 
+        # State aliases live in their own namespace alongside channel fields.
+        if state_channels:
+            if isinstance(channels, type):
+                # called with the Channels class (declared aliases only)
+                aliases = list(channels._declared_states.keys())
+            else:
+                aliases = channels.all_state_aliases()
+            if self.include is not None:
+                ordered = [a for a in self.include if a in aliases and a not in self.exclude]
+            else:
+                ordered = [a for a in aliases if a not in self.exclude]
+            return list(dict.fromkeys(ordered))
+
         for idx, field in enumerate(channels.fields()):
             # avoid duplicates
             if field in names:
                 continue
 
-            # TODO not this `s_` business...
-            # Return state channels or regular channels
-            if state_field := field.startswith("s_"):
-                if not state_channels:
-                    continue
-                field = field[2:]
-            else:
-                if state_channels:
-                    continue
-
             # Check whether static
             outer_type = channels.get_outer_type(field)
             if is_dict_basket(outer_type) or isTsType(outer_type):
@@ -103,9 +106,6 @@ def select_from(
             if field in self.exclude:
                 continue
 
-            if state_field:
-                field = f"s_{field}"
-
             names[field] = idx
 
         return [k for k, _ in sorted(names.items(), key=lambda x: x[1])]
diff --git a/csp_gateway/server/web/app.py b/csp_gateway/server/web/app.py
index 524e902..d89920c 100644
--- a/csp_gateway/server/web/app.py
+++ b/csp_gateway/server/web/app.py
@@ -382,21 +382,45 @@ def add_send_available_channels(self, fields: Optional[Set[str]] = None) -> None
         add_send_available_channels(api_router=api_router, fields=fields)
 
     def add_state_api(self, field: str) -> None:
-        api_router = self.get_router("state")
-
-        # Prune s_ from start
-        name_without_state = field[2:]
+        """Mount REST routes for the given state ``field``.
 
-        dict_basket = self._is_dict_basket_field(field=name_without_state)
+        ``field`` must be a known state name on the gateway's channels — either
+        declared via ``Annotated[..., State(...)]`` or registered dynamically
+        via ``set_state`` during a module's ``connect``.
+        """
+        api_router = self.get_router("state")
 
-        if dict_basket:
-            dict_basket_key_type, model = dict_basket
-            subroute_key = dict_basket_key_type
+        spec = self.gateway.channels._states.get(field) or self.gateway.channels_model._declared_states.get(field)
+        if spec is None:
+            raise ValueError("Unknown state '{}' on {}".format(field, self.gateway.channels_model.__name__))
+
+        if spec.source_field is not None:
+            dict_basket = self._is_dict_basket_field(field=spec.source_field)
+            if dict_basket:
+                dict_basket_key_type, model = dict_basket
+                # If the annotation pinned an indexer, expose as a non-keyed route on that one key.
+                subroute_key = None if spec.indexer is not None else dict_basket_key_type
+            else:
+                model = self._get_field_pydantic_type(spec.source_field)
+                subroute_key = None
         else:
-            model = self._get_field_pydantic_type(name_without_state)
+            # set_state-registered: source is a raw edge with a known type.
+            state_edge = self.gateway.channels._state_edges.get((field, spec.indexer))
+            model = None
             subroute_key = None
-
-        add_state_routes(api_router=api_router, field=field, model=model, subroute_key=subroute_key)
+            if state_edge is not None:
+                inner = state_edge.tstype.typ
+                # Unwrap State[T] -> T for the response model
+                model = getattr(inner, "_typ", inner)
+
+        add_state_routes(
+            api_router=api_router,
+            field=field,
+            model=model,
+            subroute_key=subroute_key,
+            keyby=tuple(spec.keyby) if spec.keyby else (),
+            indexer=spec.indexer,
+        )
 
     def add_state_available_channels(self, fields: Optional[Set[str]] = None) -> None:
         api_router = self.get_router("state")
diff --git a/csp_gateway/server/web/routes/state.py b/csp_gateway/server/web/routes/state.py
index 6d64dc5..88cc20a 100644
--- a/csp_gateway/server/web/routes/state.py
+++ b/csp_gateway/server/web/routes/state.py
@@ -1,4 +1,5 @@
-from typing import Any, List, Optional, Set, Union, get_origin
+from inspect import cleandoc
+from typing import Any, List, Optional, Set, Tuple, Union, get_origin
 
 from fastapi import APIRouter, HTTPException, Request
 from pydantic import BaseModel
@@ -20,101 +21,69 @@ def add_state_routes(
     field: str = "",
     model: Union[BaseModel, List[BaseModel]] = None,
     subroute_key: Any = None,
+    keyby: Tuple[str, ...] = (),
+    indexer: Optional[Union[str, int]] = None,
 ) -> None:
+    """Mount REST routes for a single state ``field``.
+
+    The state mechanism is now name-based; there is no implicit ``s_`` prefix.
+    Routes are exposed at ``/state/<field>`` (and a variant with ``_``
+    replaced by ``-`` for URL friendliness).
+
+    ``keyby`` / ``indexer`` are surfaced in the route description so the
+    accumulation semantics are visible in the OpenAPI schema.
+    """
     if model and get_origin(model) is list:
         list_model = model
     else:
         list_model = List[model]
 
-    # Get the fully qualified type name for the description
     fq_type_name = get_fully_qualified_type_name(model)
 
+    keyby_lines = []
+    if keyby:
+        keyby_lines.append("**Keyed by:** ``{}``".format(", ".join(repr(k) for k in keyby)))
+    if indexer is not None:
+        keyby_lines.append("**Indexer:** ``{!r}``".format(indexer))
+    keyby_prefix = ("\n\n".join(keyby_lines) + "\n\n") if keyby_lines else ""
+
     if subroute_key:
-        # Prune s_ from start
-        name_without_state = field[2:]
 
         async def get_state(key: str, query: Optional[Query] = query_json(), request: Request = None) -> list_model:  # type: ignore[valid-type]
             """
             Get state value on a dictionary basket channel, where `key` is the key of the dictionary basket.
             If such a key does not exist or is not mounted, this endpoint will raise a `404` error.
 
-            States may be queried by certain conditions. Currently only filtering is supported. Filters can be used to evaluate
-            objects in state and compare them to either scalar values or other attributes on the object. Here are some simple examples
-            using the demo application provided with `csp-gateway`:
-
-            ```
-            # Filter only records where `record.x` == 5
-            api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":5,"where":"=="}}]}
-
-
-            # Filter only records where `record.x` < 10
-            /api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":10,"where":"<"}}]}
-
-            # Filter only records where `record.timestamp` < "2023-03-30T14:45:26.394000"
-            /api/v1/state/example?query={"filters":[{"attr":"timestamp","by":{"when":"2023-03-30T14:45:26.394000","where":"<"}}]}
-
-            # Filter only records where `record.id` < `record.y`
-            /api/v1/state/example?query={"filters":[{"attr":"id","by":{"attr":"y","where":"<"}}]}
-
-            # Filter only records where `record.x` < 50 and `record.x` >= 30
-            /api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":50,"where":"<"}},{"attr":"x","by":{"value":30,"where":">="}}]}
-            ```
+            States may be queried by certain conditions. See `/state/<field>` for query examples.
             """
-            channel = request.app.gateway.channels._ensure_state_field(field)
-
-            if not hasattr(request.app.gateway.channels, channel):
-                raise HTTPException(
-                    status_code=404,
-                    detail="State channel not found: {}".format(channel),
-                )
-
-            # FIXME this wont work, is dicts
             try:
-                res = request.app.gateway.channels.query(
-                    getattr(request.app.gateway.channels_model, channel),
-                    key,
-                    query=query,
-                )
+                res = request.app.gateway.channels.query(field, key, query=query)
             except NoProviderException:
                 raise HTTPException(
                     status_code=404,
-                    detail="Channel not found: {}/{}".format(field, key),
+                    detail="State not found: {}/{}".format(field, key),
                 )
 
             return prepare_response(res, is_list_model=True)
 
-        api_router.get(
-            "/{}/{{key:path}}".format(name_without_state),
-            responses=get_default_responses(),
-            response_model=list_model,  # type: ignore[valid-type]
-            name="Get State {} by key".format(name_without_state),
-            openapi_extra={"type_": fq_type_name} if fq_type_name else None,
-        )(get_state)
-
-        api_router.get(
-            "/{}/{{key:path}}".format(name_without_state.replace("_", "-")),
-            responses=get_default_responses(),
-            response_model=list_model,  # type: ignore[valid-type]
-            include_in_schema=False,
-        )(get_state)
-
         api_router.get(
             "/{}/{{key:path}}".format(field),
             responses=get_default_responses(),
             response_model=list_model,  # type: ignore[valid-type]
-            include_in_schema=False,
+            name="Get State {} by key".format(field),
+            description=keyby_prefix + cleandoc(get_state.__doc__ or ""),
+            openapi_extra={"type_": fq_type_name} if fq_type_name else None,
         )(get_state)
 
-        api_router.get(
-            "/{}/{{key:path}}".format(field.replace("_", "-")),
-            responses=get_default_responses(),
-            response_model=list_model,  # type: ignore[valid-type]
-            include_in_schema=False,
-        )(get_state)
+        if "_" in field:
+            api_router.get(
+                "/{}/{{key:path}}".format(field.replace("_", "-")),
+                responses=get_default_responses(),
+                response_model=list_model,  # type: ignore[valid-type]
+                include_in_schema=False,
+            )(get_state)
 
     if model:
-        # Prune s_ from start
-        name_without_state = field[2:]
 
         async def get_state(query: Optional[Query] = query_json(), request: Request = None) -> list_model:  # type: ignore[misc, valid-type]
             """Get state value on a non-dict basket channel. This endpoint will flatten the state structure and return a list of the elements.
@@ -126,70 +95,47 @@ async def get_state(query: Optional[Query] = query_json(), request: Request = No
 
             ```
             # Filter only records where `record.x` == 5
-            api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":5,"where":"=="}}]}
+            api/v1/state/example_with_state?query={"filters":[{"attr":"x","by":{"value":5,"where":"=="}}]}
 
             # Filter only records where `record.x` < 10
-            /api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":10,"where":"<"}}]}
+            /api/v1/state/example_with_state?query={"filters":[{"attr":"x","by":{"value":10,"where":"<"}}]}
 
             # Filter only records where `record.timestamp` < "2023-03-30T14:45:26.394000"
-            /api/v1/state/example?query={"filters":[{"attr":"timestamp","by":{"when":"2023-03-30T14:45:26.394000","where":"<"}}]}
+            /api/v1/state/example_with_state?query={"filters":[{"attr":"timestamp","by":{"when":"2023-03-30T14:45:26.394000","where":"<"}}]}
 
             # Filter only records where `record.id` < `record.y`
-            /api/v1/state/example?query={"filters":[{"attr":"id","by":{"attr":"y","where":"<"}}]}
+            /api/v1/state/example_with_state?query={"filters":[{"attr":"id","by":{"attr":"y","where":"<"}}]}
 
             # Filter only records where `record.x` < 50 and `record.x` >= 30
-            /api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":50,"where":"<"}},{"attr":"x","by":{"value":30,"where":">="}}]}
+            /api/v1/state/example_with_state?query={"filters":[{"attr":"x","by":{"value":50,"where":"<"}},{"attr":"x","by":{"value":30,"where":">="}}]}
             ```
             """
-            channel = request.app.gateway.channels._ensure_state_field(field)
-
-            if not hasattr(request.app.gateway.channels, channel):
-                raise HTTPException(
-                    status_code=404,
-                    detail="State channel not found: {}".format(channel),
-                )
-
             try:
-                res = request.app.gateway.channels.query(
-                    getattr(request.app.gateway.channels_model, channel),
-                    query=query,
-                )
+                res = request.app.gateway.channels.query(field, query=query)
             except NoProviderException:
                 raise HTTPException(
                     status_code=404,
-                    detail="Channel not found: {}".format(channel),
+                    detail="State not found: {}".format(field),
                 )
 
             return prepare_response(res, is_list_model=True)
 
-        api_router.get(
-            "/{}".format(name_without_state),
-            responses=get_default_responses(),
-            response_model=list_model,  # type: ignore[valid-type]
-            name="Get State {}".format(name_without_state),
-            openapi_extra={"type_": fq_type_name} if fq_type_name else None,
-        )(get_state)
-
-        api_router.get(
-            "/{}".format(name_without_state.replace("_", "-")),
-            responses=get_default_responses(),
-            response_model=list_model,  # type: ignore[valid-type]
-            include_in_schema=False,
-        )(get_state)
-
         api_router.get(
             "/{}".format(field),
             responses=get_default_responses(),
             response_model=list_model,  # type: ignore[valid-type]
-            include_in_schema=False,
+            name="Get State {}".format(field),
+            description=keyby_prefix + cleandoc(get_state.__doc__ or ""),
+            openapi_extra={"type_": fq_type_name} if fq_type_name else None,
         )(get_state)
 
-        api_router.get(
-            "/{}".format(field.replace("_", "-")),
-            responses=get_default_responses(),
-            response_model=list_model,  # type: ignore[valid-type]
-            include_in_schema=False,
-        )(get_state)
+        if "_" in field:
+            api_router.get(
+                "/{}".format(field.replace("_", "-")),
+                responses=get_default_responses(),
+                response_model=list_model,  # type: ignore[valid-type]
+                include_in_schema=False,
+            )(get_state)
 
 
 def add_state_available_channels(
@@ -203,13 +149,10 @@ def add_state_available_channels(
     )
     async def get_state(request: Request) -> List[str]:
         """
-        This endpoint will return a list of string values of all available state channels under the `/state` route.
+        Return the list of all available state names under the `/state` route.
 
-        Note: This endpoint does not support filtering
+        Note: This endpoint does not support filtering.
         """
-        # TODO: Change state channel stuff
         return sorted(
-            field[2:]
-            for field in ChannelSelection().select_from(request.app.gateway.channels, state_channels=True)
-            if fields is None or field in fields
+            name for name in ChannelSelection().select_from(request.app.gateway.channels, state_channels=True) if fields is None or name in fields
         )
diff --git a/csp_gateway/server/web/templates/channels_graph.html.j2 b/csp_gateway/server/web/templates/channels_graph.html.j2
index 6b6d9b6..ae61fdf 100644
--- a/csp_gateway/server/web/templates/channels_graph.html.j2
+++ b/csp_gateway/server/web/templates/channels_graph.html.j2
@@ -9,7 +9,7 @@
                 "dagre-d3-es": "https://cdn.jsdelivr.net/npm/dagre-d3-es/+esm"
             }
         }
-    </script>    
+    </script>
     <script type="module" src="https://cdn.jsdelivr.net/npm/d3@7/+esm"></script>
     <script type="module" src="https://cdn.jsdelivr.net/npm/lodash-es/+esm"></script>
     <script type="module" src="https://cdn.jsdelivr.net/npm/dagre-d3-es/+esm"></script>
@@ -35,7 +35,7 @@
         top: 0;
         display: flex;
     }
-    
+
     div.dagred3 svg {
         position: absolute;
         left: 0;
@@ -44,7 +44,7 @@
         top: 0;
         display: flex;
     }
-    
+
     .dagred3 .node rect,
     .dagred3 .node circle,
     .dagred3 .node ellipse,
@@ -52,21 +52,21 @@
         stroke: #555;
         fill: #fff;
     }
-    
+
     .dagred3 .edgePath path {
         stroke: #555;
         fill: transparent;
         stroke-width: 1.5px;
     }
-    
+
     .dagred3 .edgeLabel foreignObject {
         overflow: visible;
     }
-    
+
     .dagred3 .edgeLabel foreignObject u {
         text-decoration: none;
     }
-    
+
     .dagred3 .node text {
         pointer-events: none;
     }
@@ -74,27 +74,24 @@
 <script type="module">
     import {graphlib, render} from 'dagre-d3-es'
     import * as d3 from "d3";
-    
+
     const data = JSON.parse('{{ channels_graph|safe }}');
     const div = document.querySelector("div.dagred3");
     const el = d3.select(div);
-    
+
     const svg = el.append("svg");
     svg.attr("height", "100%");
     svg.attr("width", "100%");
-    
+
     const inner = svg.append("g");
     inner.attr("height", "100%");
     inner.attr("width", "100%");
-    
+
     const graph = new graphlib.Graph({ directed: true }).setGraph({height: "100%", width: "100%"});
     const renderer = new render();
-    
+
     // render
     Object.keys(data).forEach((field) => {
-        if(field.startsWith("s_")) {
-            return;
-        }
         graph.setNode(field, {label: field, shape: "diamond"})
         data[field].getters.forEach((getter) => {
             graph.setNode(getter, {label: getter});
@@ -105,7 +102,7 @@
                 arrowheadStyle: "fill: #666;",
                 // style: "stroke: #66f; stroke-width: 3px; stroke-dasharray: 5, 5;",
                 // arrowheadStyle: "fill: #66f",
-                curve: d3.curveBasis 
+                curve: d3.curveBasis
             });
         });
         data[field].setters.forEach((setter) => {
@@ -115,7 +112,7 @@
                 style: "stroke: #f66; stroke-width: 2px;",
                 arrowhead: "vee",
                 arrowheadStyle: "fill: #f66",
-                curve: d3.curveBasis 
+                curve: d3.curveBasis
             });
         })
     });
@@ -125,16 +122,16 @@
         inner.attr("transform", event.transform);
     });
     svg.call(zoom);
-        
+
     // Center the graph
     const initialScale = 0.5;
     const svgWidth = svg.offsetWidth || window.innerWidth;
     const windowWidth = div.offsetWidth || window.innerWidth;
     svg.call(zoom.transform, d3.zoomIdentity.translate(10, window.innerHeight / 4).scale(initialScale));
-        
+
     // svg.attr("height", (div.innerHeight || 0) * initialScale + 40);
-        
+
     renderer(inner, graph);
-        
+
     </script>
     </html>
\ No newline at end of file
diff --git a/csp_gateway/testing/harness.py b/csp_gateway/testing/harness.py
index e4f3204..c0f98c5 100644
--- a/csp_gateway/testing/harness.py
+++ b/csp_gateway/testing/harness.py
@@ -418,10 +418,7 @@ def _run_test_events(self, event: ts[BaseGatewayTestEvent], channel_values: Dict
                 if channel not in s_values:
                     s_values[channel] = []
 
-                if isinstance(channel, str) and channel.startswith("s_"):
-                    s_values[channel].append((csp.now(), value.query()))
-                else:
-                    s_values[channel].append((csp.now(), value))
+                s_values[channel].append((csp.now(), value))
 
         if csp.ticked(event):
             try:
diff --git a/csp_gateway/testing/shared_helpful_classes.py b/csp_gateway/testing/shared_helpful_classes.py
index 729da8c..060115b 100644
--- a/csp_gateway/testing/shared_helpful_classes.py
+++ b/csp_gateway/testing/shared_helpful_classes.py
@@ -1,5 +1,5 @@
 from datetime import timedelta
-from typing import Dict, List, Optional, Type
+from typing import Annotated, Dict, List, Optional, Type
 
 import csp
 import numpy as np
@@ -47,10 +47,8 @@ class MyGatewayChannels(GatewayChannels):
     my_static: float = 0.0
     my_static_dict: Dict[str, float] = {}
     my_static_list: List[str] = []
-    my_channel: ts[MyStruct] = None
-    s_my_channel: ts[State[MyStruct]] = None
-    my_list_channel: ts[List[MyStruct]] = None
-    s_my_list_channel: ts[State[MyStruct]] = None
+    my_channel: Annotated[ts[MyStruct], State(keyby="id")] = None
+    my_list_channel: Annotated[ts[List[MyStruct]], State(keyby="id")] = None
     my_enum_basket: Dict[MyEnum, ts[MyStruct]] = None
     my_str_basket: Dict[str, ts[MyStruct]] = None
     my_enum_basket_list: Dict[MyEnum, ts[List[MyStruct]]] = None
@@ -58,8 +56,7 @@ class MyGatewayChannels(GatewayChannels):
 
 
 class MySmallGatewayChannels(GatewayChannels):
-    example: ts[int] = None
-    s_example: ts[State[int]] = None
+    example: Annotated[ts[int], State(keyby="id")] = None
     my_str_basket: Dict[str, ts[float]] = None
 
 
@@ -88,9 +85,7 @@ def connect(self, channels: MyGatewayChannels) -> None:
             self.my_list_data,
         )
         channels.add_send_channel(MyGatewayChannels.my_channel)
-        channels.set_state(MyGatewayChannels.my_channel, "id")
         channels.add_send_channel(MyGatewayChannels.my_list_channel)
-        channels.set_state(MyGatewayChannels.my_list_channel, "id")
 
         channels.set_channel(MyGatewayChannels.my_array_channel, csp.const(np.array([1.0, 2.0])))
 
@@ -180,7 +175,6 @@ def connect(self, channels: MySmallGatewayChannels):
             self.subscribe(csp.timer(interval=self.interval, value=True)),
         )
         channels.add_send_channel(MySmallGatewayChannels.example)
-        channels.set_state(MySmallGatewayChannels.example, "id")
 
 
 class MyDictBasketModule(GatewayModule):
diff --git a/csp_gateway/tests/server/gateway/csp/test_channels.py b/csp_gateway/tests/server/gateway/csp/test_channels.py
index d312f90..c222b92 100644
--- a/csp_gateway/tests/server/gateway/csp/test_channels.py
+++ b/csp_gateway/tests/server/gateway/csp/test_channels.py
@@ -1,5 +1,5 @@
 from datetime import date, datetime
-from typing import Dict, List, Optional, get_type_hints
+from typing import Annotated, Dict, List, Optional, get_type_hints
 
 import csp
 import numpy as np
@@ -58,10 +58,8 @@ class MyGatewayChannels(GatewayChannels):
     my_static: float = 0.0
     my_static_dict: Dict[str, float] = {}
     my_static_list: List[str] = []
-    my_channel: ts[MyStruct] = None
-    s_my_channel: ts[State[MyStruct]] = None
-    my_list_channel: ts[List[MyStruct]] = None
-    s_my_list_channel: ts[State[MyStruct]] = None
+    my_channel: Annotated[ts[MyStruct], State(keyby="id")] = None
+    my_list_channel: Annotated[ts[List[MyStruct]], State(keyby="id")] = None
     my_enum_basket: Dict[MyEnum, ts[MyStruct]] = None
     my_str_basket: Dict[str, ts[MyStruct]] = None
     my_enum_basket_list: Dict[MyEnum, ts[List[MyStruct]]] = None
diff --git a/csp_gateway/tests/server/gateway/csp/test_state.py b/csp_gateway/tests/server/gateway/csp/test_state.py
index 61c1d1b..4773f8c 100644
--- a/csp_gateway/tests/server/gateway/csp/test_state.py
+++ b/csp_gateway/tests/server/gateway/csp/test_state.py
@@ -94,6 +94,22 @@ def test_state_keyby_unset():
         assert res == [ts]
 
 
+def test_state_keyby_dot_path():
+    for specialization, struct_type in zip(_STATES, _STRUCTS):
+        s = State[struct_type](keyby="g.suba")
+        assert s.state_type() == specialization
+        ts1 = struct_type(a=0, b="hello", g=CspSubStruct(suba=1))
+        ts2 = struct_type(a=1, b="world", g=CspSubStruct(suba=2))
+        s.insert(ts1)
+        s.insert(ts2)
+        assert sorted(s.query(), key=lambda r: r.g.suba) == [ts1, ts2]
+
+        # Overwriting with same nested key replaces the record.
+        ts3 = struct_type(a=2, b="again", g=CspSubStruct(suba=1))
+        s.insert(ts3)
+        assert sorted(s.query(), key=lambda r: r.g.suba) == [ts3, ts2]
+
+
 def test_state_keyby_two():
     for specialization, struct_type in zip(_STATES, _STRUCTS):
         s = State[struct_type](keyby=("a", "b"))
diff --git a/csp_gateway/tests/server/gateway/test_gateway.py b/csp_gateway/tests/server/gateway/test_gateway.py
index 51c2656..dd28439 100644
--- a/csp_gateway/tests/server/gateway/test_gateway.py
+++ b/csp_gateway/tests/server/gateway/test_gateway.py
@@ -3,7 +3,7 @@
 import time
 from datetime import datetime, timedelta
 from io import StringIO
-from typing import Any, Callable, Dict, List, Optional, Set, Type, Union
+from typing import Annotated, Any, Callable, Dict, List, Optional, Set, Type, Union
 
 import csp
 import numpy as np
@@ -50,10 +50,8 @@ class MyGatewayChannels(GatewayChannels):
     my_static_dict: Dict[str, float] = {}
     my_static_list: List[str] = []
     my_static_dict_of_objects: Dict[str, Any] = {}
-    my_channel: ts[MyStruct] = None
-    s_my_channel: ts[State[MyStruct]] = None
-    my_list_channel: ts[List[MyStruct]] = None
-    s_my_list_channel: ts[State[MyStruct]] = None
+    my_channel: Annotated[ts[MyStruct], State(keyby="id")] = None
+    my_list_channel: Annotated[ts[List[MyStruct]], State(keyby="id")] = None
     my_enum_basket: Dict[MyEnum, ts[MyStruct]] = None
     my_str_basket: Dict[str, ts[MyStruct]] = None
     my_enum_basket_list: Dict[MyEnum, ts[List[MyStruct]]] = None
@@ -84,9 +82,7 @@ def connect(self, channels: MyGatewayChannels) -> None:
             self.my_list_data,
         )
         channels.add_send_channel(MyGatewayChannels.my_channel)
-        channels.set_state(MyGatewayChannels.my_channel, "id")
         channels.add_send_channel(MyGatewayChannels.my_list_channel)
-        channels.set_state(MyGatewayChannels.my_list_channel, "id")
 
         channels.set_channel(MyGatewayChannels.my_array_channel, csp.const(np.array([1.0, 2.0])))
 
@@ -766,14 +762,10 @@ def dynamic_state_channels(self) -> Optional[Set[str]]:
         return {self.scalar_channel_name, self.list_channel_name}
 
     def connect(self, channels: MyGatewayChannels) -> None:
-        channels.set_channel(
-            self.scalar_channel_name,
-            csp.const(MyStruct(foo=1.0)),
-        )
-        channels.set_channel(
-            self.list_channel_name,
-            csp.const([MyStruct(foo=2.0), MyStruct(foo=3.0)]),
-        )
+        scalar_edge = csp.const(MyStruct(foo=1.0))
+        list_edge = csp.const([MyStruct(foo=2.0), MyStruct(foo=3.0)])
+        channels.set_channel(self.scalar_channel_name, scalar_edge)
+        channels.set_channel(self.list_channel_name, list_edge)
         channels.set_state(self.scalar_channel_name, keyby="id")
         channels.set_state(self.list_channel_name, keyby="id")
         if self.connect_channels_assertion:
@@ -795,11 +787,11 @@ def connect(self, channels: MyGatewayChannels) -> None:
         )
         csp.add_graph_output(
             f"s_{self.scalar_channel_name}",
-            channels.get_channel(f"s_{self.scalar_channel_name}"),
+            channels.get_state(self.scalar_channel_name),
         )
         csp.add_graph_output(
             f"s_{self.list_channel_name}",
-            channels.get_channel(f"s_{self.list_channel_name}"),
+            channels.get_state(self.list_channel_name),
         )
 
 
@@ -873,6 +865,34 @@ def test_conflicting_dynamic_channels():
         csp.build_graph(MyGateway(modules=[setter_1, setter_2], channels=MyGatewayChannels()).graph)
 
 
+@pytest.mark.parametrize("module_order", ["setter_first", "getter_first"])
+def test_dynamic_channels_state_module_order_independence(module_order):
+    """Cross-module get_state/set_state must work regardless of connect() order.
+
+    The setter module declares a dynamic channel and registers state on it from
+    inside connect(). The getter module calls get_state() on that name from
+    inside its own connect(). Whichever module's connect() runs first, the
+    getter must end up wired to the real state edge via DelayedEdge binding.
+    """
+    setter = MySetModuleDynamicChannels(
+        scalar_channel_name="my_dynamic_scalar_channel",
+        list_channel_name="my_dynamic_list_channel",
+    )
+    getter = MyGetModuleDynamicChannels(
+        scalar_channel_name="my_dynamic_scalar_channel",
+        list_channel_name="my_dynamic_list_channel",
+    )
+    modules = [setter, getter] if module_order == "setter_first" else [getter, setter]
+    gateway = MyGateway(modules=modules, channels=MyGatewayChannels())
+
+    out = csp.run(gateway.graph, starttime=datetime(2020, 1, 1), endtime=timedelta(1))
+    assert len(out["my_dynamic_scalar_channel"]) == 1
+    assert out["my_dynamic_scalar_channel"][0][1].foo == 1.0
+    assert len(out["my_dynamic_list_channel"]) == 1
+    assert len(out["s_my_dynamic_scalar_channel"]) == 1
+    assert len(out["s_my_dynamic_list_channel"]) == 2
+
+
 def test_dynamic_channels_same_static_properties():
     my_object = {}
     channel = MyGatewayChannels(my_static_dict_of_objects={"a": my_object})
@@ -990,9 +1010,7 @@ def test_gateway_channels_fields_classmethod():
         "my_static_list",
         "my_static_dict_of_objects",
         "my_channel",
-        "s_my_channel",
         "my_list_channel",
-        "s_my_list_channel",
         "my_enum_basket",
         "my_str_basket",
         "my_enum_basket_list",
diff --git a/csp_gateway/tests/server/modules/io/test_json.py b/csp_gateway/tests/server/modules/io/test_json.py
index 96960d4..2a35274 100644
--- a/csp_gateway/tests/server/modules/io/test_json.py
+++ b/csp_gateway/tests/server/modules/io/test_json.py
@@ -1,5 +1,5 @@
 from datetime import datetime, timedelta
-from typing import Dict, Type
+from typing import Annotated, Dict, Type
 
 import csp
 import orjson
@@ -31,9 +31,8 @@ class MyStrStruct(GatewayStruct):
 
 
 class MySmallGatewayChannels(GatewayChannels):
-    example: ts[int] = None
+    example: Annotated[ts[int], State(keyby="id")] = None
     struct_with_str: ts[MyStrStruct] = None
-    s_example: ts[State[int]] = None
     my_str_basket: Dict[str, ts[float]] = None
 
 
diff --git a/csp_gateway/tests/server/modules/io/test_json_converter.py b/csp_gateway/tests/server/modules/io/test_json_converter.py
index 8947ec8..93401ae 100644
--- a/csp_gateway/tests/server/modules/io/test_json_converter.py
+++ b/csp_gateway/tests/server/modules/io/test_json_converter.py
@@ -74,9 +74,7 @@ def connect(self, channels: MyGatewayChannels) -> None:
             self.my_list_data,
         )
         channels.add_send_channel(MyGatewayChannels.my_channel)
-        channels.set_state(MyGatewayChannels.my_channel, "id")
         channels.add_send_channel(MyGatewayChannels.my_list_channel)
-        channels.set_state(MyGatewayChannels.my_list_channel, "id")
 
         if self.by_key:
             channels.set_channel(MyGatewayChannels.my_enum_basket, self.my_data, MyEnum.ONE)
@@ -415,7 +413,6 @@ def test_encode_filter_channels(by_key):
         by_key=by_key,
     )
     channels = MyGatewayChannels.fields()
-    assert MyGatewayChannels.s_my_channel in channels
     assert MyGatewayChannels.my_array_channel in channels
     json_encoder = MyJsonEncoder(channels_list=channels)
     gateway = MyGateway(modules=[setter, json_encoder], channels=MyGatewayChannels())
@@ -429,7 +426,6 @@ def test_encode_filter_channels(by_key):
     assert snapshot_model.my_enum_basket[MyEnum.ONE].foo == 1.0
     assert snapshot_model.my_enum_basket[MyEnum.TWO].foo == 2.0
     assert snapshot_model.my_enum_basket_list[MyEnum.ONE] == snapshot_model.my_enum_basket_list[MyEnum.TWO]
-    assert MyGatewayChannels.s_my_channel not in type(snapshot_model).model_fields
     assert getattr(snapshot_model, _CSP_ENGINE_CYCLE_TIMESTAMP_FIELD) == datetime(2020, 1, 1)
 
 
diff --git a/csp_gateway/tests/server/modules/logging/test_printing.py b/csp_gateway/tests/server/modules/logging/test_printing.py
index ef61862..99fa300 100644
--- a/csp_gateway/tests/server/modules/logging/test_printing.py
+++ b/csp_gateway/tests/server/modules/logging/test_printing.py
@@ -1,5 +1,5 @@
 from datetime import datetime, timedelta
-from typing import Dict, List
+from typing import Annotated, Dict, List
 
 import csp
 import numpy as np
@@ -21,10 +21,8 @@ class MyPrintGatewayChannels(GatewayChannels):
     my_static: float = 0.0
     my_static_dict: Dict[str, float] = {}
     my_static_list: List[str] = []
-    my_channel: ts[int] = None
-    s_my_channel: ts[State[int]] = None
-    my_list_channel: ts[[int]] = None
-    s_my_list_channel: ts[State[int]] = None
+    my_channel: Annotated[ts[int], State(keyby="id")] = None
+    my_list_channel: Annotated[ts[[int]], State(keyby="id")] = None
     my_enum_basket: Dict[MyEnum, ts[int]] = None
     my_str_basket: Dict[str, ts[int]] = None
     my_enum_basket_list: Dict[MyEnum, ts[[int]]] = None
@@ -48,9 +46,7 @@ def connect(self, channels: MyPrintGatewayChannels) -> None:
             csp.const(self.my_list_data),
         )
         channels.add_send_channel(MyPrintGatewayChannels.my_channel)
-        channels.set_state(MyPrintGatewayChannels.my_channel, "id")
         channels.add_send_channel(MyPrintGatewayChannels.my_list_channel)
-        channels.set_state(MyPrintGatewayChannels.my_list_channel, "id")
 
         channels.set_channel(MyPrintGatewayChannels.my_array_channel, csp.const(np.array([1.0, 2.0])))
 
diff --git a/csp_gateway/tests/server/shared/test_channel_selection.py b/csp_gateway/tests/server/shared/test_channel_selection.py
index eae254e..d06d0a8 100644
--- a/csp_gateway/tests/server/shared/test_channel_selection.py
+++ b/csp_gateway/tests/server/shared/test_channel_selection.py
@@ -18,8 +18,8 @@ def test_selection():
 
     assert selection.select_from(MyGatewayChannels) == channels
     assert selection.select_from(MyGatewayChannels, state_channels=True) == [
-        "s_my_channel",
-        "s_my_list_channel",
+        "my_channel",
+        "my_list_channel",
     ]
     assert selection.select_from(MyGatewayChannels, static_fields=True) == static
 
@@ -30,19 +30,18 @@ def test_selection_duplicates():
     selection = ChannelSelection(include=channels + channels + static)
 
     assert selection.select_from(MyGatewayChannels) == channels
-    assert selection.select_from(MyGatewayChannels, state_channels=True) == ["s_my_channel"]
+    assert selection.select_from(MyGatewayChannels, state_channels=True) == ["my_channel"]
     assert selection.select_from(MyGatewayChannels, static_fields=True) == static
 
 
 def test_selection_all_fields():
-    channels = ["my_enum_basket", "my_channel", "s_my_channel"]
+    channels = ["my_enum_basket", "my_channel"]
     static = ["my_static", "my_static_dict"]
     selection = ChannelSelection(include=channels + static)
 
     assert selection.select_from(MyGatewayChannels, all_fields=True) == [
         "my_enum_basket",
         "my_channel",
-        "s_my_channel",
         "my_static",
         "my_static_dict",
     ]
@@ -54,7 +53,7 @@ def test_selection_include():
     selection = ChannelSelection(include=channels + static)
 
     assert selection.select_from(MyGatewayChannels) == channels
-    assert selection.select_from(MyGatewayChannels, state_channels=True) == ["s_my_channel"]
+    assert selection.select_from(MyGatewayChannels, state_channels=True) == ["my_channel"]
     assert selection.select_from(MyGatewayChannels, static_fields=True) == static
     assert selection.select_from(MyGatewayChannels, all_fields=True) == [
         "my_channel",
@@ -75,7 +74,7 @@ def test_selection_exclude():
     static = ["my_static_dict", "my_static_list"]
 
     assert selection.select_from(MyGatewayChannels) == channels
-    assert selection.select_from(MyGatewayChannels, state_channels=True) == ["s_my_list_channel"]
+    assert selection.select_from(MyGatewayChannels, state_channels=True) == ["my_list_channel"]
     assert selection.select_from(MyGatewayChannels, static_fields=True) == static
 
 
diff --git a/csp_gateway/tests/server/web/test_webserver.py b/csp_gateway/tests/server/web/test_webserver.py
index 3761707..01a3bc2 100644
--- a/csp_gateway/tests/server/web/test_webserver.py
+++ b/csp_gateway/tests/server/web/test_webserver.py
@@ -127,12 +127,12 @@ def test_openapi_includes_fully_qualified_type_name(self, rest_client: TestClien
             f"Expected fully qualified type name in /api/v1/lookup/example/{{id}} type_, got: {lookup_example_type}"
         )
 
-        # Test /api/v1/state/example route
-        state_example_path = paths.get("/api/v1/state/example", {})
+        # Test /api/v1/state/example_with_state route
+        state_example_path = paths.get("/api/v1/state/example_with_state", {})
         state_example_get = state_example_path.get("get", {})
         state_example_type = state_example_get.get("type_", "")
         assert state_example_type == "csp_gateway.server.demo.omnibus.ExampleData", (
-            f"Expected fully qualified type name in /api/v1/state/example type_, got: {state_example_type}"
+            f"Expected fully qualified type name in /api/v1/state/example_with_state type_, got: {state_example_type}"
         )
 
         # Test /api/v1/controls/heartbeat route
@@ -359,7 +359,7 @@ def test_csp_state(self, rest_client: TestClient):
         last_data = self._wait_for_data(rest_client=rest_client)
 
         # Get state data
-        response = rest_client.get("/api/v1/state/example?token=test")
+        response = rest_client.get("/api/v1/state/example_with_state?token=test")
         assert response.status_code == 200
 
         state_data = response.json()
@@ -371,7 +371,7 @@ def test_csp_state_query(self, rest_client: TestClient):
         self._wait_for_data(rest_client=rest_client)
 
         # Get state data
-        response = rest_client.get('/api/v1/state/example?token=test&query={"filters":[{"attr":"x","by":{"value":1,"where":"=="}}]}')
+        response = rest_client.get('/api/v1/state/example_with_state?token=test&query={"filters":[{"attr":"x","by":{"value":1,"where":"=="}}]}')
         assert response.status_code == 200
 
         state_data = response.json()
@@ -508,6 +508,8 @@ def test_csp_toplevel_last(self, rest_client: TestClient):
             "controls",
             "example",
             "example_list",
+            "example_with_state",
+            "example_with_state_multiple",
             "never_ticks",
             "str_basket",
         ]
@@ -522,6 +524,8 @@ def test_csp_toplevel_next(self, rest_client: TestClient):
             "controls",
             "example",
             "example_list",
+            "example_with_state",
+            "example_with_state_multiple",
             "never_ticks",
             "str_basket",
         ]
@@ -530,7 +534,12 @@ def test_csp_toplevel_state(self, rest_client: TestClient):
         self._wait_for_data(rest_client=rest_client)
         response_state = rest_client.get("/api/v1/state?token=test")
         assert response_state.status_code == 200
-        assert sorted(response_state.json()) == ["example"]
+        assert sorted(response_state.json()) == [
+            "example",
+            "example_with_state",
+            "example_with_state_alternative",
+            "example_with_state_multiple",
+        ]
 
     def test_csp_toplevel_send(self, rest_client: TestClient):
         self._wait_for_data(rest_client=rest_client)
@@ -557,7 +566,12 @@ def test_gateway_client(self, mock_get, mock_post, rest_client: TestClient, gate
         assert sorted(list(gateway_client.openapi_spec.keys())) == ["components", "info", "openapi", "paths"]
         assert "/api/v1/last/example" in gateway_client.openapi_spec["paths"].keys()
         assert "/api/v1/last/example_list" in gateway_client.openapi_spec["paths"].keys()
-        assert response_state == ["example"]
+        assert sorted(response_state) == [
+            "example",
+            "example_with_state",
+            "example_with_state_alternative",
+            "example_with_state_multiple",
+        ]
 
         for route in ["example", "example_list"]:
             data = gateway_client.last(route)
@@ -602,7 +616,7 @@ def test_gateway_client(self, mock_get, mock_post, rest_client: TestClient, gate
             send_data["y"],
         )
 
-        return_data = gateway_client.state("example", query=Query(filters=[Filter(attr="x", by={"value": 1, "where": "=="})]))
+        return_data = gateway_client.state("example_with_state", query=Query(filters=[Filter(attr="x", by={"value": 1, "where": "=="})]))
         assert isinstance(return_data, list)
         return_datum = return_data[0]
         assert "id" in return_datum
@@ -798,6 +812,8 @@ def test_perspective_tables(self, rest_client: TestClient):
             "controls",
             "example",
             "example_list",
+            "example_with_state",
+            "example_with_state_multiple",
             "my_custom_table",
             # "never_ticks",  # NOTE: never_ticks has no data, so table is omitted
             "str_basket",
@@ -831,6 +847,8 @@ def test_stream(self, rest_client: TestClient):
             "controls",
             "example",
             "example_list",
+            "example_with_state",
+            "example_with_state_multiple",
             "heartbeat",
             "never_ticks",
             "str_basket/a",
diff --git a/docs/wiki/API.md b/docs/wiki/API.md
index 00adbcc..b4b96ef 100644
--- a/docs/wiki/API.md
+++ b/docs/wiki/API.md
@@ -32,10 +32,39 @@ As described in [Overview#Channels](Overview#Channels), the `csp-gateway` REST A
 
 ## State
 
-A `GatewayModule` can call `set_state` in its `connect` method to allow for this API to be available.
+State in `csp-gateway` exposes the accumulated history of a channel via the REST API as `/api/v1/state/<alias>`.
 State is collected by one or more instance attributes into an in-memory [DuckDB](https://duckdb.org/) instance.
 
-For example, suppose I have the following type:
+There are two ways to declare state on a channel:
+
+1. **Annotation** on the channel definition, using `State(keyby, alias=...)`:
+
+   ```python
+   from typing import Annotated
+   from csp_gateway import State, GatewayChannels, ts
+
+   class MyChannels(GatewayChannels):
+       # implicit alias "example_with_state" (matches the field name)
+       example_with_state: Annotated[ts[ExampleData], State(("id", "x"))] = None
+
+       # multiple state views on a single channel, each addressable by alias
+       example_multi: Annotated[
+           ts[ExampleData],
+           State(("id", "x")),                          # alias = "example_multi"
+           State(("id", "y"), alias="example_multi_alt"),
+       ] = None
+   ```
+
+1. **Imperative** call from a `GatewayModule`'s `connect` method:
+
+   ```python
+   def connect(self, channels: MyChannels):
+       edge = ...  # produce a csp.ts edge
+       channels.set_channel(MyChannels.example, edge)
+       channels.set_state(edge, "example_from_connect", ("id", "x", "z"))
+   ```
+
+For example, given:
 
 ```python
 class ExampleData(GatewayStruct):
@@ -44,12 +73,7 @@ class ExampleData(GatewayStruct):
     z: str
 ```
 
-If my `GatewayModule` called `set_state("example", ("x",))`, state would be collected as the last tick of `ExampleData` per each unique value of `x`. If called with `set_state("example", ("x", "y"))`, it would be collected as the last tick per each unique pair `x,y`, etc.
-
-> [!IMPORTANT]
->
-> This code and API will likely change a bit as we allow for more granular collection of records,
-> and expose more DuckDB functionality.
+Declaring `State(("x",))` collects the last tick of `ExampleData` per each unique value of `x`. Declaring `State(("x", "y"))` collects the last tick per each unique pair `(x, y)`, etc.
 
 ## Query
 
@@ -59,19 +83,19 @@ Here are some examples from the autodocumentation illustrating the use of filter
 
 ```raw
 # Filter only records where `record.x` == 5
-api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":5,"where":"=="}}]}
+api/v1/state/example_with_state?query={"filters":[{"attr":"x","by":{"value":5,"where":"=="}}]}
 
 # Filter only records where `record.x` < 10
-/api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":10,"where":"<"}}]}
+/api/v1/state/example_with_state?query={"filters":[{"attr":"x","by":{"value":10,"where":"<"}}]}
 
 # Filter only records where `record.timestamp` < "2023-03-30T14:45:26.394000"
-/api/v1/state/example?query={"filters":[{"attr":"timestamp","by":{"when":"2023-03-30T14:45:26.394000","where":"<"}}]}
+/api/v1/state/example_with_state?query={"filters":[{"attr":"timestamp","by":{"when":"2023-03-30T14:45:26.394000","where":"<"}}]}
 
 # Filter only records where `record.id` < `record.y`
-/api/v1/state/example?query={"filters":[{"attr":"id","by":{"attr":"y","where":"<"}}]}
+/api/v1/state/example_with_state?query={"filters":[{"attr":"id","by":{"attr":"y","where":"<"}}]}
 
 # Filter only records where `record.x` < 50 and `record.x` >= 30
-/api/v1/state/example?query={"filters":[{"attr":"x","by":{"value":50,"where":"<"}},{"attr":"x","by":{"value":30,"where":">="}}]}
+/api/v1/state/example_with_state?query={"filters":[{"attr":"x","by":{"value":50,"where":"<"}},{"attr":"x","by":{"value":30,"where":">="}}]}
 ```
 
 > [!IMPORTANT]
diff --git a/docs/wiki/Client.md b/docs/wiki/Client.md
index 582a431..4d5691a 100644
--- a/docs/wiki/Client.md
+++ b/docs/wiki/Client.md
@@ -282,11 +282,11 @@ client.state().as_json()
 ```
 
 ```
-['example']
+['example', 'example_with_state', 'example_with_state_alternative', 'example_with_state_multiple']
 ```
 
 ```python
-client.state("example").as_pandas_df().tail()
+client.state("example_with_state").as_pandas_df().tail()
 
 # We note that there are a large number of columns in the above dataframe.
 # This is because `mapping` is a dict with different keys for eery row.
@@ -503,7 +503,7 @@ client.send(
     }
 )
 
-client.state("example").as_pandas_df().tail()
+client.state("example_with_state").as_pandas_df().tail()
 ```
 
 <div>
diff --git a/docs/wiki/Overview.md b/docs/wiki/Overview.md
index 9485dc2..c680d27 100644
--- a/docs/wiki/Overview.md
+++ b/docs/wiki/Overview.md
@@ -165,8 +165,29 @@ Recall from above:
 
 `GatewayChannels` also have the ability to create and maintain state, which is particularly useful in the REST API. State is managed by an in-memory [DuckDB](https://duckdb.org/) instance.
 
-- `set_state(self, field: str, keyby: Union[str, Tuple[str, ...]], indexer: Union[str, int] = None) -> None`: Collect state of an edge by a certain attribute indexer
-- `get_state(self, field: str, indexer: Union[str, int] = None) -> Any`: Get the current state collected on an edge as a ticking object
+State can be declared two ways:
+
+1. **Annotation** (on the channel definition):
+
+   ```python
+   from typing import Annotated
+   from csp_gateway import State, GatewayChannels
+
+   class MyChannels(GatewayChannels):
+       example_with_state: Annotated[ts[ExampleData], State(("id", "x"))] = None
+       # multiple State() annotations are allowed; each may set an `alias`
+       example_multi: Annotated[
+           ts[ExampleData],
+           State(("id", "x")),
+           State(("id", "y"), alias="example_multi_alt"),
+       ] = None
+   ```
+
+1. **Imperative**, from a module's `connect` method via `set_state`:
+
+   - `set_state(self, edge: Edge, field: str, keyby: Union[str, Tuple[str, ...]], indexer: Union[str, int] = None) -> None`: Collect state of `edge` by a certain attribute, exposed under `field`
+
+- `get_state(self, field: str, indexer: Union[str, int] = None) -> Any`: Get the current state collected for a state `field` as a ticking object
 
 `GatewayChannels` can tell the REST API to allow a certain channel to be sent in via POST requests:
 
@@ -177,7 +198,7 @@ This is the primary API for both the [REST API](API) and the [integrated Python
 
 - `last(self, field: str, indexer: Union[str, int] = None, *, timeout=None) -> None`: Get the last ticked value on a channel
 - `next(self, field: str, indexer: Union[str, int] = None, *, timeout=None) -> None`: Get (wait for) the next ticked value on a channel
-- `state(self, field: str, indexer: Union[str, int] = None, *, timeout=None) -> Any`: Get the state collected on a channel
+- `state(self, field: str, indexer: Union[str, int] = None, *, timeout=None) -> Any`: Get the state collected for a state `field`
 - `query(self, field: str, indexer: Union[str, int] = None, query: "Query" = None) -> Any`: Query the state on a channel (see [API](API) for more about `Query`)
 - `send(self, field: str, value: Any, indexer: Union[str, int] = None) -> None`: Send new data to a channel, from outside the `csp` graph