symfonic.capabilities.human.ledger

ledger

The authoritative token issuance/consumption ledger — operated platform only.

Four things live here that live nowhere else.

One linearization point. In operated mode the ledger both issues and consumes, so "exactly one winner" is a property of a single object rather than an agreement between two. That is why it satisfies the consumption port itself.

Authoritative in the strong sense. A jti the ledger never issued is not a token that might be fine — it is refused. Without that inversion any token minted outside it would be invisible to the count the retirement gate reads.

Bounded purposes. The drain proof and the retirement horizon are operated-only (LIB-TL-4, CUT-AIR-5). A library deployment does not get a degraded version of them; it does not get this object.

Derived durability, and derived reach. Two tables decide a redemption: the issued set and the consumed set. The ledger is exactly as durable as the weaker of them, and it reaches exactly as far. Handing in a store moves the winner decision onto a shared atomic conditional write; handing in records moves the issuance table and the retirement horizon onto the same backend. Neither is inferred from the operated role: a ledger that forgets every issuance when the process exits refuses every legitimate resume after a restart, cannot enforce one node's horizon on the next, and can only count its own worker's rows — so it says durable=False, deployment_wide=False, and declines to call a drain drained.

DrainProof dataclass

DrainProof(drained: bool, outstanding: int, reason: str, horizon: float | None = None, deadline: float | None = None, scope: str = PROCESS)

Whether every legacy-pinned token has drained, and why not if not.

IssuedToken dataclass

IssuedToken(jti: str, scope_hash: str, name: str, issued_at: float, expires_at: float, legacy_pinned: bool = False, vector_hash: str = '')

One issuance row. Ids, times, and the one bit the drain gate reads.

RetirementHorizon dataclass

RetirementHorizon(at: float, reason: str)

SCP-FRZ-2: the date, and the operator's reason for it.

TokenConsumption dataclass

TokenConsumption(jti: str, scope_hash: str, name: str, consumed_at: float)

One redemption row: who won, and when. Losers are not recorded here.

TokenIssuanceLedger

TokenIssuanceLedger(*, maximum_ttl_seconds: float, clock: Callable[[], float] = time.time, store: Any = None, records: Any = None)

Issuance, consumption, the maximum-TTL bound, and the retirement horizon.

Source code in src/symfonic/capabilities/human/ledger.py

def __init__(
    self,
    *,
    maximum_ttl_seconds: float,
    clock: Callable[[], float] = time.time,
    store: Any = None,
    records: Any = None,
) -> None:
    if maximum_ttl_seconds <= 0:
        raise TokenTTLError(
            "the maximum pause-token lifetime must be positive; it is the "
            "bound the drain deadline is computed from"
        )
    self._max_ttl = float(maximum_ttl_seconds)
    self._clock = clock
    self._lock = asyncio.Lock()
    # The library-mode conditional write, reused: a second implementation
    # here would be a second place that believes it decides the winner.
    self._writes = (
        None if store is None else ConditionalWriteConsumption(store, clock=clock)
    )
    self._records = _records_for(records)

deployment_wide property

deployment_wide: bool

Whether both tables are shared, rather than this worker's memory.

durable property

durable: bool

As durable as the weaker of the two tables a redemption reads.

drain_deadline async

drain_deadline() -> float | None

The horizon plus the maximum TTL: the last moment anything can live.

Source code in src/symfonic/capabilities/human/ledger.py

async def drain_deadline(self) -> float | None:
    """The horizon plus the maximum TTL: the last moment anything can live."""
    recorded = await self._records.horizon()
    return None if recorded is None else recorded.at + self._max_ttl

outstanding async

outstanding(now: float | None = None) -> tuple[str, ...]

Issued, unconsumed, and not yet expired.

Source code in src/symfonic/capabilities/human/ledger.py

async def outstanding(self, now: float | None = None) -> tuple[str, ...]:
    """Issued, unconsumed, and not yet expired."""
    moment = self._clock() if now is None else now
    return tuple(sorted(row.jti for row in await self._live(moment)))

record_retirement_horizon async

record_retirement_horizon(at: float, *, reason: str) -> None

SCP-FRZ-2 — the date after which nothing legacy-pinned may outlive.

Source code in src/symfonic/capabilities/human/ledger.py

async def record_retirement_horizon(self, at: float, *, reason: str) -> None:
    """SCP-FRZ-2 — the date after which nothing legacy-pinned may outlive."""
    if at <= self._clock():
        raise RetirementHorizonError(
            "a retirement horizon must be in the future; a backdated one "
            "would retire tokens that nothing had stopped supporting"
        )
    async with self._lock:
        standing, recorded = await self._records.claim_horizon(float(at), reason)
    if not recorded:
        raise RetirementHorizonError(
            f"a retirement horizon is already recorded at {standing.at} "
            f"({standing.reason!r}); the operator who declared the "
            "retirement is the one whose date and reason stand"
        )