diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7b8f11f6..acad92cd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
 # Changelog
 
+## 0.5.2
+### Changed
+- **Breaking:** `Repository.aggregate()` yields `WrappedAggregate` instead of `Aggregate` directly
+
 ## 0.5.1
 ### Added
 - In transaction subscription to category
diff --git a/docs/adr/20260314-aggregate-metadata.md b/docs/adr/20260314-aggregate-metadata.md
new file mode 100644
index 00000000..e1aef13a
--- /dev/null
+++ b/docs/adr/20260314-aggregate-metadata.md
@@ -0,0 +1,143 @@
+# Aggregate Metadata
+
+Date: 2026-03-14
+
+## Status
+
+Accepted
+
+## Context
+
+Currently `Repository.aggregate()` yields only the aggregate instance. The caller has no access to stream-level information such as:
+
+- What is the current version of the aggregate?
+- Is this a new aggregate (no prior events in the stream)?
+- When was the stream created or last modified?
+- What stream identity was used?
+
+This information is essential for many use cases:
+- Detecting whether an aggregate is being created for the first time (e.g. to enforce idempotent creation)
+- Knowing the version for external optimistic concurrency checks or ETags
+- Accessing timestamps for auditing or display purposes
+
+Other event sourcing frameworks consistently expose this kind of metadata:
+- **Marten** returns `StreamState` with `Version`, `Created`, `LastTimestamp`, `AggregateType` via `FetchForWriting<T>`
+- **Eventide** returns `(entity, version)` tuple, with `:no_stream` sentinel for non-existent streams
+- **Axon** exposes `@AggregateVersion` and `AggregateLifecycle.isLive()`
+- **Ecotone** enriches events with `_aggregate_version`, `_aggregate_type` headers
+
+## Decisions
+
+### D1: Wrapper pattern — `WrappedAggregate`
+
+Following the established pattern of `WrappedEvent` which wraps `Event` with metadata, the repository will return a `WrappedAggregate` that wraps the aggregate instance with stream metadata. The aggregate itself remains a pure domain object with no knowledge of infrastructure concerns.
+
+```python
+with repo.aggregate(uuid, LightSwitch()) as wrapped:
+    if wrapped.is_new:
+        wrapped.aggregate.turn_on()
+```
+
+**Rationale:** This is consistent with `WrappedEvent[TEvent]` which carries `event`, `version`, `uuid`, `created_at`, and `context` alongside the event itself. `WrappedAggregate[TAggregate]` follows the same convention — the wrapper holds metadata, the inner object stays pure.
+
+**Migration:** This is a breaking change. Existing callers using `as aggregate` will need to change to `as wrapped` and access `wrapped.aggregate`. Given the library is pre-1.0, this is acceptable.
+
+### D2: Fields on `WrappedAggregate`
+
+| Field | Type | Source |
+|---|---|---|
+| `aggregate` | `TAggregate` | The aggregate instance |
+| `version` | `int` (computed) | Stored version + count of pending changes in aggregate |
+| `is_new` | `bool` (computed) | `True` when stored version is 0 (no events existed before load) |
+| `stream_id` | `StreamId` | Built from UUID + aggregate category |
+| `created_at` | `datetime \| None` | Timestamp of the first event in the stream |
+| `updated_at` | `datetime \| None` | Timestamp of the last event in the stream |
+| `context` | `Context \| None` | The context passed to the repository for this operation |
+
+**`version` is computed dynamically.** It reflects the stored version plus any uncommitted changes the aggregate has emitted. This means:
+- After loading an aggregate with 5 events: `version == 5`
+- After emitting 2 more events: `version == 7`
+- After persisting (context exit): version reflects the final stored state
+
+This is achieved by reading the aggregate's `__changes__` property and computing `stored_version + len(aggregate.__changes__)` on access.
+
+**`is_new` is based on stored version only**, not including pending changes. An aggregate that was just created and has emitted its first event is still `is_new == True` — it didn't exist before this session. This matches Eventide's `:no_stream` semantics.
+
+### D3: Class naming and location
+
+The class will be called `WrappedAggregate`. It is a generic dataclass `WrappedAggregate[TAggregate]`, mirroring `WrappedEvent[TEvent]`. It lives in the `event_sourcery.event_sourcing` module alongside `Aggregate` and `Repository`.
+
+**Rationale:** `WrappedAggregate` directly mirrors `WrappedEvent` — both are generic wrappers that pair a domain object with infrastructure metadata. The naming convention is already established in the project. A dataclass is used instead of a Pydantic model because `WrappedAggregate` is a simple data holder — it does not need serialization, validation, or schema generation. A dataclass keeps the dependency footprint minimal and the implementation straightforward.
+
+## Solution Proposal
+
+### New class: `WrappedAggregate`
+
+```python
+# event_sourcery/event_sourcing/aggregate.py
+
+TAggregate = TypeVar("TAggregate", bound=Aggregate)
+
+@dataclass
+class WrappedAggregate(Generic[TAggregate]):
+    aggregate: TAggregate
+    stream_id: StreamId
+    context: Context | None
+    created_at: datetime | None
+    updated_at: datetime | None
+    stored_version: int
+
+    @property
+    def version(self) -> int:
+        return self.stored_version + len(
+            getattr(self.aggregate, "__changes__", [])
+        )
+
+    @property
+    def is_new(self) -> bool:
+        return self.stored_version == 0
+
+```
+
+### Changes to `Repository`
+
+```python
+# event_sourcery/event_sourcing/repository.py
+
+@contextmanager
+def aggregate(
+    self,
+    uuid: StreamUUID,
+    aggregate: TAggregate,
+    context: Context | None = None,
+) -> Iterator[WrappedAggregate[TAggregate]]:
+    stream_id = StreamId(uuid=uuid, name=uuid.name, category=aggregate.category)
+    stored_version, created_at, updated_at = self._load(stream_id, aggregate)
+    wrapped = WrappedAggregate(
+        aggregate=aggregate,
+        stream_id=stream_id,
+        context=context,
+        created_at=created_at,
+        updated_at=updated_at,
+        stored_version=stored_version,
+    )
+    yield wrapped
+    self._save(aggregate, stored_version, stream_id, context)
+```
+
+The `_load` method will be updated to extract `created_at` from the first event's timestamp and `updated_at` from the last event's timestamp during replay.
+
+### Public API exports
+
+`WrappedAggregate` will be exported from `event_sourcery.event_sourcing` package `__init__.py`.
+
+## Consequences
+
+- **Breaking change** in `Repository.aggregate()` yield type — all callers must update from `as aggregate` to `as wrapped` and access `wrapped.aggregate`.
+- Consistent with the existing `WrappedEvent` pattern — both domain objects (`Event`, `Aggregate`) are wrapped with metadata by infrastructure, never polluted directly.
+- The aggregate remains a pure domain object — no infrastructure leakage.
+- `version` property provides a live view of the aggregate's version including pending changes, useful for logging/debugging.
+- `is_new` enables idempotent aggregate creation patterns without checking version manually.
+- `created_at` / `updated_at` are derived from event timestamps — no additional storage or queries needed.
+- `context` reference enables callers to inspect or pass along the context used for the current operation.
+- `version` reads `aggregate.__changes__` dynamically, so it updates automatically as events are emitted.
diff --git a/docs/code/test_recipes.py b/docs/code/test_recipes.py
index 84adfbbe..604553ea 100644
--- a/docs/code/test_recipes.py
+++ b/docs/code/test_recipes.py
@@ -289,9 +289,9 @@ def switch_on(self) -> None:
     from event_sourcery import StreamUUID
 
     stream_id = StreamUUID(name="light_switch/1")
