diff --git a/README.md b/README.md
index e19e183..f5111b4 100644
--- a/README.md
+++ b/README.md
@@ -28,6 +28,7 @@ Ensure operations execute exactly once, even when called multiple times with the
 - **Async First** — built for asyncio applications
 - **Graceful Degradation** — high availability over strict exactly-once
 - **In-flight Reservation** — a retry that arrives while the original is still running waits for its result or gets a 409; the action runs once
+- **Request Fingerprints** — name the parameters that identify a request, and a key reused for a different one is refused instead of replayed
 - **Observability** — built-in metrics (hits, misses, collisions, latency)
 - **Bulk Operations** — efficient `get_many`, `save_many`, `delete_many`
 - **Redis Cluster Compatible** — non-transactional pipelines
diff --git a/docs/agents.md b/docs/agents.md
index 1a68139..5a3735a 100644
--- a/docs/agents.md
+++ b/docs/agents.md
@@ -32,15 +32,17 @@ method that sounds plausible.
 ## Scope
 
 **It does** cache the result of an async operation under a caller-supplied key, replay that
-result on a repeat call, and hold the key while the first caller's action runs: a second
-caller that arrives with the same key in that window waits for the first one's result, or is
-refused, instead of running the action too. It ships a Redis repository, a metrics protocol
-with a Prometheus implementation, a decorator that hides the whole flow, and Dishka providers
-that wire the pieces together.
+result on a repeat call, hold the key while the first caller's action runs — a second caller
+that arrives with the same key in that window waits for the first one's result, or is
+refused, instead of running the action too — and, given a fingerprint of the request, refuse
+a key that comes back for a different request instead of replaying the first one's result.
+It ships a Redis repository, a metrics protocol with a Prometheus implementation, a decorator
+that hides the whole flow, and Dishka providers that wire the pieces together.
 
 **It does not** derive the key — the caller supplies it, and the request body is not part of
-it; it does not roll anything back; it does not cache failures; it does not retry; it has no
-sync API; and it stores nothing but JSON. The reservation is a lease, not a lock: an action
+it unless you name the parameters that fingerprint the request; it does not roll anything
+back; it does not cache failures; it does not retry; it has no sync API; and it stores
+nothing but JSON. The reservation is a lease, not a lock: an action
 that outlives its lease can run twice, and when storage is down the action runs unreserved.
 It is a result cache with an in-flight reservation, not a distributed transaction.
 
@@ -49,9 +51,10 @@ It is a result cache with an in-flight reservation, not a distributed transactio
 Four nouns and one flow.
 
 * **`IdempotencyRecord`** — a frozen Pydantic model: `operation`, `idempotency_key`, the
-  JSON `result`, `created_at`, `expires_at`, and a `status` that is `"completed"` for a
-  stored result and `"pending"` for an in-flight reservation. `expires_at` is what decides
-  whether a record is still a hit; on a pending record it is the lease.
+  JSON `result`, `created_at`, `expires_at`, a `status` that is `"completed"` for a stored
+  result and `"pending"` for an in-flight reservation, and an optional `fingerprint` of the
+  request the record was made for. `expires_at` is what decides whether a record is still a
+  hit; on a pending record it is the lease.
 * **`AsyncIdempotencyRepository`** — the storage protocol: `get` / `save` / `replace` /
   `delete` and the bulk twins of `get`, `save` and `delete`. `save` is a write that fails if
   the key is already there; that failure, `IdempotencyKeyCollisionError`, is how a
@@ -63,10 +66,11 @@ Four nouns and one flow.
 * **`AsyncIdempotencyCoordinator`** — the flow: reserve the key by writing a pending
   record under `SET NX` with the lease as its TTL. If that succeeds, run the action, encode
   the result and `replace` the reservation with the completed record. If it fails, read
-  what holds the key: a completed record is decoded and returned; a pending one means
-  another caller is in flight, and `in_flight` decides — `"wait"` polls until the record
-  arrives, `"raise"` raises `IdempotencyInProgressError`, `"run"` is the old flow of read,
-  run, `SET NX` and adopt the winner's result on a collision. A record whose `expires_at`
+  what holds the key: a record whose `fingerprint` differs from the caller's raises
+  `IdempotencyKeyReuseError`, pending or not; a completed record is decoded and returned; a
+  pending one means another caller is in flight, and `in_flight` decides — `"wait"` polls
+  until the record arrives, `"raise"` raises `IdempotencyInProgressError`, `"run"` is the
+  old flow of read, run, `SET NX` and adopt the winner's result on a collision. A record whose `expires_at`
   has passed is a miss even if the repository handed it back; an expired lease is an
   abandoned reservation. An action that raises deletes its reservation. Storage and decode
   failures are swallowed and the action runs, which is the deliberate trade of exactly-once
@@ -111,6 +115,7 @@ class CreateOrder:
         adapter=PydanticResultAdapter(OrderDTO),
         ttl_seconds=3600,
         infra_param="coordinator",
+        fingerprint_params=("dto",),
     )
     async def execute(self, dto: CreateOrderDTO, *, idempotency_key: str | None = None) -> OrderDTO:
         order = await self._orders.create(dto)
@@ -121,9 +126,11 @@ class CreateOrder:
 the call site has to name it. The decorator reads it from the keyword arguments, or from the
 positional ones when the parameter can be passed that way. `infra_param="coordinator"` names
 the attribute rather than leaving the decorator to find a coordinator by type.
+`fingerprint_params=("dto",)` makes `dto` part of what the key identifies: the same key with
+a different `dto` raises `IdempotencyKeyReuseError` instead of replaying the first order.
 
 The same call without the decorator — the five leading arguments are positional-only, and
-everything after them is forwarded to the action:
+everything after them is forwarded to the action except `idempotency_fingerprint`:
 
 ```python
 result = await coordinator.coordinate(
@@ -133,6 +140,7 @@ result = await coordinator.coordinate(
     PydanticResultAdapter(OrderDTO),
     self.execute_uncached,               # the action
     dto,                                 # *args and **kwargs go to the action
+    idempotency_fingerprint=fingerprint_of(dto=dto),   # optional; None means key only
 )
 ```
 
