Preserve fractional approval expiry through atomic consumption

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

tests/test_timestamp_precision.py

@@ -0,0 +1,130 @@
+"""Actual source clock, locking and original immutable receipt precision."""
+from datetime import timedelta, timezone
+import sqlite3
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from approval_engine.errors import Conflict, StoreUnavailable
+from approval_engine.store import Engine, iso, utc_now
+from tests.conftest import FROZEN, binding
+
+DIGEST = "sha256:" + "ab" * 32
+
+
+def issue(engine, start, end):
+    obj = engine.create(binding(), {"not_before": start.isoformat(), "expires_at": end.isoformat()})
+    return engine.add_entry(obj.id, "human", principal_type="human")
+
+
+def uses(engine):
+    return [row for row in engine.undrained() if row["class"] == "use"]
+
+
+def test_clock_and_iso_preserve_aware_precision_and_legacy_seconds():
+    now = FROZEN.replace(microsecond=750123)
+    offset = now.astimezone(timezone(timedelta(hours=2)))
+    assert utc_now(lambda: offset) == now
+    assert iso(offset) == "2026-08-29T12:00:00.750123+00:00"
+    assert iso(FROZEN) == "2026-08-29T12:00:00+00:00"
+
+
+def test_fractional_expiry_refuses_actual_canonical_counterexample():
+    engine = Engine(clock=lambda: FROZEN)
+    obj = issue(engine, FROZEN-timedelta(seconds=1), FROZEN.replace(microsecond=500000))
+    engine.clock = lambda: FROZEN.replace(microsecond=750000)
+    assert engine.claim(obj.id)["valid_now"] is False
+    with pytest.raises(Conflict):
+        engine.consume(obj.id, DIGEST, decision_id="decision:original")
+    assert uses(engine) == []
+
+
+def test_immediate_fractional_receipt_restart_and_outbox_use_one_instant(tmp_path):
+    current = [FROZEN.replace(microsecond=750123)]
+    path = tmp_path/"approval.sqlite"
+    engine = Engine(path, clock=lambda: current[0])
+    obj = issue(engine, FROZEN.replace(microsecond=700000), FROZEN+timedelta(seconds=1))
+    samples = []
+    def clock():
+        value = current[0] + timedelta(microseconds=len(samples))
+        samples.append(value)
+        return value
+    engine.clock = clock
+    result = engine.consume(obj.id, DIGEST, decision_id="decision:original")
+    assert samples == [current[0]]
+    assert result["consumed_at"] == current[0].isoformat()
+    assert result["consumed_at"] >= FROZEN.replace(microsecond=700001).isoformat()
+    event = uses(engine)[0]
+    assert event["created_at"] == event["payload"]["observed_at"] == result["consumed_at"]
+    engine.close()
+    engine = Engine(path, clock=lambda: FROZEN+timedelta(days=1), auto_migrate=False)
+    assert engine.get(obj.id).as_dict()["consumption"]["consumed_at"] == result["consumed_at"]
+    retry = engine.consume(obj.id, DIGEST, decision_id="decision:new")
+    assert retry == {**result, "idempotent": True}
+    assert uses(engine) == [event]
+    engine.close()
+
+
+@pytest.mark.parametrize("after,accepted", [(450000, True), (750000, False)])
+def test_consume_samples_after_real_sqlite_lock_wait(tmp_path, after, accepted):
+    current = [FROZEN.replace(microsecond=250000)]
+    path = tmp_path/"approval.sqlite"
+    engine = Engine(path, clock=lambda: current[0])
+    obj = issue(engine, FROZEN-timedelta(seconds=1), FROZEN.replace(microsecond=500000))
+    entered = threading.Event()
+    initialized = threading.Event()
+    start = threading.Event()
+    samples = []
+    def consume():
+        conn = engine._conn()
+        conn.set_trace_callback(lambda statement: entered.set() if statement == "BEGIN IMMEDIATE" else None)
+        initialized.set()
+        assert start.wait(2)
+        def clock():
+            samples.append(current[0])
+            return current[0]
+        engine.clock = clock
+        try:
+            return engine.consume(obj.id, DIGEST, decision_id="decision:original")
+        finally:
+            engine.close()
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        future = pool.submit(consume)
+        assert initialized.wait(2)
+        lock = sqlite3.connect(path)
+        lock.execute("BEGIN IMMEDIATE")
+        start.set()
+        assert entered.wait(2)
+        current[0] = FROZEN.replace(microsecond=after)
+        lock.commit()
+        lock.close()
+        if accepted:
+            assert future.result(timeout=2)["consumed_at"] == current[0].isoformat()
+        else:
+            with pytest.raises(Conflict):
+                future.result(timeout=2)
+    assert samples == [current[0]]
+    assert len(uses(engine)) == int(accepted)
+    engine.close()
+
+
+def test_old_second_precision_receipt_and_payload_are_never_backfilled(tmp_path):
+    engine = Engine(tmp_path/"old.sqlite", clock=lambda: FROZEN)
+    obj = issue(engine, FROZEN-timedelta(seconds=1), FROZEN+timedelta(seconds=1))
+    original = engine.consume(obj.id, DIGEST, decision_id="decision:old")
+    event = uses(engine)[0]
+    assert original["consumed_at"] == "2026-08-29T12:00:00+00:00"
+    engine.clock = lambda: FROZEN + timedelta(days=1, microseconds=123456)
+    assert engine.consume(obj.id, DIGEST, decision_id="decision:new") == {**original, "idempotent": True}
+    assert uses(engine) == [event]
+
+
+def test_fractional_use_outbox_failure_rolls_back(engine):
+    engine.clock = lambda: FROZEN.replace(microsecond=123456)
+    obj = issue(engine, FROZEN, FROZEN+timedelta(seconds=1))
+    engine.fail_outbox = True
+    with pytest.raises(StoreUnavailable):
+        engine.consume(obj.id, DIGEST)
+    assert engine.get(obj.id).consumed_at is None
+    assert uses(engine) == []