-    with repository.aggregate(stream_id, LightSwitch()) as light_switch:
-        light_switch.switch_on()
-        light_switch.switch_on()
+    with repository.aggregate(stream_id, LightSwitch()) as wrapped:
+        wrapped.aggregate.switch_on()
+        wrapped.aggregate.switch_on()
     # --8<-- [end:event_sourcing_03]
 
     events = backend.event_store.load_stream(
diff --git a/docs/concepts/basics.md b/docs/concepts/basics.md
index 7fcd5155..284e82fc 100644
--- a/docs/concepts/basics.md
+++ b/docs/concepts/basics.md
@@ -188,71 +188,84 @@ To sum up, [Event Sourcing] comes down to:
 Event Sourcery provides a base class for an [Aggregate] and repository implementation that makes it much easy to create or read/change [Aggregate].
 
 ```python
+from event_sourcery.event import Event
+from event_sourcery.event_sourcing import Aggregate
+
+
+class TurnedOn(Event):
+    pass
+
+
+class TurnedOff(Event):
+    pass
+
+
 class LightSwitch(Aggregate):
     """A simple aggregate that models a light switch."""
+    category = "light_switch"  # stream category for all LightSwitch aggregates
+
     class AlreadyTurnedOn(Exception):
         pass
 
     class AlreadyTurnedOff(Exception):
         pass
 
-    def __init__(
-        self, past_events: list[Event], changes: list[Event], stream_id: StreamId
-    ) -> None:
+    def __init__(self) -> None:
         # init any state you need in aggregate class to check conditions
         self._shines = False
 
-        # required for base class
-        super().__init__(past_events, changes, stream_id)
-
-    def _apply(self, event: Event) -> None:
-        # each aggregate need an _apply method to parse events
+    def __apply__(self, event: Event) -> None:
+        # each aggregate needs an __apply__ method to handle events
         match event:
-            case TurnedOn() as event:
+            case TurnedOn():
                 self._shines = True
-            case TurnedOff() as event:
+            case TurnedOff():
                 self._shines = False
 
     def turn_on(self) -> None:
-        # this is one of command methods
-        # we can rejest it (i.e. raise an exception)
+        # this is a command method
+        # we can reject it (i.e. raise an exception)
         # if current state does not allow this to proceed
-        # e.g. light is already on
         if self._shines:
             raise LightSwitch.AlreadyTurnedOn
-        self._event(TurnedOn)
+        self._emit(TurnedOn())
 
     def turn_off(self) -> None:
         if not self._shines:
             raise LightSwitch.AlreadyTurnedOff
-
-        self._event(TurnedOff)
+        self._emit(TurnedOff())
 ```
 
-To create a [Repository] tailored for a particular [Aggregate] class, we need  that class and [EventStore] instance: 
+To create a [Repository] tailored for a particular [Aggregate] class, we need an [EventStore] instance:
 
 ```python
-repository = Repository[LightSwitch](event_store, LightSwitch)
+from event_sourcery.event_sourcing import Repository
+
+repository = Repository[LightSwitch](event_store)
 ```
 
-A [Repository] exposes method to create a new instance of [Aggregate]:
+A [Repository] exposes a context manager to work with an [Aggregate]. It returns a [WrappedAggregate] that wraps the aggregate with stream metadata (version, timestamps, etc.):
 
 ```python
-stream_id = uuid4()
+from event_sourcery import StreamUUID
+
+uuid = StreamUUID()
 
-with repository.new(stream_id=stream_id) as switch:
-    switch.turn_on()
+with repository.aggregate(uuid, LightSwitch()) as wrapped:
+    wrapped.aggregate.turn_on()
+    print(wrapped.version)   # 1
+    print(wrapped.is_new)    # True
 ```
 
-...or to work with existing [Aggregate], making sure changes are saved at the end:
+Loading an existing [Aggregate] works the same way — events are replayed automatically:
 
 ```python
-with repository.aggregate(stream_id=stream_id) as switch_second_incarnation:
+with repository.aggregate(uuid, LightSwitch()) as wrapped:
     try:
-        switch_second_incarnation.turn_on()
+        wrapped.aggregate.turn_on()
     except LightSwitch.AlreadyTurnedOn:
         # o mon Dieu, I made a mistake!
-        switch_second_incarnation.turn_off()
+        wrapped.aggregate.turn_off()
 ```
 
 A [Repository] is a thin wrapper over Event Store. One can also write Aggregates even without using our base class and use [EventStore] directly!
@@ -262,3 +275,4 @@ A [Repository] is a thin wrapper over Event Store. One can also write Aggregates
 [Event Sourcing]: ../recipes/event_sourcing.md
 [Aggregate]: ../reference/event_sourcing/Aggregate.md
 [Repository]: ../reference/event_sourcing/Repository.md
+[WrappedAggregate]: ../reference/event_sourcing/WrappedAggregate.md
diff --git a/docs/recipes/event_sourcing.md b/docs/recipes/event_sourcing.md
index d436bb26..83d0daaa 100644
--- a/docs/recipes/event_sourcing.md
+++ b/docs/recipes/event_sourcing.md
@@ -2,7 +2,7 @@
 
 Event Sourcery provides a few building blocks to work with event sourcing.
 
-These are [Aggregate] and [Repository] base classes.
+These are [Aggregate], [Repository] and [WrappedAggregate] classes.
 
 ## Usage
 
@@ -12,7 +12,7 @@ There are three required attributes that need to be defined:
 
 1. `category` class-level constant that will be added to all streams from all aggregates of this type
 2.  `__init__` if defined, must not accept any arguments
-3. `__apply__` method that will change internal state of the aggregate based on the event applied during reading state from the event store 
+3. `__apply__` method that will change internal state of the aggregate based on the event applied during reading state from the event store
 
 ```python
 --8<--
@@ -28,7 +28,7 @@ docs/code/test_recipes.py:event_sourcing_02_repo
 --8<--
 ```
 
-From now on, regardless if you want to work with a given aggregate for the first time or load existing one, you should use `repository.aggregate` context manager:
+From now on, regardless if you want to work with a given aggregate for the first time or load existing one, you should use `repository.aggregate` context manager. It returns a [WrappedAggregate] — a wrapper that provides the aggregate instance along with stream metadata such as `version`, `is_new`, `created_at`, and `updated_at`:
 
 ```python
 --8<--
@@ -36,6 +36,14 @@ docs/code/test_recipes.py:event_sourcing_03
 --8<--
 ```
 
+The aggregate itself is accessed via `wrapped.aggregate`. The wrapper also exposes useful properties:
+
+- `wrapped.version` — current version including pending (not yet persisted) changes
+- `wrapped.is_new` — `True` if no events existed before this session
+- `wrapped.created_at` / `wrapped.updated_at` — timestamps of first and last event in the stream
+- `wrapped.stream_id` — the stream identity (UUID + category)
+
 [Aggregate]: ../reference/event_sourcing/Aggregate.md
 [Repository]: ../reference/event_sourcing/Repository.md
+[WrappedAggregate]: ../reference/event_sourcing/WrappedAggregate.md
 [EventStore]: ../reference/event_store/EventStore.md
diff --git a/docs/reference/event_sourcing/WrappedAggregate.md b/docs/reference/event_sourcing/WrappedAggregate.md
new file mode 100644
index 00000000..b4cb482a
--- /dev/null
+++ b/docs/reference/event_sourcing/WrappedAggregate.md
@@ -0,0 +1 @@
+::: event_sourcery.event_sourcing.WrappedAggregate
diff --git a/event_sourcery/event_sourcing/__init__.py b/event_sourcery/event_sourcing/__init__.py
index 4a7af2c9..4195d3b1 100644
--- a/event_sourcery/event_sourcing/__init__.py
+++ b/event_sourcery/event_sourcing/__init__.py
@@ -1,7 +1,8 @@
 __all__ = [
     "Aggregate",
     "Repository",
+    "WrappedAggregate",
 ]
 
-from event_sourcery.event_sourcing.aggregate import Aggregate
+from event_sourcery.event_sourcing.aggregate import Aggregate, WrappedAggregate
 from event_sourcery.event_sourcing.repository import Repository
diff --git a/event_sourcery/event_sourcing/aggregate.py b/event_sourcery/event_sourcing/aggregate.py
index 07eae3bf..6886f5d3 100644
--- a/event_sourcery/event_sourcing/aggregate.py
+++ b/event_sourcery/event_sourcing/aggregate.py
@@ -1,9 +1,11 @@
+import dataclasses
 from collections.abc import Iterator
 from contextlib import contextmanager
-from typing import ClassVar
+from datetime import datetime
+from typing import ClassVar, Generic, TypeVar
 
-from event_sourcery import StreamCategory
-from event_sourcery.event import Event
+from event_sourcery import StreamCategory, StreamId
+from event_sourcery.event import Context, Event
 
 
 class Aggregate:
@@ -19,12 +21,16 @@ class Aggregate:
 
     Attributes:
         category (ClassVar[StreamCategory]): StreamCategory for the aggregate type (group streams).
-        _changes (list[Event]): List of yet not persisted events.
+        __changes__ (list[Event]): List of yet not persisted events.
     """
 
     category: ClassVar[StreamCategory]
     _changes: list[Event]
 
+    @property
+    def __changes__(self) -> list[Event]:
+        return list(getattr(self, "_changes", []))
+
     @contextmanager
     def __persisting_changes__(self) -> Iterator[Iterator[Event]]:
         """
@@ -36,8 +42,7 @@ def __persisting_changes__(self) -> Iterator[Iterator[Event]]:
         Returns:
             Iterator[Iterator[Event]]: Iterator over unpersisted events.
         """
-        yield iter(getattr(self, "_changes", []))
-        self._changes = []
+        yield iter(self.__changes__)
 
     def __apply__(self, event: Event) -> None:
         """
@@ -65,3 +70,54 @@ def _emit(self, event: Event) -> None:
             self._changes = []
         self.__apply__(event)
         self._changes.append(event)
+
+
+TAggregate = TypeVar("TAggregate", bound=Aggregate)
+TContext = TypeVar("TContext", bound=Context)
+
+
+@dataclasses.dataclass()
+class WrappedAggregate(Generic[TAggregate]):
+    """
+    Wraps an aggregate instance with stream-level metadata.
+
+    Provides access to the aggregate alongside information such as version,
+    timestamps, and whether the aggregate is newly created. Follows the same
+    wrapper pattern as ``WrappedEvent``.
+
+    Attributes:
+        aggregate: The aggregate instance.
+        stream_id: The stream identity (UUID + category).
+        context: The context passed for this operation, if any.
+        created_at: Timestamp of the first event in the stream.
+        updated_at: Timestamp of the last event in the stream.
+        stored_version: Number of events persisted before this session.
+    """
+
+    aggregate: TAggregate
+    stream_id: StreamId
+    context: Context = dataclasses.field(default_factory=Context)
+    stored_version: int = 0
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+    @property
+    def version(self) -> int:
+        """Current version of the aggregate, including pending events."""
+        return self.stored_version + len(getattr(self.aggregate, "__changes__", []))
+
+    @property
+    def is_new(self) -> bool:
+        """Whether the aggregate is newly created (no events existed before)."""
+        return self.stored_version == 0
+
+    def get_context(self, context_type: type[TContext]) -> TContext:
+        """Convert the stored context to a specific context type.
+
+        Args:
+            context_type: The target context class to validate against.
+
+        Returns:
+            An instance of *context_type* populated from the stored context data.
+        """
+        return context_type.model_validate(self.context.model_dump())
diff --git a/event_sourcery/event_sourcing/repository.py b/event_sourcery/event_sourcing/repository.py
index 89ef44c4..3a9a29f4 100644
--- a/event_sourcery/event_sourcing/repository.py
+++ b/event_sourcery/event_sourcing/repository.py
@@ -5,6 +5,7 @@
 from event_sourcery import EventStore, StreamId, StreamUUID
 from event_sourcery.event import Context, Event, WrappedEvent
 from event_sourcery.event_sourcing import Aggregate
+from event_sourcery.event_sourcing.aggregate import WrappedAggregate
 
 TAggregate = TypeVar("TAggregate", bound=Aggregate)
 TEvent = TypeVar("TEvent", bound=Event)
@@ -28,13 +29,13 @@ def aggregate(
         uuid: StreamUUID,
         aggregate: TAggregate,
         context: Context | None = None,
-    ) -> Iterator[TAggregate]:
+    ) -> Iterator[WrappedAggregate[TAggregate]]:
         """
         Context manager for loading an aggregate instance.
 
         Loads the aggregate's event stream, replays events to reconstruct its state,
-        yields the aggregate for use, and persists any new events emitted during the
-        context.
+        yields a ``WrappedAggregate`` containing the aggregate and stream metadata,
+        and persists any new events emitted during the context.
 
         Args:
             uuid (StreamUUID): The unique identifier of the aggregate's stream.
@@ -42,33 +43,33 @@ def aggregate(
             context (Context | None): Optional context to attach to all emitted events.
 
         Yields:
-            TAggregate: The loaded and ready-to-use aggregate instance.
+            WrappedAggregate[TAggregate]: The aggregate wrapped with stream metadata.
         """
         stream_id = StreamId(uuid=uuid, name=uuid.name, category=aggregate.category)
