#!/usr/bin/env python
"""Live check of the capability claim mechanics (erp_platform.capabilities, rule 1) on a real DataK3 bucket.
tests/test_capabilities.py proves the rules on SQLite. The rules rest on four things the ENGINE must do, and
this checks each on DataK3 through a throwaway provider brick whose claims tables come from
`capabilities.ddl(...)` — the real Provided.invoke, the real _Store, the real SQL:
1. the claim is a plain INSERT, and a concurrent duplicate is rejected 23505: N threads, one key → the
handler runs ONCE (plus a raw race: N concurrent INSERTs of one key, 1 acknowledged);
2. a terminal result is stored on the claim and replayed (idempotent_replay, no second run);
3. a release (DELETE … WHERE status IS NULL) drops a non-terminal claim so a retry acts — and leaves a
terminal one alone;
4. a takeover after the lease expired inserts generation n+1, and a takeover race has one winner.
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/live_capability_check.py
Every statement goes through `dodil data pg` (erp_platform.livecheck), so it needs only a CLI login. Each
_Store operation is ONE statement in its own transaction, so the CLI's one-statement sessions are faithful
here. Threads are real concurrent sessions at the engine. Harness difference, handled below: the CLI renders a
TIMESTAMP in the machine's local zone (+03:00); a driver hands back naive UTC, so it is converted back.
"""
import tempfile
import threading
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from erp_platform import capabilities, livecheck as lc
from erp_platform.bricks import Brick
from erp_platform.capabilities import Call, Transient
NAME = "capcheck"
T = capabilities.tables(NAME)
TABLES = [T.claims.name, T.log.name]
N = int(__import__("os").environ.get("THREADS", "10"))
class Conn(lc.Connection):
"""livecheck's connection, with TIMESTAMPs back as naive UTC (what psycopg returns for TIMESTAMP)."""
def execute(self, stmt):
res = super().execute(stmt)
for row in res:
for k, v in list(row.items()):
if isinstance(v, datetime) and v.tzinfo:
dict.__setitem__(row, k, v.astimezone(timezone.utc).replace(tzinfo=None))
return res
@contextmanager
def connect():
yield Conn()
PROV = Brick(NAME, instructions="A throwaway provider for the live claim check.")
PROV.capability_connection = connect
RUNS: list = []
_lock = threading.Lock()
SCRIPT = {"delay": 0.0, "fail": False}
@PROV.capability("drive.create_folder", attempts=1, lease_seconds=60)
def create_folder(call: Call):
with _lock:
RUNS.append(call.idempotency_key)
if SCRIPT["delay"]:
time.sleep(SCRIPT["delay"])
if SCRIPT["fail"]:
raise Transient("vendor 503")
return "fld-1", {"folder_id": "fld-1", "url": "https://drive.example/fld-1"}, "Folder created."
PROVIDED = create_folder # the Provided object: invoke() is what the generated route and tool call
CALLER = {"sub": "live-check", "email": "live-check@dodil.io"}
INPUT = {"name": "Starter docs", "parent": None, "share_with": []}
def claims(key):
return lc.rows(f"SELECT generation, status FROM {T.claims.name} "
f"WHERE capability = 'drive.create_folder' AND idempotency_key = '{key}' ORDER BY generation")
def race(key, n):
out, errs = [], []
def go():
try:
out.append(PROVIDED.invoke(CALLER, key, INPUT))
except Exception as e: # noqa: BLE001 — counted, reported
errs.append(f"{type(e).__name__}: {str(e)[-160:]}")
ts = [threading.Thread(target=go) for _ in range(n)]
t0 = time.monotonic()
for t in ts: t.start()
for t in ts: t.join()
return out, errs, time.monotonic() - t0
def main() -> int:
if not lc.BUCKET:
raise SystemExit("set BUCKET=erp-platform-test DODIL_ORG=dodil")
with tempfile.NamedTemporaryFile("w", suffix=".sql", delete=False) as f:
f.write(capabilities.ddl(NAME)) # exactly what a provider puts in its schema.sql
lc.create_schema(f.name, TABLES) # drops, then creates
# 0. the raw engine property: N concurrent plain INSERTs of one key
acked, rejected, other = [], [], []
def raw(i):
try:
lc.pg(f"INSERT INTO {T.claims.name} (capability, idempotency_key, generation, input_hash, attempt_id, "
f"claimed_at, lease_until) VALUES ('raw', 'k', 1, 'h', 'a{i}', TIMESTAMP '2026-10-17 00:00:00', "
f"TIMESTAMP '2026-10-17 00:00:00')", attempts=1)
acked.append(i)
except RuntimeError as e:
(rejected if "23505" in str(e) else other).append(str(e)[-160:])
ts = [threading.Thread(target=raw, args=(i,)) for i in range(N)]
for t in ts: t.start()
for t in ts: t.join()
stored = lc.rows(f"SELECT attempt_id FROM {T.claims.name} WHERE capability = 'raw'")
print(f" raw race: {N} INSERTs → {len(acked)} acknowledged, {len(rejected)} rejected 23505, "
f"{len(other)} other; {len(stored)} row(s) stored")
lc.check(f"raw: {N} concurrent INSERTs of one key → 1 acknowledged, {N - 1} rejected 23505, 1 row",
len(acked) == 1 and len(rejected) == N - 1 and len(stored) == 1 and stored[0]["attempt_id"] == f"a{acked[0]}",
{"acked": acked, "rejected": len(rejected), "other": other[:2], "stored": stored})
# 1. the claim race through invoke: N threads, one key, the handler slow enough that they overlap
SCRIPT["delay"] = 3.0
out, errs, secs = race("race-1", N)
SCRIPT["delay"] = 0.0
statuses = sorted(r.status for r in out)
print(f" invoke race: {N} threads in {secs:.1f}s → handler ran {RUNS.count('race-1')}x; "
f"answers {dict((s, statuses.count(s)) for s in set(statuses))}; errors {len(errs)}")
lc.check(f"claim race: {N} concurrent invokes of one key → the handler ran once",
RUNS.count("race-1") == 1 and not errs, {"runs": RUNS.count("race-1"), "errs": errs[:2]})
lc.check("claim race: exactly one 'done' (the winner); the rest in_progress or a replay",
statuses.count("done") >= 1 and sum(1 for r in out if r.status == "done" and not r.idempotent_replay) == 1
and all(s in ("done", "in_progress") for s in statuses), statuses)
lc.check("claim race: one claim row, generation 1, terminal", claims("race-1") == [{"generation": 1, "status": "done"}],
claims("race-1"))
# 2. a terminal result is stored and replayed
again = PROVIDED.invoke(CALLER, "race-1", INPUT)
lc.check("replay: the stored result comes back, idempotent_replay, no second run",
again.status == "done" and again.idempotent_replay and again.provider_ref == "fld-1"
and again.output == {"folder_id": "fld-1", "url": "https://drive.example/fld-1"}
and RUNS.count("race-1") == 1, again.as_dict())
other_input = lc.refused(PROVIDED.invoke, CALLER, "race-1", {**INPUT, "name": "Other"})
lc.check("same key, other input → 409", other_input and other_input[0] == 409, other_input)
# 3. release: a non-terminal outcome deletes its own claim (status IS NULL), so a retry acts
SCRIPT["fail"] = True
r1 = PROVIDED.invoke(CALLER, "rel-1", INPUT)
SCRIPT["fail"] = False
lc.check("release: a retryable failure leaves no claim", r1.status == "failed" and r1.retryable
and claims("rel-1") == [], {"result": r1.as_dict(), "claims": claims("rel-1")})
r2 = PROVIDED.invoke(CALLER, "rel-1", INPUT)
lc.check("release: the retry acts (handler ran twice) and is terminal",
r2.status == "done" and not r2.idempotent_replay and RUNS.count("rel-1") == 2
and claims("rel-1") == [{"generation": 1, "status": "done"}], claims("rel-1"))
PROVIDED.store.release("drive.create_folder", "rel-1", 1)
lc.check("release: the status IS NULL filter leaves a terminal claim alone",
claims("rel-1") == [{"generation": 1, "status": "done"}], claims("rel-1"))
# 4. takeover: a crashed caller's claim (gen 1, short lease) expires; the next call takes generation 2
store = PROVIDED.store
digest = PROVIDED.capability.input_hash(PROVIDED.capability.validate_input(INPUT))
assert store.insert_claim("drive.create_folder", "take-1", 1, digest, "crashed", lease=2.0)
busy = PROVIDED.invoke(CALLER, "take-1", INPUT)
lc.check("takeover: inside the lease the key is in_progress, nothing run",
busy.status == "in_progress" and RUNS.count("take-1") == 0, busy.as_dict())
time.sleep(3.0)
SCRIPT["delay"] = 3.0
out, errs, secs = race("take-1", N)
SCRIPT["delay"] = 0.0
rows = claims("take-1")
print(f" takeover race: {N} threads in {secs:.1f}s → handler ran {RUNS.count('take-1')}x; claims {rows}")
lc.check(f"takeover: {N} racing callers after the lease → one run, generation 2",
RUNS.count("take-1") == 1 and not errs
and rows == [{"generation": 1, "status": None}, {"generation": 2, "status": "done"}],
{"runs": RUNS.count("take-1"), "errs": errs[:2], "rows": rows})
lc.check("takeover: replay after it returns the generation-2 result",
PROVIDED.invoke(CALLER, "take-1", INPUT).idempotent_replay and RUNS.count("take-1") == 1)
log_n = lc.rows(f"SELECT COUNT(*) AS n FROM {T.log.name}")[0]["n"]
print(f" action log: {log_n} row(s), one per call")
return lc.finish(TABLES)
if __name__ == "__main__":
raise SystemExit(main())