"""idempotency — an `Idempotency-Key` on a command that is not idempotent by itself, once for every brick.
`open_quote`, `log_activity`, `add_line`: each call makes something new, so a client that sent one, lost
the reply (a timeout, a dropped connection, an agent's retry) and sent it again has made two. The CRM audit
(2026-10-17) logged the same call twice and got two activities. The fix every payment API uses: the client
names the attempt — the HTTP header `Idempotency-Key: <its own unique string>`, or the MCP argument
`idempotency_key` — and a repeat of the same key by the same caller gets the FIRST answer back instead of a
second write.
THE RULES (the same measured mechanics as erp_platform.capabilities):
* **The claim is a plain INSERT** into `erp_idempotency` on (caller, use case, key, generation). Concurrent
duplicates of one key: exactly one INSERT is acknowledged, the rest fail 23505 (DataK3 0.2.1, measured
2026-10-17) — so exactly one request runs the use case. An upsert would let both "win".
* **Scoped per caller** (the identity's `sub`) and per use case: two people cannot collide, and one key
cannot be replayed into another tool.
* **Only success is stored and replayed.** A refusal or a crash RELEASES the claim, so the same key may try
again: a refused command wrote nothing, and its next attempt may well succeed (the partner was unblocked).
* **The same key with a different request is 422**, compared by a SHA-256 of the canonical JSON of the
validated input. **While the first is still running, a repeat is 409** "still being processed".
* **A claim whose request died** (pod gone between the claim and the answer) expires after the lease, and
the next attempt takes over with generation n+1 — again a plain INSERT, so a takeover race has one winner.
The window this leaves — the use case committed, the pod died before the answer was stored — is the same
one a capability call has; it is a lease (120 s), not a lost write.
The key is optional everywhere: without one, a command behaves exactly as before.
"""
import hashlib
import json
import re
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, Optional
from sqlalchemy import Column, Integer, MetaData, String, Table, TIMESTAMP, and_, delete, insert, select
HEADER = "Idempotency-Key"
ARGUMENT = "idempotency_key"
LEASE_SECONDS = 120.0
_KEY = re.compile(r"[\x21-\x7e]{1,255}") # printable ASCII, no spaces: what a UUID or ULID is
md = MetaData()
# The NOT NULLs are platform.sql's, here too: DataK3 checks them on an upsert's VALUES even when the row conflicts
# (an answer stored without claimed_at was refused 23502 on crm-demo, 2026-10-17 — the tests' tables had no NOT NULL)
claims = Table(
"erp_idempotency", md,
Column("caller", String, primary_key=True), Column("use_case", String, primary_key=True),
Column("idempotency_key", String, primary_key=True),
Column("generation", Integer, primary_key=True, autoincrement=False),
Column("request_hash", String, nullable=False), Column("claimed_at", TIMESTAMP, nullable=False),
Column("lease_until", TIMESTAMP, nullable=False),
Column("status", Integer), Column("response", String), Column("finished_at", TIMESTAMP))
class KeyRefused(Exception):
"""What the key's own rules refuse (reused for another request, still running, malformed)."""
def __init__(self, status: int, reason: str):
super().__init__(reason)
self.status, self.reason = status, reason
def valid_key(key: str) -> str:
if not isinstance(key, str) or not _KEY.fullmatch(key):
raise KeyRefused(422, f"{HEADER} is 1–255 printable characters with no spaces (a UUID is ideal)")
return key
def request_hash(use_case: str, inputs: dict) -> str:
canonical = json.dumps({"use_case": use_case, "input": inputs}, sort_keys=True, default=str,
separators=(",", ":"))
return hashlib.sha256(canonical.encode()).hexdigest()
def _utcnow() -> datetime:
"""Real time, naive UTC — what a TIMESTAMP column holds and hands back on DataK3 and SQLite."""
return datetime.now(timezone.utc).replace(tzinfo=None)
def _naive(v: Any) -> Optional[datetime]:
if v is None:
return None
if isinstance(v, str):
v = datetime.fromisoformat(v)
return v.astimezone(timezone.utc).replace(tzinfo=None) if v.tzinfo else v
def _is_duplicate(exc: BaseException) -> bool:
from sqlalchemy.exc import IntegrityError
text = str(exc).lower()
return isinstance(exc, IntegrityError) or "23505" in text or "duplicate key" in text
def _default_connect():
from erp_platform import db # imported late: it needs BUCKET
return db.engine.begin()
class Store:
"""The claims, each operation its own short transaction (the use case runs in its own)."""
def __init__(self, connect: Optional[Callable[[], Any]] = None):
self.connect = connect or _default_connect
def rows(self, caller: str, use_case: str, key: str) -> list:
c = claims.c
with self.connect() as conn:
result = conn.execute(select(claims).where(and_(
c.caller == caller, c.use_case == use_case, c.idempotency_key == key)).order_by(c.generation))
return [dict(r._mapping) for r in result]
def claim(self, caller: str, use_case: str, key: str, gen: int, digest: str, lease: float) -> Optional[dict]:
"""The claim: a PLAIN INSERT. The claimed row if this request won it; None on a duplicate key."""
now = _utcnow()
row = {"caller": caller, "use_case": use_case, "idempotency_key": key, "generation": gen,
"request_hash": digest, "claimed_at": now, "lease_until": now + timedelta(seconds=lease)}
try:
with self.connect() as conn:
conn.execute(insert(claims).values(**row))
return row
except Exception as exc: # noqa: BLE001 — re-raised unless a duplicate
if _is_duplicate(exc):
return None
raise
def finish(self, row: dict, response: Any) -> None:
"""Store the answer on this request's claim — the WHOLE row, claimed_at and lease_until included."""
from erp_platform.repository import upsert
full = {**row, "status": 200, "response": json.dumps(response, default=str), "finished_at": _utcnow()}
with self.connect() as conn:
upsert(conn, claims, [full], ("caller", "use_case", "idempotency_key", "generation"), insert_only=())
def release(self, caller: str, use_case: str, key: str, gen: int) -> None:
c = claims.c
with self.connect() as conn:
conn.execute(delete(claims).where(and_(c.caller == caller, c.use_case == use_case,
c.idempotency_key == key, c.generation == gen,
c.status.is_(None))))
def begin(store: Store, caller: str, use_case: str, key: str, digest: str,
lease: float = LEASE_SECONDS) -> tuple:
"""Claim the key, or say what to answer instead. Returns ("replay", response) | ("run", claim_row).
Raises KeyRefused for a key reused with another request (422) or one still being processed (409)."""
for _ in range(3):
rows = store.rows(caller, use_case, key)
done = [r for r in rows if r["status"] is not None]
if done:
if done[-1]["request_hash"] != digest:
raise KeyRefused(422, f"this {HEADER} was already used for a different request — use a new "
f"key for a new request")
return "replay", json.loads(done[-1]["response"])
now = _utcnow()
live = [r for r in rows if r["status"] is None and (_naive(r["lease_until"]) or now) > now]
if live:
if live[-1]["request_hash"] != digest:
raise KeyRefused(422, f"this {HEADER} is in use for a different request — use a new key")
raise KeyRefused(409, f"a request with this {HEADER} is still being processed — retry in a moment "
f"for its answer")
gen = max((int(r["generation"]) for r in rows), default=0) + 1
won = store.claim(caller, use_case, key, gen, digest, lease)
if won:
return "run", won
raise KeyRefused(409, f"a request with this {HEADER} is still being processed — retry in a moment")
def ddl() -> str:
"""The CREATE TABLE (it is in platform.sql; tests create it from `md`)."""
from sqlalchemy.dialects import postgresql
from sqlalchemy.schema import CreateTable
return str(CreateTable(claims).compile(dialect=postgresql.dialect())).strip() + ";"