-        old_version = self._load(stream_id, aggregate)
-        yield aggregate
-        self._save(aggregate, old_version, stream_id, context)
 
-    def _load(self, stream_id: StreamId, aggregate: TAggregate) -> int:
-        stream = self._event_store.load_stream(stream_id)
-        last_version = 0
-        for envelope in stream:
-            aggregate.__apply__(envelope.event)
-            last_version = cast(int, envelope.version)
+        wrapped = WrappedAggregate(
+            aggregate=aggregate,
+            stream_id=stream_id,
+            context=context or Context(),
+        )
+        self._load(wrapped)
+        yield wrapped
+        self._save(wrapped)
 
-        return last_version
+    def _load(self, wrapped: WrappedAggregate[TAggregate]) -> None:
+        stream = self._event_store.load_stream(wrapped.stream_id)
+        for envelope in stream:
+            wrapped.aggregate.__apply__(envelope.event)
+            wrapped.stored_version = cast(int, envelope.version)
+            if wrapped.created_at is None:
+                wrapped.created_at = envelope.created_at
+            wrapped.updated_at = envelope.created_at
 
-    def _save(
-        self,
-        aggregate: TAggregate,
-        old_version: int,
-        stream_id: StreamId,
-        context: Context | None = None,
-    ) -> None:
-        with aggregate.__persisting_changes__() as pending:
-            start_from = old_version + 1
+    def _save(self, wrapped: WrappedAggregate[TAggregate]) -> None:
+        with wrapped.aggregate.__persisting_changes__() as pending:
+            start_from = wrapped.stored_version + 1
             events = [
-                WrappedEvent.wrap(event, version, context=context)
+                WrappedEvent.wrap(event, version, context=wrapped.context)
                 for version, event in enumerate(pending, start=start_from)
             ]
 
