"""The run engine's concurrency guard, measured: two ticks at the same moment act ONCE on a step, because each tick
claims (run_id, step_no, attempt, minute) with a plain INSERT before it acts and the loser skips."""
import threading
import time
from erp_platform import work
from brick_hr.application import engine as run_engine
from brick_hr.domain.run import Outcome
from brick_hr.infrastructure import capability_acl as acl
from conftest import headers
from seed import hire_body, seed
HR = headers("hr_admin")
def test_two_ticks_at_once_call_the_capability_once_per_step(engine, api, drafter, monkeypatch):
work.set_clock(at=work.parse_instant("2026-10-19T10:00 Europe/London"), freeze=True)
seed(lambda p, b: api("POST", p, HR, b))
s, h = api("POST", "/hires", HR, hire_body("Tia", "Ng", "2026-11-02"))
api("POST", f"/runs/{h['run_id']}/steps/3/complete", HR, {"attachment": {"filename": "c.pdf",
"content_base64": "JVBERg=="}})
api("POST", f"/people/{h['person_id']}/compliance", HR, {"kind": "right_to_work_check", "checked_on": "2026-10-19",
"checked_by": "hr", "check_type": "manual_document_check"})
calls, lock = [], threading.Lock()
def slow_call(name, key, data, dry_run=False, actor=""):
with lock:
calls.append(key)
time.sleep(0.3) # both ticks are inside the act at the same moment
return Outcome("manual", manual_input=data, plan="by hand", manual_reason="no provider")
monkeypatch.setattr(acl, "call", slow_call)
at = work.parse_instant("2026-10-28T09:05 Europe/London")
gate = threading.Barrier(2)
def tick():
gate.wait()
run_engine.run_due_steps(at)
threads = [threading.Thread(target=tick) for _ in range(2)]
for t in threads:
t.start()
for t in threads:
t.join()
account = [k for k in calls if k.endswith("|5")]
assert account == [f"{h['run_id']}|5"], calls # one act on the account step, not two