@@ -142,7 +150,8 @@ result = await coordinator.coordinate(
 
 | Name | Signature | What it is |
 |---|---|---|
-| `async_idempotent` | `(operation, adapter, ttl_seconds=None, key_param="idempotency_key", infra_param=None)` | decorator for an async function or method |
+| `async_idempotent` | `(operation, adapter, ttl_seconds=None, key_param="idempotency_key", infra_param=None, fingerprint_params=None)` | decorator for an async function or method; `fingerprint_params` names the parameters whose values identify the request |
+| `fingerprint_of` | `(**values)` | SHA-256 hex of the JSON form of the named values, keys sorted; what the decorator computes from `fingerprint_params`, for a caller of `coordinate()` |
 | `AsyncIdempotencyCoordinator` | `(repository, domain_service, operation_ttls=None, metrics=None, enabled=True, in_flight="wait", in_flight_lease_seconds=30)` | the flow; `operation_ttls` is `dict[str, int]` in seconds; `enabled=False` runs the action and nothing else; `in_flight` is `"wait"`, `"raise"` or `"run"` |
 | `IdempotencyDomainService` | `(*, default_ttl_minutes=60, min_ttl_seconds=60, max_ttl_seconds=2592000)` | record factory and TTL bounds; keyword-only, defaults from `core.constants` |
 | `IdempotencyRecord` | frozen Pydantic model | the cached result |
@@ -154,7 +163,7 @@ result = await coordinator.coordinate(
 | `VoidResultAdapter` | `()` | stores JSON `null`, decodes back to `None` |
 | `IdempotencyMetricsProtocol` | runtime-checkable `Protocol` | metrics contract |
 | `NoOpIdempotencyMetrics` | `()` | the default collector |
-| `IdempotencyError` and its six subclasses | | see [Errors](#errors) |
+| `IdempotencyError` and its seven subclasses | | see [Errors](#errors) |
 
 ### Not exported from the root
 
@@ -170,9 +179,9 @@ result = await coordinator.coordinate(
 
 | Method | Returns | Notes |
 |---|---|---|
-| `AsyncIdempotencyCoordinator.coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, **kwargs)` | `T` | never raises for storage or decode trouble; raises `IdempotencyInProgressError` when the key is in flight and `in_flight` says so |
-| `IdempotencyDomainService.create_record(operation, idempotency_key, result, *, ttl_minutes=None)` | `IdempotencyRecord` | raises `IdempotencyInvalidTTLError`, `IdempotencyValidationError` |
-| `IdempotencyDomainService.create_pending_record(operation, idempotency_key, *, lease_seconds)` | `IdempotencyRecord` | the reservation; not held to the TTL bounds; raises `IdempotencyValidationError` |
+| `AsyncIdempotencyCoordinator.coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, idempotency_fingerprint=None, **kwargs)` | `T` | never raises for storage or decode trouble; raises `IdempotencyInProgressError` when the key is in flight and `in_flight` says so, `IdempotencyKeyReuseError` when the record's fingerprint differs from `idempotency_fingerprint` |
+| `IdempotencyDomainService.create_record(operation, idempotency_key, result, *, ttl_minutes=None, fingerprint=None)` | `IdempotencyRecord` | raises `IdempotencyInvalidTTLError`, `IdempotencyValidationError` |
+| `IdempotencyDomainService.create_pending_record(operation, idempotency_key, *, lease_seconds, fingerprint=None)` | `IdempotencyRecord` | the reservation; not held to the TTL bounds; raises `IdempotencyValidationError` |
 | `IdempotencyDomainService.validate_record(record)` | `None` | raises `IdempotencyRecordExpiredError`; the coordinator calls it on every record it reads |
 
 ### Record
@@ -184,8 +193,9 @@ result = await coordinator.coordinate(
 | `result` | `JsonValue` | whatever the adapter encoded; `null` for a void result |
 | `created_at` / `expires_at` | `datetime` | UTC, set by `create` and `pending`; on a pending record `expires_at` is the lease |
 | `status` | `"pending" \| "completed"` | `"completed"` unless said otherwise, which is how a record written before the field existed reads |
-| `IdempotencyRecord.create(operation, idempotency_key, result, ttl_seconds)` | `IdempotencyRecord` | classmethod; `ttl_seconds` is a `float` here |
-| `IdempotencyRecord.pending(operation, idempotency_key, lease_seconds)` | `IdempotencyRecord` | classmethod; the reservation, `result` is `null` |
+| `fingerprint` | `str \| None` | what the caller said the request was; `None` — also what a record written before the field existed reads as — never raises |
+| `IdempotencyRecord.create(operation, idempotency_key, result, ttl_seconds, fingerprint=None)` | `IdempotencyRecord` | classmethod; `ttl_seconds` is a `float` here |
+| `IdempotencyRecord.pending(operation, idempotency_key, lease_seconds, fingerprint=None)` | `IdempotencyRecord` | classmethod; the reservation, `result` is `null` |
 | `.is_pending` | `bool` | `status == "pending"` |
 | `.is_expired` | `bool` | `now >= expires_at` |
 | `.ttl_seconds` | `float` | remaining, `0.0` once expired |
@@ -228,7 +238,7 @@ second caller records on finding the pending record: a waiter then records a hit
 result arrives, a refused caller records nothing more and raises. In `"run"` mode it is the
 loser's `SET NX` failing after both ran, as before. The error types the coordinator reports
 are `storage_get_error`, `storage_reserve_error`, `storage_save_error`,
-`storage_release_error` and `record_validation_error`.
+`storage_release_error`, `record_validation_error` and `key_reuse`.
 
 ### Settings and Dishka
 
@@ -266,10 +276,17 @@ fields existed is read as enabled, `"wait"` and 30 seconds.
 
 ## Rules that hold or break the code
 
-1. **The key is the whole identity; the arguments are not.** Nothing hashes the request
-   body. Two calls with the same `operation` and `idempotency_key` and different payloads
-   replay the first result. A key must be unique per intended effect, and one client request
-   must not reuse a key across two different operations' worth of work.
+1. **The key is the whole identity; the arguments are not, unless you say which ones are.**
+   Nothing hashes the request body on its own: two calls with the same `operation` and
+   `idempotency_key` and different payloads replay the first result. Name the parameters
+   that identify the request — `fingerprint_params=("dto",)` on the decorator, or
+   `idempotency_fingerprint=fingerprint_of(dto=dto)` on `coordinate()` — and the fingerprint
+   is stored with the record; the same key back with a different fingerprint raises
+   `IdempotencyKeyReuseError` (422 in HTTP terms) instead of replaying, from a completed
+   record and from a pending one alike, before the action runs. Either side without a
+   fingerprint means no comparison: a record written without one, or before the field
+   existed, never raises, and a caller without one gets the key-only behaviour. A key must
+   still be unique per intended effect; the fingerprint is the guard for when it is not.
 2. **A second caller with the same key does not run your business logic while the first is
    in flight — unless you ask for that.** The coordinator reserves the key with a pending
    record (`SET NX`, TTL `in_flight_lease_seconds`, 30 s by default) before the action and
@@ -361,6 +378,19 @@ fields existed is read as enabled, `"wait"` and 30 seconds.
     behaviour, but `JsonResultAdapter` and `VoidResultAdapter` replay the `None`. Roll out
     with `in_flight="run"` and switch once every instance is on the new version, or accept
     the window.
+23. **Fingerprint what identifies the request, not what varies between retries.** The
+    decorator binds the call to the signature with defaults applied, so an argument passed
+    at its default and one left out agree, turns the named values into their JSON form
+    (Pydantic models, dataclasses, UUIDs, datetimes and Decimals included), sorts keys and
+    hashes. A timestamp, a trace id or `self` in `fingerprint_params` makes every honest
+    retry a key reuse. A name the function does not have raises `TypeError` at decoration;
+    a value with no JSON form raises `PydanticSerializationError` at call time.
+24. **`idempotency_fingerprint` is the coordinator's keyword, not the action's.** It is the
+    one keyword `coordinate()` keeps for itself, so an action cannot have a parameter of
+    that name; the decorator passes it only when `fingerprint_params` is set. In
+    `in_flight="run"` mode the fingerprint is checked on the read before the action; a
+    mismatch discovered on the collision after the action is logged and counted as
+    `key_reuse`, and the caller keeps its own result, because the side effect has happened.
 
 ## Common mistakes
 
@@ -424,6 +454,27 @@ async def charge(self, dto, *, idempotency_key: str | None = None) -> ChargeDTO:
 coordinator = AsyncIdempotencyCoordinator(repository, service, in_flight_lease_seconds=60)
 ```
 
+```python
+# WRONG — a key derived from the order, reused for a second, different request on the same
+# order: the first charge is replayed and nothing says so
+@async_idempotent(operation="payment.charge", adapter=PydanticResultAdapter(ChargeDTO))
+async def charge(self, dto: ChargeDTO, *, idempotency_key: str | None = None) -> ChargeDTO: ...
+
+await charge(ChargeDTO(amount=1999), idempotency_key=f"order-{order.id}")
+await charge(ChargeDTO(amount=5), idempotency_key=f"order-{order.id}")     # -> the 1999 charge
+
+# RIGHT — name what identifies the request, and the second call is refused before it runs
+@async_idempotent(
+    operation="payment.charge", adapter=PydanticResultAdapter(ChargeDTO), fingerprint_params=("dto",)
+)
+async def charge(self, dto: ChargeDTO, *, idempotency_key: str | None = None) -> ChargeDTO: ...
+
+try:
+    return await use_case.charge(dto, idempotency_key=key)
+except IdempotencyKeyReuseError:
+    raise HTTPException(422, detail="this Idempotency-Key was already used for a different request")
+```
+
 ```python
 # WRONG — expecting the coordinator to tell you Redis is down
 try:
@@ -463,13 +514,15 @@ All derive from `IdempotencyError`, which is exported alongside them.
 | `IdempotencyKeyCollisionError` | `(operation, key)` | `save` found the key already there; `key` is a `str`, or a `list[str]` from `save_many`. Carries `.operation` and `.key` |
 | `IdempotencyRecordExpiredError` | `(operation, key)` | `validate_record` was given an expired record. The coordinator raises it internally and turns it into a miss; through the repository or your own call to `validate_record` you meet it directly |
 | `IdempotencyInProgressError` | `(operation, key)` | another call with the same key is still running its action. `coordinate()` and the decorator raise it at once with `in_flight="raise"`, and after a whole lease of waiting with `"wait"`. Carries `.operation` and `.key` |
+| `IdempotencyKeyReuseError` | `(operation, key, stored_fingerprint, fingerprint)` | the record under the key was made for a different request. `coordinate()` and the decorator raise it before the action runs, from a completed record and from a pending one. Carries all four |
 | `IdempotencyStorageError` | `(message, operation=None, original_error=None)` | the backend failed. Carries `.operation` and `.original_error` |
 | `IdempotencyValidationError` | `(message, errors=None)` | an identifier or a result failed validation. `.errors` holds the Pydantic error list when there is one |
 | `IdempotencyInvalidTTLError` | `(ttl_seconds, min_ttl, max_ttl)` | the TTL is outside the domain service's range. Carries all three |
 
-Through `coordinate()` and the decorator only `IdempotencyInProgressError` reaches the
-caller: the collision is resolved into a wait or the winner's result, and every other one is
-logged, counted and swallowed. The rest are the contract of the repository and the domain
+Through `coordinate()` and the decorator only `IdempotencyInProgressError` and
+`IdempotencyKeyReuseError` reach the caller — both are about the request, not about storage:
+the collision is resolved into a wait or the winner's result, and every other one is logged,
+counted and swallowed. The rest are the contract of the repository and the domain
 service, which is where you meet them if you drive those directly.
 
 ## Documentation map
@@ -480,7 +533,7 @@ Fetch a page when the task is the one named beside it.
 |---|---|
 | [Home](index.md) | placing the library — what it is for, the four shapes of caller |
 | [Quick Start](quickstart.md) | the first integration, and what makes a good key |
-| [User Guide](user_guide.md) | in-flight handling and the lease, bulk operations, graceful degradation, Dishka wiring, worked services |
+| [User Guide](user_guide.md) | in-flight handling and the lease, fingerprints and key reuse, bulk operations, graceful degradation, Dishka wiring, worked services |
 | [Architecture](architecture.md) | the layers, the request-flow diagrams, the cluster reasoning |
 | [API Reference](api_reference.md) | an exact field, default or constructor argument |
 | [Testing](testing_conventions.md) | writing tests against this library, or contributing to it |
diff --git a/docs/api_reference.md b/docs/api_reference.md
index 1351127..e01373d 100644
--- a/docs/api_reference.md
+++ b/docs/api_reference.md
@@ -21,9 +21,10 @@ A frozen Pydantic model representing an idempotency result. Inherits from `Idemp
   - `created_at` (datetime): When the record was created.
   - `expires_at` (datetime): When the record will expire. On a pending record this is the in-flight lease.
   - `status` (`"pending" | "completed"`, default `"completed"`): `"pending"` while the action runs under an in-flight reservation, `"completed"` once the result is stored. A record written before the field existed reads as completed.
+  - `fingerprint` (`str | None`, default `None`): Fingerprint of the request the record was made for. A hit with a different one is a key reuse; `None` — also how a record written before the field existed reads — never raises.
 - **Methods**:
-  - `create(operation: str, idempotency_key: str, result: JsonValue, ttl_seconds: float) -> IdempotencyRecord`: Class method to create a new record.
-  - `pending(operation: str, idempotency_key: str, lease_seconds: float) -> IdempotencyRecord`: Class method to create the in-flight reservation; `result` is `null`.
+  - `create(operation: str, idempotency_key: str, result: JsonValue, ttl_seconds: float, fingerprint: str | None = None) -> IdempotencyRecord`: Class method to create a new record.
+  - `pending(operation: str, idempotency_key: str, lease_seconds: float, fingerprint: str | None = None) -> IdempotencyRecord`: Class method to create the in-flight reservation; `result` is `null`.
   - `is_pending`: Property returning `True` for an in-flight reservation.
   - `is_expired`: Property returning `True` if current time is after `expires_at`.
   - `ttl_seconds`: Property returning remaining TTL in seconds.
@@ -38,8 +39,8 @@ Service for creating and validating records.
   - *Note*: These three defaults live in `idempotency_kit.core.constants` and are also the field defaults of `BaseIdempotencySettings`.
   - *Note*: Constructor validates that `default_ttl_minutes` (converted to seconds) is within the `[min_ttl_seconds, max_ttl_seconds]` range.
 - **Methods**:
-  - `create_record(operation, idempotency_key, result, *, ttl_minutes=None)`: Creates a new `IdempotencyRecord` with validation and TTL management.
-  - `create_pending_record(operation, idempotency_key, *, lease_seconds)`: Creates the in-flight reservation. The lease is not held to the TTL bounds; it must be at least a second.
+  - `create_record(operation, idempotency_key, result, *, ttl_minutes=None, fingerprint=None)`: Creates a new `IdempotencyRecord` with validation and TTL management.
+  - `create_pending_record(operation, idempotency_key, *, lease_seconds, fingerprint=None)`: Creates the in-flight reservation. The lease is not held to the TTL bounds; it must be at least a second.
   - `validate_record(record)`: Validates that a record is still usable, raising `IdempotencyRecordExpiredError` if not. The coordinator calls it on every record it reads.
 
 ### AsyncIdempotencyCoordinator
@@ -55,7 +56,21 @@ The flow: reserve the key, run the action, store the result; replay a stored res
   - `in_flight_lease_seconds` (int, default: 30): How long a reservation is held before it counts as abandoned, and the longest a waiting caller waits. Must be at least 1 and longer than the action can take.
   - *Note*: The constructor raises `TypeError` when the repository has no `replace` and `in_flight` is not `"run"`.
 - **Methods**:
-  - `coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, **kwargs)`: Runs the flow and returns the action's result type. Never raises for storage or decode trouble; raises `IdempotencyInProgressError` when the key is in flight and `in_flight` says so.
+  - `coordinate(operation, idempotency_key, ttl_seconds, adapter, action, /, *args, idempotency_fingerprint=None, **kwargs)`: Runs the flow and returns the action's result type. Everything after the five positional-only arguments goes to the action except `idempotency_fingerprint`, the fingerprint of the request, which is stored with the record and compared on a hit. Never raises for storage or decode trouble; raises `IdempotencyInProgressError` when the key is in flight and `in_flight` says so, and `IdempotencyKeyReuseError` when the record's fingerprint differs from the caller's.
+
+### async_idempotent
+The same flow as a decorator: it finds the key in the call's arguments and the coordinator in the call's arguments or on `self`, then delegates to `coordinate()`.
+
+- **Parameters**:
+  - `operation` (str): Operation name.
+  - `adapter` (ResultAdapter): Encodes the result for storage and decodes it back.
+  - `ttl_seconds` (int, optional): TTL for the record; the coordinator's `operation_ttls` win over it.
+  - `key_param` (str, default: `"idempotency_key"`): Name of the parameter holding the key. Read from the keyword arguments, or from the positional ones when the parameter can be passed that way.
+  - `infra_param` (str, optional): Name of the argument or attribute holding the coordinator; without it the decorator searches by type.
+  - `fingerprint_params` (tuple[str, ...], optional): Names of the parameters whose bound values, defaults applied, identify the request. Their fingerprint is computed with `fingerprint_of` and passed to `coordinate()` as `idempotency_fingerprint`. A name the function does not have raises `TypeError` at decoration time.
+
+### fingerprint_of
+`fingerprint_of(**values) -> str`: the SHA-256 hex digest of the JSON form of the named values, keys sorted. Pydantic models, dataclasses, UUIDs, datetimes and Decimals are serialised the way Pydantic serialises them; a value with no JSON form raises `pydantic_core.PydanticSerializationError`. It is what the decorator computes from `fingerprint_params`, for a caller of `coordinate()` to use directly.
 
 ### AsyncIdempotencyRepository (Protocol)
 Interface for idempotency storage.
@@ -102,6 +117,7 @@ Redis implementation of the repository protocol.
   - `key` can be a single `str` or a `list[str]` for bulk operations.
 - **`IdempotencyRecordExpiredError(operation, key)`**: Raised when record exists but is expired.
 - **`IdempotencyInProgressError(operation, key)`**: Raised by `coordinate()` and the decorator when another call with the same key is still running its action.
+- **`IdempotencyKeyReuseError(operation, key, stored_fingerprint, fingerprint)`**: Raised by `coordinate()` and the decorator when the record under the key was made for a request with a different fingerprint.
 - **`IdempotencyStorageError(message, operation, original_error)`**: Raised on storage failure.
 - **`IdempotencyValidationError(message, errors=None)`**: Raised for invalid input (e.g. empty key, too long string).
   - `errors` (list, optional): Detailed Pydantic validation errors.
diff --git a/docs/quickstart.md b/docs/quickstart.md
index cf49e6d..47d0390 100644
--- a/docs/quickstart.md
+++ b/docs/quickstart.md
@@ -144,6 +144,12 @@ idempotency_key = "create_order"  # Same for all requests
 idempotency_key = str(datetime.now())  # Changes every millisecond
 ```
 
+The key is the whole identity: the same key with a different payload replays the first result.
+To have the payload count, name the parameters that identify the request with
+`fingerprint_params=("dto",)` on `@async_idempotent`; a key reused for a different request then
+raises `IdempotencyKeyReuseError` — see
+[Key reuse and fingerprints](user_guide.md#key-reuse-and-fingerprints) in the User Guide.
+
 ### Operation Name
 
 A string identifying the type of operation:
diff --git a/docs/user_guide.md b/docs/user_guide.md
index 66ea580..f0f2e9d 100644
--- a/docs/user_guide.md
+++ b/docs/user_guide.md
@@ -159,6 +159,81 @@ to be migrated. Two things to know before turning the default on across a fleet:
   `in_flight="run"` and switch to `"wait"` once every instance is on the new version, or
   accept that window.
 
+## Key reuse and fingerprints
+
+The key is the identity, and the request body is not part of it: two calls with the same
+`operation` and `idempotency_key` and different payloads replay the first result. For a
+well-behaved client that never happens. It happens with a client that reuses keys by mistake
+— a key derived from the order id, then a second, different operation on the same order — and
+without a fingerprint the library answers such a request with a result for a different
+request, and nothing says so.
+
+Name the parameters that identify the request and the decorator fingerprints them:
+
+```python
+from idempotency_kit import IdempotencyKeyReuseError, PydanticResultAdapter, async_idempotent
+
+class ChargeCard:
+    @async_idempotent(
+        operation="payment.charge",
+        adapter=PydanticResultAdapter(ChargeDTO),
+        fingerprint_params=("dto",),
+    )
+    async def execute(self, dto: ChargeRequest, *, idempotency_key: str | None = None) -> ChargeDTO:
+        return await self._psp.charge(dto)
+```
+
+The named values are bound to the call with defaults applied, turned into their JSON form —
+Pydantic models, dataclasses, UUIDs, datetimes and Decimals included — serialised with sorted
+keys, and hashed with SHA-256. The fingerprint is stored with the record. The same key coming
+back with a different fingerprint raises `IdempotencyKeyReuseError` before the action runs,
+whether the record is completed or still pending under an in-flight reservation, so a
+retry with another payload is refused at once rather than handed someone else's result after
+waiting. The error carries `operation`, `key`, `stored_fingerprint` and `fingerprint`; map
+it to HTTP 422, which is what Stripe does:
+
+```python
+@app.post("/charges")
+async def charge(dto: ChargeRequest, idempotency_key: str | None = Header(None, alias="Idempotency-Key")):
+    try:
+        return await use_case.execute(dto, idempotency_key=idempotency_key)
+    except IdempotencyKeyReuseError:
+        raise HTTPException(422, detail="this Idempotency-Key was already used for a different request")
+```
+
+Without the decorator, compute the fingerprint with `fingerprint_of` — the same function the
+decorator uses, so both paths agree — and pass it to `coordinate()` as its one keyword:
+
+```python
+from idempotency_kit import fingerprint_of
+
+result = await coordinator.coordinate(
+    "payment.charge",
+    idempotency_key,
+    3600,
+    PydanticResultAdapter(ChargeDTO),
+    self._psp.charge,
+    dto,
+    idempotency_fingerprint=fingerprint_of(dto=dto),
+)
+```
+
+**Either side missing means no comparison.** A record without a fingerprint — written by a
+caller that passed none, or before the field existed — never raises, and a caller without a
+fingerprint gets the key-only behaviour. Nothing changes for anyone who does not opt in, and
+no record needs migrating.
+
+**Fingerprint what identifies the request, not what varies between retries.** A timestamp,
+a trace id or `self` in `fingerprint_params` turns every honest retry into a key reuse. A
+name the function does not have raises `TypeError` at decoration time; a value with no JSON
+form raises `PydanticSerializationError` when the call is made.
+
+**With `in_flight="run"`** the check is on the read, before the action. If the mismatch only
+shows up on the collision after both callers ran, it is logged and counted as `key_reuse`
+and the caller keeps its own result: raising then would make the caller retry a completed
+operation. Every refusal is counted under `record_error(operation, "key_reuse")`; a client
+that reuses keys is worth an alert.
+
 ## Advanced Use Cases
 
 ### Custom TTL
@@ -379,6 +454,7 @@ The library defines several exceptions to handle various idempotency scenarios:
 
 - **`IdempotencyKeyCollisionError`**: Raised by `repository.save()` when you try to save a result for a key that already exists. This typically means another identical request is either being processed or has already finished.
 - **`IdempotencyInProgressError`**: Raised by `coordinator.coordinate()` and the decorator when another call with the same key is still running its action — at once with `in_flight="raise"`, after a whole lease of waiting with `in_flight="wait"`. Map it to HTTP 409.
+- **`IdempotencyKeyReuseError`**: Raised by `coordinator.coordinate()` and the decorator when the record under the key was made for a request with a different fingerprint. Carries both fingerprints. Map it to HTTP 422.
 - **`IdempotencyRecordExpiredError`**: Raised by `service.validate_record()` if the record exists but its TTL has passed.
 - **`IdempotencyInvalidTTLError`**: Raised by `service.create_record()` if the requested TTL is outside the allowed range (configured in `IdempotencyDomainService`).
 - **`IdempotencyValidationError`**: Raised by `service.create_record()` if validation of `operation` or `idempotency_key` fails (e.g., empty string or too long).
@@ -390,8 +466,9 @@ The library defines several exceptions to handle various idempotency scenarios:
 1. **Natural Keys**: Use natural unique identifiers as idempotency keys if possible (e.g., `order_id`, `message_id`).
 2. **Atomic Operations**: Always save the result to the cache *after* the business logic has successfully completed.
 3. **Lease Longer Than the Action**: Set `in_flight_lease_seconds` above the longest the action can take, timeouts and retries included; a reservation that expires mid-run lets the next caller run the action again.
-4. **Pydantic Support**: The library works best with Pydantic models. Use `model_dump(mode="json")` when saving and `**cached.result` when restoring.
-5. **Graceful Degradation**: Decide whether your service should fail if idempotency storage is down. For most high-availability services, it's better to log an error and proceed (at-least-once delivery) than to crash (exactly-once requirement).
+4. **Fingerprint the Request**: Name the parameters that identify the request in `fingerprint_params`, so a key reused by mistake for a different request is refused with `IdempotencyKeyReuseError` rather than answered with the first result.
+5. **Pydantic Support**: The library works best with Pydantic models. Use `model_dump(mode="json")` when saving and `**cached.result` when restoring.
+6. **Graceful Degradation**: Decide whether your service should fail if idempotency storage is down. For most high-availability services, it's better to log an error and proceed (at-least-once delivery) than to crash (exactly-once requirement).
 
 ## Production Examples
 
diff --git a/idempotency_kit/__init__.py b/idempotency_kit/__init__.py
index 9d8d843..f414e62 100644
--- a/idempotency_kit/__init__.py
+++ b/idempotency_kit/__init__.py
@@ -11,10 +11,12 @@
     IdempotencyInProgressError,
     IdempotencyInvalidTTLError,
     IdempotencyKeyCollisionError,
+    IdempotencyKeyReuseError,
     IdempotencyRecordExpiredError,
     IdempotencyStorageError,
     IdempotencyValidationError,
 )
+from .core.fingerprint import fingerprint_of
 from .core.models.entities import IdempotencyIdentifiers, IdempotencyRecord
 from .core.protocols.adapter import ResultAdapter
 from .core.protocols.aio.repository import AsyncIdempotencyRepository
@@ -31,6 +33,7 @@
     "IdempotencyInProgressError",
     "IdempotencyInvalidTTLError",
     "IdempotencyKeyCollisionError",
+    "IdempotencyKeyReuseError",
     "IdempotencyMetricsProtocol",
     "IdempotencyRecord",
     "IdempotencyRecordExpiredError",
@@ -42,4 +45,5 @@
     "ResultAdapter",
     "VoidResultAdapter",
     "async_idempotent",
+    "fingerprint_of",
 ]
diff --git a/idempotency_kit/core/decorators/aio/idempotent.py b/idempotency_kit/core/decorators/aio/idempotent.py
index 1446712..a8ab18f 100644
--- a/idempotency_kit/core/decorators/aio/idempotent.py
+++ b/idempotency_kit/core/decorators/aio/idempotent.py
@@ -4,6 +4,7 @@
 from collections.abc import Awaitable, Callable
 from typing import Any, TypeVar
 
+from idempotency_kit.core.fingerprint import fingerprint_of
 from idempotency_kit.core.protocols.adapter import ResultAdapter
 from idempotency_kit.core.services.aio.coordinator import AsyncIdempotencyCoordinator
 
@@ -26,12 +27,40 @@ def _positional_index(func: Callable[..., Any], key_param: str) -> int | None:
     return None
 
 
+def _fingerprint_resolver(
+    func: Callable[..., Any], fingerprint_params: tuple[str, ...] | None
+) -> Callable[[tuple[Any, ...], dict[str, Any]], str] | None:
+    """Build the function that fingerprints a call from the named parameters, or ``None`` when there are none.
+
+    Resolved at decoration time so that a parameter the function does not have, or a
+    signature ``inspect`` cannot describe, fails at import rather than on the first call.
+    """
+    if not fingerprint_params:
+        return None
+    try:
+        signature = inspect.signature(func)
+    except (TypeError, ValueError) as e:
+        raise TypeError(f"fingerprint_params needs a signature inspect can describe; {func!r} has none") from e
+    unknown = [name for name in fingerprint_params if name not in signature.parameters]
+    if unknown:
+        raise TypeError(f"fingerprint_params names parameters {func.__qualname__} does not have: {unknown}")
+
+    def resolve(args: tuple[Any, ...], kwargs: dict[str, Any]) -> str:
+        # Defaults applied, so an argument passed at its default and one left out agree.
+        bound = signature.bind(*args, **kwargs)
+        bound.apply_defaults()
+        return fingerprint_of(**{name: bound.arguments[name] for name in fingerprint_params})
+
+    return resolve
+
+
 def async_idempotent(
     operation: str,
     adapter: ResultAdapter[T],
     ttl_seconds: int | None = None,
     key_param: str = "idempotency_key",
     infra_param: str | None = None,
+    fingerprint_params: tuple[str, ...] | None = None,
 ) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]:
     """Decorator for asynchronous idempotent operations.
 
@@ -48,10 +77,15 @@ def async_idempotent(
             be passed positionally.
         infra_param: Optional name of the argument or attribute containing AsyncIdempotencyCoordinator.
             If not provided, searches for AsyncIdempotencyCoordinator by type.
+        fingerprint_params: Names of the parameters whose values identify the request. Their
+            bound values, defaults applied, are hashed with ``fingerprint_of`` and stored with
+            the record; a later call under the same key with a different fingerprint raises
+            ``IdempotencyKeyReuseError``. ``None`` means the key alone is the identity.
     """
 
     def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
         key_index = _positional_index(func, key_param)
+        fingerprint = _fingerprint_resolver(func, fingerprint_params)
 
         @functools.wraps(func)
         async def wrapper(*args: Any, **kwargs: Any) -> T:
@@ -78,6 +112,8 @@ async def wrapper(*args: Any, **kwargs: Any) -> T:
                 return await func(*args, **kwargs)
 
             # 3. Delegate to coordinator
+            if fingerprint is not None:
+                kwargs = {"idempotency_fingerprint": fingerprint(args, kwargs), **kwargs}
             return await coordinator.coordinate(
                 operation,
                 idempotency_key,
diff --git a/idempotency_kit/core/exceptions.py b/idempotency_kit/core/exceptions.py
index 24e59b8..f4464ca 100644
--- a/idempotency_kit/core/exceptions.py
+++ b/idempotency_kit/core/exceptions.py
@@ -63,3 +63,17 @@ def __init__(self, operation: str, key: str) -> None:
         self.operation = operation
         self.key = key
         super().__init__(f"Idempotency record for operation '{operation}', key '{key}' is still in flight")
+
+
+class IdempotencyKeyReuseError(IdempotencyError):
+    """Raised when a key is reused for a request with a different fingerprint."""
+
+    def __init__(self, operation: str, key: str, stored_fingerprint: str, fingerprint: str) -> None:
+        self.operation = operation
+        self.key = key
+        self.stored_fingerprint = stored_fingerprint
+        self.fingerprint = fingerprint
+        super().__init__(
+            f"Idempotency key '{key}' for operation '{operation}' was used for a different request: "
+            f"stored fingerprint '{stored_fingerprint}', got '{fingerprint}'"
+        )
diff --git a/idempotency_kit/core/fingerprint.py b/idempotency_kit/core/fingerprint.py
new file mode 100644
index 0000000..134d765
--- /dev/null
+++ b/idempotency_kit/core/fingerprint.py
@@ -0,0 +1,22 @@
+"""The request fingerprint: what makes two calls under one key the same request."""
+
+import hashlib
+import json
+from typing import Any
+
+from pydantic_core import to_jsonable_python
+
+
+def fingerprint_of(**values: Any) -> str:
+    """Hash the named values into the fingerprint of a request.
+
+    Pydantic models, dataclasses, UUIDs, datetimes, Decimals and the JSON types are turned
+    into JSON-able Python first, keys are sorted, and the SHA-256 hex digest is returned, so
+    two equal payloads built in different orders agree. This is what ``async_idempotent``
+    computes from ``fingerprint_params``; a caller of ``coordinate()`` uses it directly.
+
+    Raises:
+        pydantic_core.PydanticSerializationError: a value has no JSON form
+    """
+    canonical = json.dumps(to_jsonable_python(values), sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
diff --git a/idempotency_kit/core/models/entities.py b/idempotency_kit/core/models/entities.py
index a4cc256..9f7e961 100644
--- a/idempotency_kit/core/models/entities.py
+++ b/idempotency_kit/core/models/entities.py
@@ -55,6 +55,13 @@ class IdempotencyRecord(IdempotencyIdentifiers):
         description="'pending' while the action runs under a reservation, 'completed' once the result is stored",
     )
 
+    # What the caller said the request was; None means the record carries no fingerprint
+    # and a hit never raises for it.
+    fingerprint: str | None = Field(
+        default=None,
+        description="Fingerprint of the request the result belongs to; a hit with a different one is a key reuse",
+    )
+
     @classmethod
     def create(
         cls,
@@ -62,6 +69,7 @@ def create(
         idempotency_key: str,
         result: JsonValue,
         ttl_seconds: float,
+        fingerprint: str | None = None,
     ) -> Self:
         """Create a new record with calculated expiration."""
         now = datetime.now(UTC)
@@ -71,6 +79,7 @@ def create(
             result=result,
             created_at=now,
             expires_at=now + timedelta(seconds=ttl_seconds),
+            fingerprint=fingerprint,
         )
 
     @classmethod
@@ -79,6 +88,7 @@ def pending(
         operation: str,
         idempotency_key: str,
         lease_seconds: float,
+        fingerprint: str | None = None,
     ) -> Self:
         """Create the in-flight reservation for an action that is about to run."""
         now = datetime.now(UTC)
@@ -89,6 +99,7 @@ def pending(
             created_at=now,
             expires_at=now + timedelta(seconds=lease_seconds),
             status="pending",
+            fingerprint=fingerprint,
         )
 
     @property
diff --git a/idempotency_kit/core/services/aio/coordinator.py b/idempotency_kit/core/services/aio/coordinator.py
index f482f15..29d4ccb 100644
--- a/idempotency_kit/core/services/aio/coordinator.py
+++ b/idempotency_kit/core/services/aio/coordinator.py
@@ -16,9 +16,11 @@
     IdempotencyInProgressError,
     IdempotencyInvalidTTLError,
     IdempotencyKeyCollisionError,
+    IdempotencyKeyReuseError,
     IdempotencyRecordExpiredError,
     IdempotencyValidationError,
 )
+from idempotency_kit.core.models.entities import IdempotencyRecord
 from idempotency_kit.core.protocols.adapter import ResultAdapter
 from idempotency_kit.core.protocols.aio.repository import AsyncIdempotencyRepository
 from idempotency_kit.core.protocols.metrics import IdempotencyMetricsProtocol, NoOpIdempotencyMetrics
@@ -110,27 +112,34 @@ async def coordinate(
         action: Callable[..., Awaitable[T]],
         /,
         *args: Any,
+        idempotency_fingerprint: str | None = None,
         **kwargs: Any,
     ) -> T:
         """Coordinate an idempotent operation.
 
+        Everything after the five positional-only arguments goes to the action, except
+        ``idempotency_fingerprint``: what the caller says the request is. It is stored with
+        the record, and a record under the same key with a different fingerprint is a key
+        reuse. ``None`` on either side means no comparison, which is the key-only behaviour.
+
         With ``enabled=False`` the action is simply run: nothing is read, nothing is
         written, and no metric is recorded.
 
         Storage and decode trouble never raise: the coordinator degrades to running the
         action. What does raise is ``IdempotencyInProgressError``, when another call with
         the same key is still running its action and ``in_flight`` is ``"raise"``, or
-        ``"wait"`` and a whole lease has passed.
+        ``"wait"`` and a whole lease has passed; and ``IdempotencyKeyReuseError``, when the
+        record under the key was made for a different request.
         """
         if not self._enabled or not idempotency_key:
             return await action(*args, **kwargs)
 
         if self._in_flight == "run":
             return await self._coordinate_unreserved(
-                operation, idempotency_key, ttl_seconds, adapter, action, *args, **kwargs
+                operation, idempotency_key, ttl_seconds, adapter, action, idempotency_fingerprint, *args, **kwargs
             )
 
-        claim = await self._claim(operation, idempotency_key, adapter)
+        claim = await self._claim(operation, idempotency_key, adapter, idempotency_fingerprint)
         if isinstance(claim, _Hit):
             return claim.value
 
@@ -144,7 +153,9 @@ async def coordinate(
             raise
 
         ttl_minutes = self._resolve_ttl_minutes(operation, ttl_seconds)
-        completed = await self._try_complete(operation, idempotency_key, result, adapter, ttl_minutes)
+        completed = await self._try_complete(
+            operation, idempotency_key, result, adapter, ttl_minutes, idempotency_fingerprint
+        )
         if claim and not completed:
             # Our reservation with no result behind it would make every retry within the
             # lease wait for, or be refused over, a record that is never coming.
@@ -158,12 +169,13 @@ async def _coordinate_unreserved(
         ttl_seconds: int | None,
         adapter: ResultAdapter[T],
         action: Callable[..., Awaitable[T]],
+        fingerprint: str | None,
         *args: Any,
         **kwargs: Any,
     ) -> T:
         """The flow without a reservation: read, run, ``SET NX``, and adopt the winner's result on a collision."""
         # 1. Try to get from storage
-        hit = await self._try_get_cached(operation, idempotency_key, adapter)
+        hit = await self._try_get_cached(operation, idempotency_key, adapter, fingerprint)
         if isinstance(hit, _Hit):
             return hit.value
 
@@ -172,26 +184,28 @@ async def _coordinate_unreserved(
 
         # 3. Cache the result
         ttl_minutes = self._resolve_ttl_minutes(operation, ttl_seconds)
-        return await self._try_save_result(operation, idempotency_key, result, adapter, ttl_minutes)
+        return await self._try_save_result(operation, idempotency_key, result, adapter, ttl_minutes, fingerprint)
 
     async def _claim(
         self,
         operation: str,
         idempotency_key: str,
         adapter: ResultAdapter[T],
+        fingerprint: str | None,
     ) -> _Hit[T] | bool:
         """Reserve the key, or replay the record that holds it.
 
         Returns the hit when a completed record is there, ``True`` when the reservation is
         ours, and ``False`` when the key holds nothing usable and the action has to run
         unreserved. Raises ``IdempotencyInProgressError`` for a key another caller holds,
-        at once in ``"raise"`` mode and after a whole lease of waiting in ``"wait"`` mode.
+        at once in ``"raise"`` mode and after a whole lease of waiting in ``"wait"`` mode,
+        and ``IdempotencyKeyReuseError`` for a key that holds a different request.
         """
         deadline = time.monotonic() + self._in_flight_lease_seconds
         read = False
         waiting = False
         while True:
-            reserved = await self._try_reserve(operation, idempotency_key)
+            reserved = await self._try_reserve(operation, idempotency_key, fingerprint)
             if reserved is None:
                 return False
             if reserved:
@@ -200,7 +214,7 @@ async def _claim(
                     self._metrics.record_miss(operation)
                 return True
 
-            found = await self._try_get_cached(operation, idempotency_key, adapter)
+            found = await self._try_get_cached(operation, idempotency_key, adapter, fingerprint)
             read = True
             if isinstance(found, _Hit):
                 return found
@@ -239,7 +253,7 @@ def _resolve_ttl_minutes(self, operation: str, ttl_seconds: int | None) -> int |
             return None
         return max(1, effective_ttl_seconds // 60)
 
-    async def _try_reserve(self, operation: str, idempotency_key: str) -> bool | None:
+    async def _try_reserve(self, operation: str, idempotency_key: str, fingerprint: str | None) -> bool | None:
         """Write the pending record under ``SET NX``.
 
         ``True`` when the reservation is ours, ``False`` when the key is already taken,
@@ -251,6 +265,7 @@ async def _try_reserve(self, operation: str, idempotency_key: str) -> bool | Non
                 operation,
                 idempotency_key,
                 lease_seconds=self._in_flight_lease_seconds,
+                fingerprint=fingerprint,
             )
             await self._repo.save(pending)
         except IdempotencyKeyCollisionError:
@@ -290,11 +305,23 @@ async def _try_get_cached(
         operation: str,
         idempotency_key: str,
         adapter: ResultAdapter[T],
+        fingerprint: str | None,
     ) -> _Hit[T] | _Lookup:
-        """Fetch and decode the record under the key; a read that fails is ``UNUSABLE``, never an exception."""
+        """Fetch and decode the record under the key.
+
+        A read that fails is ``UNUSABLE``, never an exception; the one exception that does
+        come through is ``IdempotencyKeyReuseError``, which is about the request, not storage.
+        """
         start_time = time.perf_counter()
         try:
-            found = await self._get_and_decode(operation, idempotency_key, adapter)
+            found = await self._get_and_decode(operation, idempotency_key, adapter, fingerprint)
+        except IdempotencyKeyReuseError:
+            self._metrics.record_error(operation, "key_reuse")
+            logger.warning(
+                "Idempotency key reused for a different request",
+                extra={"operation": operation, "idempotency_key": idempotency_key},
+            )
+            raise
         except Exception:
             self._metrics.record_error(operation, "storage_get_error")
             logger.exception(
@@ -322,17 +349,20 @@ async def _try_save_result(
         result: T,
         adapter: ResultAdapter[T],
         ttl_minutes: int | None,
+        fingerprint: str | None,
     ) -> T:
         """Try to save result to storage. Handles collisions and errors gracefully."""
         start_time = time.perf_counter()
         try:
-            await self._save_to_repo(operation, idempotency_key, result, adapter, ttl_minutes, replace=False)
+            await self._save_to_repo(
+                operation, idempotency_key, result, adapter, ttl_minutes, fingerprint, replace=False
+            )
             logger.info(
                 "Idempotency result saved",
                 extra={"operation": operation, "idempotency_key": idempotency_key},
             )
         except IdempotencyKeyCollisionError:
-            return await self._handle_collision(operation, idempotency_key, result, adapter)
+            return await self._handle_collision(operation, idempotency_key, result, adapter, fingerprint)
         except (IdempotencyValidationError, IdempotencyInvalidTTLError):
             self._report_unstorable_record(operation, idempotency_key, adapter)
         except Exception:
@@ -349,11 +379,14 @@ async def _try_complete(
         result: T,
         adapter: ResultAdapter[T],
         ttl_minutes: int | None,
+        fingerprint: str | None,
     ) -> bool:
         """Write the result over the reservation; ``False`` when it could not be stored, never an exception."""
         start_time = time.perf_counter()
         try:
-            await self._save_to_repo(operation, idempotency_key, result, adapter, ttl_minutes, replace=True)
+            await self._save_to_repo(
+                operation, idempotency_key, result, adapter, ttl_minutes, fingerprint, replace=True
+            )
         except (IdempotencyValidationError, IdempotencyInvalidTTLError):
             self._report_unstorable_record(operation, idempotency_key, adapter)
             return False
@@ -398,8 +431,9 @@ async def _get_and_decode(
         operation: str,
         idempotency_key: str,
         adapter: ResultAdapter[T],
+        fingerprint: str | None,
     ) -> _Hit[T] | _Lookup:
-        """Fetch record from repository and decode it safely."""
+        """Fetch record from repository and decode it safely; raises ``IdempotencyKeyReuseError`` on a mismatch."""
         cached = await self._repo.get(operation, idempotency_key)
         if cached is None:
             return _Lookup.ABSENT
@@ -414,10 +448,20 @@ async def _get_and_decode(
                 extra={"operation": operation, "idempotency_key": idempotency_key},
             )
             return _Lookup.ABSENT
+        # Before the pending check on purpose: a waiter with a different payload is refused
+        # now rather than handed someone else's result later.
+        self._check_fingerprint(cached, fingerprint)
         if cached.is_pending:
             return _Lookup.IN_FLIGHT
         return self._decode_safely(adapter, cached.result, operation, idempotency_key)
 
+    @staticmethod
+    def _check_fingerprint(record: IdempotencyRecord, fingerprint: str | None) -> None:
+        """Raise ``IdempotencyKeyReuseError`` when both sides have a fingerprint and they differ."""
+        if fingerprint is None or record.fingerprint is None or record.fingerprint == fingerprint:
+            return
+        raise IdempotencyKeyReuseError(record.operation, record.idempotency_key, record.fingerprint, fingerprint)
+
     async def _save_to_repo(
         self,
         operation: str,
@@ -425,6 +469,7 @@ async def _save_to_repo(
         result: T,
         adapter: ResultAdapter[T],
         ttl_minutes: int | None,
+        fingerprint: str | None,
         *,
         replace: bool,
     ) -> None:
@@ -434,6 +479,7 @@ async def _save_to_repo(
             idempotency_key=idempotency_key,
             result=adapter.encode(result),
             ttl_minutes=ttl_minutes,
+            fingerprint=fingerprint,
         )
         if replace:
             await self._repo.replace(record)
@@ -446,6 +492,7 @@ async def _handle_collision(
         idempotency_key: str,
         current_result: T,
         adapter: ResultAdapter[T],
+        fingerprint: str | None,
     ) -> T:
         """Handle key collision by trying to fetch the winner's result."""
         self._metrics.record_collision(operation)
@@ -454,9 +501,17 @@ async def _handle_collision(
             extra={"operation": operation, "idempotency_key": idempotency_key},
         )
         try:
-            winner = await self._get_and_decode(operation, idempotency_key, adapter)
+            winner = await self._get_and_decode(operation, idempotency_key, adapter, fingerprint)
             if isinstance(winner, _Hit):
                 return winner.value
+        except IdempotencyKeyReuseError:
+            # The action has already run, so raising would make the caller retry a
+            # completed operation. The caller keeps its own result; the reuse is on record.
+            self._metrics.record_error(operation, "key_reuse")
+            logger.warning(
+                "Idempotency key reused for a different request by a concurrent caller; keeping our own result",
+                extra={"operation": operation, "idempotency_key": idempotency_key},
+            )
         except Exception:
             logger.exception(
                 "Failed to fetch concurrent result after collision",
diff --git a/idempotency_kit/core/services/domain.py b/idempotency_kit/core/services/domain.py
index c44ca38..95ede18 100644
--- a/idempotency_kit/core/services/domain.py
+++ b/idempotency_kit/core/services/domain.py
@@ -55,6 +55,7 @@ def create_record(
         result: JsonValue,
         *,
         ttl_minutes: int | None = None,
+        fingerprint: str | None = None,
     ) -> IdempotencyRecord:
         """Create a new idempotency record.
 
@@ -63,6 +64,7 @@ def create_record(
             idempotency_key: Unique key for this operation
             result: Operation result to cache
             ttl_minutes: Custom TTL in minutes (uses default if None)
+            fingerprint: Fingerprint of the request the result belongs to, if the caller has one
 
         Returns:
             IdempotencyRecord ready to be saved
@@ -84,6 +86,7 @@ def create_record(
                 idempotency_key=idempotency_key,
                 result=result,
                 ttl_seconds=ttl_seconds,
+                fingerprint=fingerprint,
             )
         except ValidationError as e:
             # Re-map Pydantic validation error to domain validation error with detailed errors
@@ -95,6 +98,7 @@ def create_pending_record(
         idempotency_key: str,
         *,
         lease_seconds: int,
+        fingerprint: str | None = None,
     ) -> IdempotencyRecord:
         """Create the in-flight reservation for an operation whose action is about to run.
 
@@ -105,6 +109,7 @@ def create_pending_record(
             operation: Operation name (e.g., 'user.create')
             idempotency_key: Unique key for this operation
             lease_seconds: How long the reservation is held before it counts as abandoned
+            fingerprint: Fingerprint of the request being run, if the caller has one
 
         Returns:
             A pending IdempotencyRecord ready to be saved
@@ -120,6 +125,7 @@ def create_pending_record(
                 operation=operation,
                 idempotency_key=idempotency_key,
                 lease_seconds=lease_seconds,
+                fingerprint=fingerprint,
             )
         except ValidationError as e:
             raise IdempotencyValidationError(str(e), errors=e.errors()) from e
diff --git a/tests/integration/test_redis_integration.py b/tests/integration/test_redis_integration.py
index b2edb3f..fce57fc 100644
--- a/tests/integration/test_redis_integration.py
+++ b/tests/integration/test_redis_integration.py
@@ -11,8 +11,10 @@
     AsyncIdempotencyCoordinator,
     IdempotencyDomainService,
     IdempotencyKeyCollisionError,
+    IdempotencyKeyReuseError,
     IdempotencyRecord,
     JsonResultAdapter,
+    fingerprint_of,
 )
 from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository
 
@@ -168,3 +170,38 @@ async def charge_card(amount: int) -> dict[str, int | str]:
     # Assert
     assert charges == ["ch_1"]
     assert results == [{"charge_id": "ch_1", "amount": 1999}, {"charge_id": "ch_1", "amount": 1999}]
+
+
+@pytest.mark.asyncio
+async def test__coordinator__reused_key_with_a_different_fingerprint_on_real_redis__raises_key_reuse(
+    redis_client: AsyncRedisClient,
+) -> None:
+    """The key that charged 1999 comes back for 5: the caller is told, rather than handed the first charge."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(RedisAsyncIdempotencyRepository(redis_client), IdempotencyDomainService())
+    key = str(uuid4())
+
+    async def charge_card(amount: int) -> dict[str, int]:
+        return {"amount": amount}
+
+    await coordinator.coordinate(
+        "payment.charge",
+        key,
+        3600,
+        JsonResultAdapter(),
+        charge_card,
+        1999,
+        idempotency_fingerprint=fingerprint_of(amount=1999),
+    )
+
+    # Act & Assert
+    with pytest.raises(IdempotencyKeyReuseError):
+        await coordinator.coordinate(
+            "payment.charge",
+            key,
+            3600,
+            JsonResultAdapter(),
+            charge_card,
+            5,
+            idempotency_fingerprint=fingerprint_of(amount=5),
+        )
diff --git a/tests/unit/core/conftest.py b/tests/unit/core/conftest.py
index 233ec1d..d8a8153 100644
--- a/tests/unit/core/conftest.py
+++ b/tests/unit/core/conftest.py
@@ -22,12 +22,15 @@ def mock_repo() -> AsyncMock:
 def mock_domain_service() -> MagicMock:
     """Create mock domain service."""
     service = MagicMock()
-    service.create_record.side_effect = lambda operation, idempotency_key, result, ttl_minutes: IdempotencyRecord(
-        operation=operation,
-        idempotency_key=idempotency_key,
-        result=result,
-        created_at=datetime.now(UTC),
-        expires_at=datetime.now(UTC),
+    service.create_record.side_effect = lambda operation, idempotency_key, result, ttl_minutes, fingerprint=None: (
+        IdempotencyRecord(
+            operation=operation,
+            idempotency_key=idempotency_key,
+            result=result,
+            created_at=datetime.now(UTC),
+            expires_at=datetime.now(UTC),
+            fingerprint=fingerprint,
+        )
     )
     return service
 
diff --git a/tests/unit/core/test_entities.py b/tests/unit/core/test_entities.py
index 586dec3..1c63e79 100644
--- a/tests/unit/core/test_entities.py
+++ b/tests/unit/core/test_entities.py
@@ -1,6 +1,6 @@
 """Unit tests for idempotency entities."""
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from datetime import UTC, datetime, timedelta
 
 import pytest
@@ -151,3 +151,38 @@ def test__idempotency_record_pending__lease__creates_an_in_flight_reservation()
     assert not record.is_expired
     assert 29 < record.ttl_seconds <= 30
     assert record.model_dump(mode="json")["status"] == "pending"
+
+
+def test__idempotency_record__json_written_before_the_fingerprint_existed__decodes_with_none() -> None:
+    """Records already in Redis carry no ``fingerprint``; they read back as records that never raise for one."""
+    # Arrange
+    stored = (
+        '{"operation": "op", "idempotency_key": "key", "result": {"id": 1}, '
+        '"created_at": "2026-09-06T00:00:00Z", "expires_at": "2126-09-06T00:00:00Z", "status": "completed"}'
+    )
+
+    # Act
+    record = IdempotencyRecord.model_validate_json(stored)
+
+    # Assert
+    assert record.fingerprint is None
+    assert record.result == {"id": 1}
+
+
+@pytest.mark.parametrize(
+    "factory",
+    [
+        lambda: IdempotencyRecord.create("op", "key", {"id": 1}, ttl_seconds=60, fingerprint="abc"),
+        lambda: IdempotencyRecord.pending("op", "key", lease_seconds=30, fingerprint="abc"),
+    ],
+    ids=["create", "pending"],
+)
+def test__idempotency_record__fingerprint__is_carried_by_both_factories(
+    factory: Callable[[], IdempotencyRecord],
+) -> None:
+    # Act
+    record = factory()
+
+    # Assert
+    assert record.fingerprint == "abc"
+    assert record.model_dump(mode="json")["fingerprint"] == "abc"
diff --git a/tests/unit/core/test_fingerprint.py b/tests/unit/core/test_fingerprint.py
new file mode 100644
index 0000000..87fc18d
--- /dev/null
+++ b/tests/unit/core/test_fingerprint.py
@@ -0,0 +1,325 @@
+"""The request fingerprint: a reused key with a different payload is a caller error, not a replay.
+
+Regression cover for issue #27: after a charge of 1999 was recorded under a key, the same key came
+back with amount 5 and got the first charge back, with no signal that anything was off.
+"""
+
+import asyncio
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import UUID
+
+import orjson
+import pytest
+from fakeredis import FakeAsyncRedis as AsyncRedisClient
+from pydantic import BaseModel
+
+from idempotency_kit import (
+    AsyncIdempotencyCoordinator,
+    IdempotencyDomainService,
+    IdempotencyKeyCollisionError,
+    IdempotencyKeyReuseError,
+    IdempotencyMetricsProtocol,
+    IdempotencyRecord,
+    JsonResultAdapter,
+    fingerprint_of,
+)
+from idempotency_kit.infra.storage.redis.aio import RedisAsyncIdempotencyRepository
+
+
+class _Charge(BaseModel):
+    amount: int
+    currency: str = "EUR"
+
+
+async def _charge_card(amount: int) -> dict[str, int]:
+    return {"amount": amount}
+
+
+@pytest.mark.asyncio
+async def test__coordinator__reused_key_with_a_different_fingerprint__raises_key_reuse_without_running_the_action(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """The reporter's scenario: 1999 was charged under the key, and the key comes back for 5."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    action = AsyncMock(side_effect=_charge_card)
+    await coordinator.coordinate(
+        "payment.charge",
+        "order-42",
+        3600,
+        JsonResultAdapter(),
+        action,
+        1999,
+        idempotency_fingerprint=fingerprint_of(amount=1999),
+    )
+
+    # Act
+    with pytest.raises(IdempotencyKeyReuseError) as exc_info:
+        await coordinator.coordinate(
+            "payment.charge",
+            "order-42",
+            3600,
+            JsonResultAdapter(),
+            action,
+            5,
+            idempotency_fingerprint=fingerprint_of(amount=5),
+        )
+
+    # Assert
+    assert action.await_count == 1
+    assert (exc_info.value.operation, exc_info.value.key) == ("payment.charge", "order-42")
+    assert exc_info.value.stored_fingerprint == fingerprint_of(amount=1999)
+    assert exc_info.value.fingerprint == fingerprint_of(amount=5)
+    stored = await redis_repository.get("payment.charge", "order-42")
+    assert stored is not None
+    assert stored.result == {"amount": 1999}
+
+
+@pytest.mark.asyncio
+async def test__coordinator__reused_key_with_the_same_fingerprint__replays(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    action = AsyncMock(side_effect=_charge_card)
+    fingerprint = fingerprint_of(amount=1999)
+
+    # Act
+    first = await coordinator.coordinate(
+        "payment.charge", "order-42", 3600, JsonResultAdapter(), action, 1999, idempotency_fingerprint=fingerprint
+    )
+    second = await coordinator.coordinate(
+        "payment.charge", "order-42", 3600, JsonResultAdapter(), action, 1999, idempotency_fingerprint=fingerprint
+    )
+
+    # Assert
+    assert first == second == {"amount": 1999}
+    assert action.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test__coordinator__no_fingerprint__replays_on_the_key_alone_as_before(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """Nothing changes for a caller who does not opt in: the key is the whole identity."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    action = AsyncMock(side_effect=_charge_card)
+
+    # Act
+    await coordinator.coordinate("payment.charge", "order-42", 3600, JsonResultAdapter(), action, 1999)
+    replay = await coordinator.coordinate("payment.charge", "order-42", 3600, JsonResultAdapter(), action, 5)
+
+    # Assert
+    assert replay == {"amount": 1999}
+    assert action.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test__coordinator__record_without_a_fingerprint__replays_for_a_caller_with_one(
+    redis_repository: RedisAsyncIdempotencyRepository, fake_redis: AsyncRedisClient
+) -> None:
+    """A record from before fingerprints existed, or from a caller without one, never raises."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    now = datetime.now(UTC)
+    legacy = {
+        "operation": "payment.charge",
+        "idempotency_key": "order-42",
+        "result": {"amount": 1999},
+        "created_at": now.isoformat(),
+        "expires_at": (now + timedelta(hours=1)).isoformat(),
+    }
+    await fake_redis.set("probe:payment.charge:order-42", orjson.dumps(legacy))
+    action = AsyncMock(side_effect=_charge_card)
+
+    # Act
+    replay = await coordinator.coordinate(
+        "payment.charge",
+        "order-42",
+        3600,
+        JsonResultAdapter(),
+        action,
+        5,
+        idempotency_fingerprint=fingerprint_of(amount=5),
+    )
+
+    # Assert
+    assert replay == {"amount": 1999}
+    action.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test__coordinator__caller_without_a_fingerprint__replays_a_record_that_has_one(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    action = AsyncMock(side_effect=_charge_card)
+    await coordinator.coordinate(
+        "payment.charge",
+        "order-42",
+        3600,
+        JsonResultAdapter(),
+        action,
+        1999,
+        idempotency_fingerprint=fingerprint_of(amount=1999),
+    )
+
+    # Act
+    replay = await coordinator.coordinate("payment.charge", "order-42", 3600, JsonResultAdapter(), action, 5)
+
+    # Assert
+    assert replay == {"amount": 1999}
+    assert action.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test__coordinator__pending_record_with_a_different_fingerprint__refuses_the_waiter_at_once(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """A second caller with another payload is told now, not handed someone else's result after waiting."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(repository=redis_repository, domain_service=IdempotencyDomainService())
+    calls = 0
+
+    async def slow_charge(amount: int) -> dict[str, int]:
+        nonlocal calls
+        calls += 1
+        await asyncio.sleep(0.2)
+        return {"amount": amount}
+
+    # Act
+    first = asyncio.create_task(
+        coordinator.coordinate(
+            "payment.charge",
+            "order-42",
+            3600,
+            JsonResultAdapter(),
+            slow_charge,
+            1999,
+            idempotency_fingerprint=fingerprint_of(amount=1999),
+        )
+    )
+    await asyncio.sleep(0.05)
+    with pytest.raises(IdempotencyKeyReuseError):
+        await coordinator.coordinate(
+            "payment.charge",
+            "order-42",
+            3600,
+            JsonResultAdapter(),
+            slow_charge,
+            5,
+            idempotency_fingerprint=fingerprint_of(amount=5),
+        )
+
+    # Assert
+    assert await first == {"amount": 1999}
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test__coordinator__key_reuse__is_counted_as_an_error_and_the_key_is_left_alone(
+    redis_repository: RedisAsyncIdempotencyRepository,
+) -> None:
+    """A client reusing keys is worth an alert; the record it collided with is untouched."""
+    # Arrange
+    metrics = MagicMock(spec=IdempotencyMetricsProtocol)
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=redis_repository, domain_service=IdempotencyDomainService(), metrics=metrics
+    )
+    await coordinator.coordinate(
+        "op", "key", 3600, JsonResultAdapter(), _charge_card, 1999, idempotency_fingerprint=fingerprint_of(amount=1999)
+    )
+
+    # Act
+    with pytest.raises(IdempotencyKeyReuseError):
+        await coordinator.coordinate(
+            "op", "key", 3600, JsonResultAdapter(), _charge_card, 5, idempotency_fingerprint=fingerprint_of(amount=5)
+        )
+
+    # Assert
+    metrics.record_error.assert_called_once_with("op", "key_reuse")
+    metrics.record_hit.assert_not_called()
+    stored = await redis_repository.get("op", "key")
+    assert stored is not None
+    assert stored.fingerprint == fingerprint_of(amount=1999)
+
+
+@pytest.mark.asyncio
+async def test__coordinator__in_flight_run__record_with_a_different_fingerprint__raises_before_the_action(
+    mock_repo: AsyncMock, mock_adapter: MagicMock
+) -> None:
+    """The check is on the read, so the unreserved flow refuses the call before any side effect."""
+    # Arrange
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=mock_repo, domain_service=IdempotencyDomainService(), in_flight="run"
+    )
+    mock_repo.get.return_value = IdempotencyRecord.create("op", "key", {"amount": 1999}, 600, fingerprint="a")
+    action = AsyncMock()
+
+    # Act & Assert
+    with pytest.raises(IdempotencyKeyReuseError):
+        await coordinator.coordinate("op", "key", 60, mock_adapter, action, idempotency_fingerprint="b")
+    action.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test__coordinator__in_flight_run__collision_with_a_different_fingerprint__keeps_own_result(
+    mock_repo: AsyncMock, mock_adapter: MagicMock
+) -> None:
+    """After the action has run, a mismatch is logged and counted; raising would make the caller retry a done deed."""
+    # Arrange
+    metrics = MagicMock(spec=IdempotencyMetricsProtocol)
+    coordinator = AsyncIdempotencyCoordinator(
+        repository=mock_repo, domain_service=IdempotencyDomainService(), metrics=metrics, in_flight="run"
+    )
+    winner = IdempotencyRecord.create("op", "key", {"amount": 1999}, 600, fingerprint="a")
+    mock_repo.get.side_effect = [None, winner]
+    mock_repo.save.side_effect = IdempotencyKeyCollisionError("op", "key")
+    action = AsyncMock(return_value={"amount": 5})
+
+    # Act
+    result = await coordinator.coordinate("op", "key", 60, mock_adapter, action, idempotency_fingerprint="b")
+
+    # Assert
+    assert result == {"amount": 5}
+    metrics.record_error.assert_called_once_with("op", "key_reuse")
+
+
+def test__fingerprint_of__equal_values_in_another_order__agree() -> None:
+    # Act & Assert
+    assert fingerprint_of(a={"x": 1, "y": 2}, b=[1, 2]) == fingerprint_of(b=[1, 2], a={"y": 2, "x": 1})
+
+
+def test__fingerprint_of__different_values__differ() -> None:
+    # Act & Assert
+    assert fingerprint_of(amount=1999) != fingerprint_of(amount=5)
+    assert fingerprint_of(amount=5) != fingerprint_of(amount="5")
+
+
+@pytest.mark.parametrize(
+    ("value", "same_as"),
+    [
+        (_Charge(amount=5), {"amount": 5, "currency": "EUR"}),
+        (UUID("12345678-1234-5678-1234-567812345678"), "12345678-1234-5678-1234-567812345678"),
+        (Decimal("19.99"), "19.99"),
+        (datetime(2026, 9, 7, tzinfo=UTC), "2026-09-07T00:00:00Z"),
+    ],
+    ids=["pydantic-model", "uuid", "decimal", "datetime"],
+)
+def test__fingerprint_of__non_json_values__hash_as_their_json_form(value: Any, same_as: Any) -> None:
+    # Act & Assert
+    assert fingerprint_of(v=value) == fingerprint_of(v=same_as)
+
+
+def test__fingerprint_of__is_a_sha256_hex_digest() -> None:
+    # Act
+    fingerprint = fingerprint_of(amount=1999)
+
+    # Assert
+    assert len(fingerprint) == 64
+    assert int(fingerprint, 16) >= 0
diff --git a/tests/unit/core/test_idempotent.py b/tests/unit/core/test_idempotent.py
index cb64813..240d7a1 100644
--- a/tests/unit/core/test_idempotent.py
+++ b/tests/unit/core/test_idempotent.py
@@ -5,9 +5,11 @@
 from unittest.mock import ANY, MagicMock, patch
 
 import pytest
+from pydantic import BaseModel
 
 from idempotency_kit.core.decorators.aio.idempotent import async_idempotent
 from idempotency_kit.core.exceptions import IdempotencyInProgressError
+from idempotency_kit.core.fingerprint import fingerprint_of
 from idempotency_kit.core.services.aio.coordinator import AsyncIdempotencyCoordinator
 
 
@@ -263,3 +265,86 @@ async def my_func(*, idempotency_key: str | None = None, coord: AsyncIdempotency
     # Act & Assert
     with pytest.raises(IdempotencyInProgressError):
         await my_func(idempotency_key="test-key", coord=mock_coordinator)
+
+
+class _ChargeDTO(BaseModel):
+    amount: int
+    currency: str = "EUR"
+
+
+@pytest.mark.asyncio
+async def test__decorator__fingerprint_params__hands_the_coordinator_the_fingerprint_of_the_named_arguments(
+    mock_coordinator: MagicMock, mock_adapter: MagicMock
+) -> None:
+    """The named parameters, bound to the call, are what the request is."""
+    # Arrange
+    mock_coordinator.coordinate.return_value = "ok"
+    dto = _ChargeDTO(amount=1999)
+
+    @async_idempotent(operation="payment.charge", adapter=mock_adapter, fingerprint_params=("dto", "note"))
+    async def charge(
+        dto: _ChargeDTO, note: str = "", *, idempotency_key: str | None, coord: AsyncIdempotencyCoordinator
+    ) -> str:
+        return "not used"
+
+    # Act
+    result = await charge(dto, idempotency_key="order-42", coord=mock_coordinator)
+
+    # Assert
+    assert result == "ok"
+    mock_coordinator.coordinate.assert_called_once_with(
+        "payment.charge",
+        "order-42",
+        None,
+        mock_adapter,
+        ANY,
+        dto,
+        idempotency_fingerprint=fingerprint_of(dto=dto, note=""),
+        idempotency_key="order-42",
+        coord=mock_coordinator,
+    )
+
+
+@pytest.mark.asyncio
+async def test__decorator__fingerprint_params__an_argument_at_its_default_and_one_left_out_agree(
+    mock_coordinator: MagicMock, mock_adapter: MagicMock
+) -> None:
+    """Two identical requests must not disagree over how the call was spelled."""
+    # Arrange
+    mock_coordinator.coordinate.return_value = "ok"
+
+    @async_idempotent(operation="payment.charge", adapter=mock_adapter, fingerprint_params=("amount", "currency"))
+    async def charge(amount: int, currency: str = "EUR", *, idempotency_key: str | None, coord: Any) -> str:
+        return "not used"
+
+    # Act
+    await charge(5, idempotency_key="k", coord=mock_coordinator)
+    await charge(amount=5, currency="EUR", idempotency_key="k", coord=mock_coordinator)
+
+    # Assert
+    first, second = mock_coordinator.coordinate.call_args_list
+    assert first.kwargs["idempotency_fingerprint"] == second.kwargs["idempotency_fingerprint"]
+
+
+def test__decorator__fingerprint_params__unknown_parameter__raises_at_decoration(mock_adapter: MagicMock) -> None:
+    """A programming error fails at import, not on the first request."""
+    # Act & Assert
+    with pytest.raises(TypeError, match=r"does not have: \['amout'\]"):
+
+        @async_idempotent(operation="payment.charge", adapter=mock_adapter, fingerprint_params=("amout",))
+        async def charge(amount: int, *, idempotency_key: str | None = None) -> str:
+            return "not used"
+
+
+def test__decorator__fingerprint_params__uninspectable_signature__raises_at_decoration(
+    mock_adapter: MagicMock,
+) -> None:
+    # Act & Assert
+    with (
+        patch("inspect.signature", side_effect=ValueError("no signature found")),
+        pytest.raises(TypeError, match="needs a signature"),
+    ):
+
+        @async_idempotent(operation="payment.charge", adapter=mock_adapter, fingerprint_params=("amount",))
+        async def charge(**kwargs: Any) -> str:
+            return "not used"
diff --git a/tests/unit/core/test_services.py b/tests/unit/core/test_services.py
index 9c31f1c..f3fdca2 100644
--- a/tests/unit/core/test_services.py
+++ b/tests/unit/core/test_services.py
@@ -321,3 +321,17 @@ def test__coordinator_and_settings__agree_on_the_in_flight_defaults() -> None:
         DEFAULT_IN_FLIGHT_LEASE_SECONDS,
     )
     assert (DEFAULT_IN_FLIGHT_MODE, DEFAULT_IN_FLIGHT_LEASE_SECONDS) == ("wait", 30)
+
+
+def test__domain_service__fingerprint__is_stored_on_the_record_and_the_reservation() -> None:
+    """Both factories carry what the caller said the request was."""
+    # Arrange
+    service = IdempotencyDomainService()
+
+    # Act
+    record = service.create_record("op", "key", {"id": 1}, fingerprint="abc")
+    pending = service.create_pending_record("op", "key", lease_seconds=30, fingerprint="abc")
+
+    # Assert
+    assert (record.fingerprint, pending.fingerprint) == ("abc", "abc")
+    assert service.create_record("op", "key", {"id": 1}).fingerprint is None