@@ -77,6 +78,6 @@ def _save(
 
             self._event_store.append(
                 *events,
-                stream_id=stream_id,
-                expected_version=old_version,
+                stream_id=wrapped.stream_id,
+                expected_version=wrapped.stored_version,
             )
diff --git a/mkdocs.yml b/mkdocs.yml
index 3bb88b39..cd3d63df 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -28,6 +28,7 @@ nav:
         - 'Event Sourcing':
           - 'Aggregate': 'reference/event_sourcing/Aggregate.md'
           - 'Repository': 'reference/event_sourcing/Repository.md'
+          - 'WrappedAggregate': 'reference/event_sourcing/WrappedAggregate.md'
         - 'EventStore':
           - 'EventStore': 'reference/event_store/EventStore.md'
           - 'StreamId': 'reference/event_store/StreamId.md'
@@ -89,6 +90,7 @@ nav:
         - 'Packaging Of The Project Revised': 'adr/20250411-packaging-of-the-project-revised.md'
         - 'Privacy Data Encryption And Shredding': 'adr/20250729-privacy-data-encryption-and-shredding.md'
         - 'Packaging Of The Event Store Package': 'adr/20251004-packaging-of-the-event_store-package.md'
+        - 'Aggregate Metadata': 'adr/20260314-aggregate-metadata.md'
 
 plugins:
 - search
diff --git a/tests/event_sourcing/conftest.py b/tests/event_sourcing/conftest.py
new file mode 100644
index 00000000..19a982eb
--- /dev/null
+++ b/tests/event_sourcing/conftest.py
@@ -0,0 +1,10 @@
+import pytest
+
+from event_sourcery import EventStore
+from event_sourcery.event_sourcing import Repository
+from tests.event_sourcing.light_switch import LightSwitch
+
+
+@pytest.fixture()
+def repo(event_store: EventStore) -> Repository[LightSwitch]:
+    return Repository[LightSwitch](event_store)
diff --git a/tests/event_sourcing/light_switch.py b/tests/event_sourcing/light_switch.py
new file mode 100644
index 00000000..2d54352c
--- /dev/null
+++ b/tests/event_sourcing/light_switch.py
@@ -0,0 +1,45 @@
+from event_sourcery import Event
+from event_sourcery.event_sourcing import Aggregate
+
+
+class TurnedOn(Event):
+    pass
+
+
+class TurnedOff(Event):
+    pass
+
+
+class LightSwitch(Aggregate):
+    category = "light_switch"
+
+    class AlreadyTurnedOn(Exception):
+        pass
+
+    class AlreadyTurnedOff(Exception):
+        pass
+
+    def __init__(self) -> None:
+        self._shines = False
+
+    def __apply__(self, event: Event) -> None:
+        match event:
+            case TurnedOn():
+                self._shines = True
+            case TurnedOff():
+                self._shines = False
+
+    def turn_on(self) -> None:
+        if self._shines:
+            raise LightSwitch.AlreadyTurnedOn
+        self._emit(TurnedOn())
+
+    def turn_off(self) -> None:
+        if not self._shines:
+            raise LightSwitch.AlreadyTurnedOff
+
+        self._emit(TurnedOff())
+
+    @property
+    def shines(self) -> bool:
+        return self._shines
diff --git a/tests/event_sourcing/test_aggregate.py b/tests/event_sourcing/test_aggregate.py
index 07cf6537..5368c2b7 100644
--- a/tests/event_sourcing/test_aggregate.py
+++ b/tests/event_sourcing/test_aggregate.py
@@ -2,53 +2,12 @@
 
 import pytest
 
-from event_sourcery import Event, EventStore, StreamId, StreamUUID
+from event_sourcery import EventStore, StreamId, StreamUUID
 from event_sourcery.event import Context
-from event_sourcery.event_sourcing import Aggregate, Repository
+from event_sourcery.event_sourcing import Repository
 from event_sourcery.exceptions import ConcurrentStreamWriteError
 
-
-class TurnedOn(Event):
-    pass
-
-
-class TurnedOff(Event):
-    pass
-
-
-class LightSwitch(Aggregate):
-    category = "light_switch"
-
-    class AlreadyTurnedOn(Exception):
-        pass
-
-    class AlreadyTurnedOff(Exception):
-        pass
-
-    def __init__(self) -> None:
-        self._shines = False
-
-    def __apply__(self, event: Event) -> None:
-        match event:
-            case TurnedOn():
-                self._shines = True
-            case TurnedOff():
-                self._shines = False
-
-    def turn_on(self) -> None:
-        if self._shines:
-            raise LightSwitch.AlreadyTurnedOn
-        self._emit(TurnedOn())
-
-    def turn_off(self) -> None:
-        if not self._shines:
-            raise LightSwitch.AlreadyTurnedOff
-
-        self._emit(TurnedOff())
-
-    @property
-    def shines(self) -> bool:
-        return self._shines
+from .light_switch import LightSwitch, TurnedOff, TurnedOn
 
 
 def test_light_switch_aggregate_logs_events() -> None:
@@ -84,15 +43,15 @@ def test_light_switch_changes_are_preserved_by_repository(
     repo: Repository[LightSwitch],
 ) -> None:
     uuid = StreamUUID(uuid4())
-    with repo.aggregate(uuid, LightSwitch()) as switch_first_incarnation:
-        switch_first_incarnation.turn_on()
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
 
-    with repo.aggregate(uuid, LightSwitch()) as switch_second_incarnation:
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
         try:
-            switch_second_incarnation.turn_on()
+            wrapped.aggregate.turn_on()
         except LightSwitch.AlreadyTurnedOn:
             # o mon Dieu, I made a mistake!
-            switch_second_incarnation.turn_off()
+            wrapped.aggregate.turn_off()
 
 
 def test_nothing_when_no_changes_on_aggregate(
@@ -111,16 +70,16 @@ def test_repository_supports_optimistic_locking(
     repo: Repository[LightSwitch],
 ) -> None:
     uuid = StreamUUID(uuid4())
-    with repo.aggregate(uuid, LightSwitch()) as switch_first_incarnation:
-        switch_first_incarnation.turn_on()
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
 
     with pytest.raises(ConcurrentStreamWriteError):
-        with repo.aggregate(uuid, LightSwitch()) as switch_second_incarnation:
-            with repo.aggregate(uuid, LightSwitch()) as switch_third_incarnation:
-                switch_second_incarnation.turn_off()
-                switch_third_incarnation.turn_off()
+        with repo.aggregate(uuid, LightSwitch()) as second:
+            with repo.aggregate(uuid, LightSwitch()) as third:
+                second.aggregate.turn_off()
+                third.aggregate.turn_off()
 
-    assert not switch_second_incarnation.shines
+    assert not second.aggregate.shines
 
 
 def test_context_is_attached_to_events_saved_by_repository(
@@ -132,16 +91,11 @@ class RequestContext(Context):
 
     uuid = StreamUUID(uuid4())
     ctx = RequestContext(user_id="user-123")
-    with repo.aggregate(uuid, LightSwitch(), context=ctx) as switch:
-        switch.turn_on()
+    with repo.aggregate(uuid, LightSwitch(), context=ctx) as wrapped:
+        wrapped.aggregate.turn_on()
 
     stream_id = StreamId(uuid, category=LightSwitch.category)
     events = list(event_store.load_stream(stream_id))
     assert len(events) == 1
     loaded_ctx = events[0].get_context(RequestContext)
     assert loaded_ctx.user_id == "user-123"
-
-
-@pytest.fixture()
-def repo(event_store: EventStore) -> Repository[LightSwitch]:
-    return Repository[LightSwitch](event_store)
diff --git a/tests/event_sourcing/test_wrapped_aggregate.py b/tests/event_sourcing/test_wrapped_aggregate.py
new file mode 100644
index 00000000..b653d7ef
--- /dev/null
+++ b/tests/event_sourcing/test_wrapped_aggregate.py
@@ -0,0 +1,153 @@
+from uuid import uuid4
+
+from event_sourcery import StreamId, StreamUUID
+from event_sourcery._event_store.event.dto import Context
+from event_sourcery.event_sourcing import Repository
+
+from .light_switch import LightSwitch
+
+
+def test_wrapped_aggregate_provides_access_to_aggregate(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
+        assert wrapped.aggregate.shines is True
+
+
+def test_wrapped_aggregate_is_new_when_no_prior_events(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.is_new is True
+
+
+def test_wrapped_aggregate_is_not_new_after_reload(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
+
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.is_new is False
+
+
+def test_wrapped_aggregate_is_new_even_after_emitting_events(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
+        assert wrapped.is_new is True
+
+
+def test_wrapped_aggregate_version_starts_at_zero_for_new(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.version == 0
+
+
+def test_wrapped_aggregate_version_reflects_stored_events(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
+
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.version == 1
+
+
+def test_wrapped_aggregate_version_includes_pending_changes(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.version == 0
+        wrapped.aggregate.turn_on()
+        assert wrapped.version == 1
+        wrapped.aggregate.turn_off()
+        assert wrapped.version == 2
+
+
+def test_wrapped_aggregate_version_combines_stored_and_pending(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
+
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.version == 1
+        wrapped.aggregate.turn_off()
+        assert wrapped.version == 2
+
+
+def test_wrapped_aggregate_exposes_stream_id(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.stream_id == StreamId(uuid, category=LightSwitch.category)
+
+
+def test_wrapped_aggregate_timestamps_are_none_for_new(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.created_at is None
+        assert wrapped.updated_at is None
+
+
+def test_wrapped_aggregate_timestamps_after_events_persisted(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
+
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.created_at is not None
+        assert wrapped.updated_at is not None
+        assert wrapped.created_at == wrapped.updated_at
+
+
+def test_wrapped_aggregate_updated_at_differs_from_created_at_with_multiple_events(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        wrapped.aggregate.turn_on()
+        wrapped.aggregate.turn_off()
+
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.created_at is not None
+        assert wrapped.updated_at is not None
+        assert wrapped.created_at <= wrapped.updated_at
+
+
+def test_wrapped_aggregate_exposes_context(
+    repo: Repository[LightSwitch],
+) -> None:
+    class RequestContext(Context):
+        user_id: str
+
+    uuid = StreamUUID(uuid4())
+    ctx = RequestContext(user_id="user-123")
+    with repo.aggregate(uuid, LightSwitch(), context=ctx) as wrapped:
+        assert wrapped.context == ctx
+        assert wrapped.get_context(RequestContext).user_id == "user-123"
+
+
+def test_wrapped_aggregate_context_is_default_when_not_provided(
+    repo: Repository[LightSwitch],
+) -> None:
+    uuid = StreamUUID(uuid4())
+    with repo.aggregate(uuid, LightSwitch()) as wrapped:
+        assert wrapped.context == Context()