File size: 2,378 Bytes
fc3e70a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | # This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
"""Phase 4: Temporal lock — every transformation receives and preserves τ."""
from __future__ import annotations
import time
from dataclasses import dataclass, field
from typing import TypeVar, Callable
from .types import FSMState, TemporalViolation
from .ir import TemporalLock, TemporalState, TemporalTransition
T = TypeVar("T")
def stamp() -> int:
"""Monotonic integer tick based on nanoseconds."""
return time.monotonic_ns()
def apply_temporal(
fn: Callable[..., T],
t_in: TemporalState,
lock: TemporalLock,
stage: str,
*args,
**kwargs,
) -> tuple[T, TemporalState]:
"""
Run fn(*args, **kwargs) under a temporal contract.
- Records τn → τn+1 transition.
- Raises TemporalViolation if the state was locked.
- Returns (result, new_temporal_state).
"""
if t_in.locked:
raise TemporalViolation(
FSMState.TEMPORAL_LOCK,
f"Stage '{stage}' attempted to transform a locked temporal state τ{t_in.tau}",
)
result = fn(*args, **kwargs)
t_out = t_in.advance(reason=stage)
lock.record(TemporalTransition(
from_tau=t_in.tau,
to_tau=t_out.tau,
reason=stage,
stage=stage,
))
return result, t_out
def preserve_temporal(
fn: Callable[..., T],
t_in: TemporalState,
lock: TemporalLock,
stage: str,
*args,
**kwargs,
) -> tuple[T, TemporalState]:
"""
Run fn but preserve τ (no advance). Used for read-only passes.
Records a τ→τ no-op transition.
"""
result = fn(*args, **kwargs)
lock.record(TemporalTransition(
from_tau=t_in.tau,
to_tau=t_in.tau,
reason=f"preserve:{stage}",
stage=stage,
))
return result, t_in
def assert_temporal_invariant(t_before: TemporalState, t_after: TemporalState, stage: str) -> None:
"""Assert τ_after == τ_before + 1 (for transforms) or == τ_before (for reads)."""
if t_after.tau not in (t_before.tau, t_before.tau + 1):
raise TemporalViolation(
FSMState.TEMPORAL_LOCK,
f"Stage '{stage}' illegally mutated τ: {t_before.tau} → {t_after.tau}",
)
|