omnibias.core.proof.lean_lock
Coefficient recurrences, activation contracts, jet algebra, and verified arithmetic shared by the numerical backends.
import omnibias.core.proof.lean_lockModule contract & mathematical conventions
Serialize generated Lean obligations across threads and processes.
Shared formulas do not imply universal floating-point identity. Verify dtype, shape, derivative-order support, and the stated numerical assumptions for the path you use.
generated_lean_obligation
omnibias.core.proof.lean_lock.generated_lean_obligationSource ↗generated_lean_obligation(root: Path, generated: Path, source: str) -> Iterator[None]Lock through write/build/restore, restoring even after a failed build.
Parameters
| Name | Type & default | Contract |
|---|---|---|
rootpositional-or-keyword | PathRequired | See the operation contract and implementation below. |
generatedpositional-or-keyword | PathRequired | See the operation contract and implementation below. |
sourcepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Iterator[None]Executed regression · zeta value identity earns theorem prover verified
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_zeta_value_identity_earns_theorem_prover_verified(self) -> None:
verdict = check_identity_certificate(zeta_negative_odd_identity(2, _ZETA_NEG[2]))
assert verdict.theorem_prover_verifiedpython -m pytest "packages/omnibias-difference/tests/test_identities.py::TestRealLeanKernelPass::test_zeta_value_identity_earns_theorem_prover_verified" -qRegression example · threads cannot overwrite another active obligation
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_threads_cannot_overwrite_another_active_obligation(tmp_path):
generated = tmp_path / "Generated.lean"
generated.write_text("original")
def worker(label):
with generated_lean_obligation(tmp_path, generated, label):
time.sleep(0.025)
assert generated.read_text() == label
with ThreadPoolExecutor(max_workers=4) as pool:
list(pool.map(worker, ["first", "second", "third", "fourth"]))
assert generated.read_text() == "original"python -m pytest packages/omnibias-core/tests/proof/test_lean_lock.py -k test_threads_cannot_overwrite_another_active_obligation -qRead the implementation
def generated_lean_obligation(root: Path, generated: Path, source: str) -> Iterator[None]:
"""Lock through write/build/restore, restoring even after a failed build."""
with lean_project_lock(root):
original = generated.read_bytes() if generated.exists() else None
try:
generated.write_text(source, encoding="utf-8")
yield
finally:
if original is None:
generated.unlink(missing_ok=True)
else:
generated.write_bytes(original)lean_project_lock
omnibias.core.proof.lean_lock.lean_project_lockSource ↗lean_project_lock(root: Path) -> Iterator[None]Lock a Lean project across cooperating threads and processes.
OS advisory locks coordinate cooperating bridge calls in separate processes.
A manual editor or direct external lake build does not take this lock.
Parameters
| Name | Type & default | Contract |
|---|---|---|
rootpositional-or-keyword | PathRequired | See the operation contract and implementation below. |
Iterator[None]Executed regression · zeta value identity earns theorem prover verified
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_zeta_value_identity_earns_theorem_prover_verified(self) -> None:
verdict = check_identity_certificate(zeta_negative_odd_identity(2, _ZETA_NEG[2]))
assert verdict.theorem_prover_verifiedpython -m pytest "packages/omnibias-difference/tests/test_identities.py::TestRealLeanKernelPass::test_zeta_value_identity_earns_theorem_prover_verified" -qRead the implementation
def lean_project_lock(root: Path) -> Iterator[None]:
"""Lock a Lean project across cooperating threads and processes.
OS advisory locks coordinate cooperating bridge calls in separate processes.
A manual editor or direct external ``lake build`` does not take this lock.
"""
key = str(root.resolve())
with _registry_guard:
mutex = _project_locks.setdefault(key, threading.Lock())
lock_directory = root / ".lake"
lock_directory.mkdir(parents=True, exist_ok=True)
lockfile = lock_directory / "omnibias-bridge.lock"
with mutex, lockfile.open("a+b") as handle:
if sys.platform == "win32": # pragma: no cover - platform specific
import msvcrt
handle.seek(0)
handle.write(b"0")
handle.flush()
while True:
try:
handle.seek(0)
msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
break
except OSError:
time.sleep(0.05)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
if sys.platform == "win32": # pragma: no cover - platform specific
handle.seek(0)
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
else:
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)