tegwick
28f0f429b2
migrations/0003_attestations.sql: one row per Phase, no UPDATE/DELETE
grant for trf_app (a published attestation is permanent).
src/target_revenue/attestation.py: publish_attestation() is idempotent
(existing row returned unchanged, never regenerated) and derives
conversion_timestamp from the ledger itself via _find_conversion_prefix(),
which finds the earliest prefix where the fold reaches Outstanding
Target = 0 rather than trusting entries[-1] or wall-clock publish time -
so a later unrelated entry can never change an already-converged Phase's
recorded conversion moment. Raises NotConvertedError rather than
fabricating a conversion. Reuses conversion.generate_attestation()
unchanged. Guarded the psycopg import behind TYPE_CHECKING so the pure
helper stays testable under plain system Python.
service/app.py adds GET /phases/{id}/attestation (unauthenticated,
publish-on-first-observation).
tests/test_attestation.py (3 tests, no Docker/Postgres) proves the
earliest-crossing behavior. 6 new Docker-gated tests in
test_ledger_hosting.py cover pre-conversion 404, the core
legal-technical property that /metrics facts/calculations/forecasts
are identical whether or not /attestation was ever called, one-time
publication, signature verification, and DB-level UPDATE/DELETE
rejection.
112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
"""Hosted Conversion Attestation publication (WP-0006-T06).
|
|
|
|
Preserves the non-discretionary rule already implemented in
|
|
`conversion.py`: a Conversion Event is true the instant the ledger fold
|
|
first reaches `Outstanding Target = 0`, independent of whether anything
|
|
ever publishes an attestation about it. This module's only job is to
|
|
*observe* that fact and publish signed evidence of it — it must never be
|
|
on the path that determines whether a conversion has legally occurred
|
|
(TSD §3.5 legal-technical rule, TrustServicePRD §3 point 3). Any caller can
|
|
independently confirm conversion status via `conversion.conversion_status`
|
|
(exposed publicly through `/phases/{id}/metrics`'s `facts.is_converted`)
|
|
without this module, this table, or this service being reachable at all.
|
|
|
|
An attestation, once published, is never regenerated or overwritten
|
|
(`migrations/0003_attestations.sql`) — a Phase converts at most once.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from . import conversion as conversion_module
|
|
from . import fold as fold_module
|
|
from . import hashing
|
|
|
|
if TYPE_CHECKING:
|
|
from psycopg import Connection
|
|
|
|
|
|
class NotConvertedError(ValueError):
|
|
"""Raised when publication is attempted for a Phase that has not
|
|
(yet) reached Outstanding Target = 0."""
|
|
|
|
|
|
def _find_conversion_prefix(
|
|
initial_target_amount: float, entries: list[dict[str, Any]]
|
|
) -> tuple[list[dict[str, Any]], str] | None:
|
|
"""Find the shortest ledger prefix at which the fold first reaches zero.
|
|
|
|
Returns (prefix_entries, conversion_timestamp) or None if the full
|
|
ledger never reaches Outstanding Target = 0. `conversion_timestamp` is
|
|
the crossing entry's own `recognized_at` — a fact derived from the
|
|
ledger, not the wall-clock time this function happens to run
|
|
(TSD §3.5: "the moment the ledger fold first reaches Outstanding
|
|
Target = 0", not "the moment someone asked").
|
|
"""
|
|
for k in range(1, len(entries) + 1):
|
|
result = fold_module.fold_outstanding_target(initial_target_amount, entries[:k])
|
|
if result.is_converted:
|
|
return entries[:k], entries[k - 1]["recognized_at"]
|
|
return None
|
|
|
|
|
|
def get_attestation(conn: Connection, phase_id: str) -> dict[str, Any] | None:
|
|
row = conn.execute(
|
|
"SELECT attestation FROM attestations WHERE phase_id = %s", (phase_id,)
|
|
).fetchone()
|
|
return row[0] if row else None
|
|
|
|
|
|
def publish_attestation(
|
|
conn: Connection,
|
|
manifest: dict[str, Any],
|
|
entries: list[dict[str, Any]],
|
|
signing_key,
|
|
) -> dict[str, Any]:
|
|
"""Publish (or return the already-published) Conversion Attestation.
|
|
|
|
Idempotent: if a Phase is already published, that record is returned
|
|
unchanged regardless of what the current full ledger looks like — a
|
|
later, unrelated ledger entry (e.g. a subsequent Phase's own bookkeeping
|
|
quirk, or an over-crediting correction) must never cause a second,
|
|
different attestation to be generated for the same Phase.
|
|
|
|
Raises NotConvertedError if the ledger has not (yet) reached
|
|
Outstanding Target = 0 anywhere in its history — this function never
|
|
fabricates a conversion, matching `conversion.generate_attestation`'s
|
|
own guard.
|
|
"""
|
|
phase_id = manifest["phase"]["id"]
|
|
existing = get_attestation(conn, phase_id)
|
|
if existing is not None:
|
|
return existing
|
|
|
|
initial_amount = manifest["phase"]["initial_target"]["amount"]
|
|
found = _find_conversion_prefix(initial_amount, entries)
|
|
if found is None:
|
|
raise NotConvertedError(
|
|
f"phase {phase_id!r} has not reached Outstanding Target = 0"
|
|
)
|
|
prefix_entries, conversion_timestamp = found
|
|
|
|
from psycopg.types.json import Jsonb # deferred: not needed for pure helpers/tests
|
|
|
|
attestation = conversion_module.generate_attestation(
|
|
manifest, prefix_entries, conversion_timestamp
|
|
)
|
|
signed = {**attestation, "signature": hashing.sign_record(attestation, signing_key)}
|
|
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO attestations (phase_id, attestation, signature)
|
|
VALUES (%s, %s, %s)
|
|
ON CONFLICT (phase_id) DO NOTHING
|
|
""",
|
|
(phase_id, Jsonb(signed), signed["signature"]),
|
|
)
|
|
# Someone else may have published concurrently between our SELECT and
|
|
# INSERT; re-read so every caller sees the single, first-published
|
|
# record rather than two independently signed copies of "the same"
|
|
# attestation with different (but both individually valid) signatures.
|
|
return get_attestation(conn, phase_id) or signed
|