441 lines
18 KiB
Python
441 lines
18 KiB
Python
|
"""Dev-only pose, replay, and conflict fixture.
|
||
|
|
|
||
|
|
Not imported by the API runtime and not scheduled. It never reads a token
|
||
|
|
file, mints a credential, or dials a default host. Tests inject the transport
|
||
|
|
and, when a pose should carry a header, a token callable. The callable's
|
||
|
|
value is not retained.
|
||
|
|
|
||
|
|
Choosing the workload identity that projects an intake-audience token, and
|
||
|
|
reviewing that identity into the submitter registry, is an owner step. This
|
||
|
|
fixture does not name or admit an account.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import sqlite3
|
||
|
|
import time
|
||
|
|
from contextlib import contextmanager
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Callable, Iterator
|
||
|
|
|
||
|
|
from api.services.informed_decision_contract import (
|
||
|
|
PROVENANCE_NOTE,
|
||
|
|
ContractError,
|
||
|
|
bounded_receipt,
|
||
|
|
correlate_resolution,
|
||
|
|
resolution_applies,
|
||
|
|
safe_error_code,
|
||
|
|
submission_digest,
|
||
|
|
validate_envelope,
|
||
|
|
)
|
||
|
|
|
||
|
|
Transport = Callable[[str, str, dict, dict], tuple[int, dict]]
|
||
|
|
TokenReader = Callable[[], str]
|
||
|
|
|
||
|
|
_RETRYABLE = frozenset({408, 429})
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class ForwardResult:
|
||
|
|
status: str
|
||
|
|
idempotency_key: str
|
||
|
|
receipt: dict | None
|
||
|
|
drafted_by: str | None
|
||
|
|
drafted_by_provenance: str
|
||
|
|
provenance_note: str
|
||
|
|
detail: str = ""
|
||
|
|
|
||
|
|
|
||
|
|
class DevIntakeFixture:
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
path: str | Path,
|
||
|
|
*,
|
||
|
|
transport: Transport,
|
||
|
|
intake_base_url: str,
|
||
|
|
hub_base_url: str,
|
||
|
|
token_reader: TokenReader | None = None,
|
||
|
|
clock: Callable[[], float] = time.time,
|
||
|
|
) -> None:
|
||
|
|
self.path = str(path)
|
||
|
|
self.transport = transport
|
||
|
|
self.intake_base_url = _base_url(intake_base_url)
|
||
|
|
self.hub_base_url = _base_url(hub_base_url)
|
||
|
|
self.token_reader = token_reader
|
||
|
|
self.clock = clock
|
||
|
|
Path(self.path).parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
with self._connect() as db:
|
||
|
|
db.executescript(
|
||
|
|
"""
|
||
|
|
CREATE TABLE IF NOT EXISTS pose_ledger (
|
||
|
|
idempotency_key TEXT PRIMARY KEY,
|
||
|
|
submission_digest TEXT NOT NULL,
|
||
|
|
envelope_json TEXT NOT NULL,
|
||
|
|
drafted_by TEXT,
|
||
|
|
drafted_by_provenance TEXT NOT NULL
|
||
|
|
CHECK (drafted_by_provenance IN ('asserted', 'absent')),
|
||
|
|
provenance_note TEXT NOT NULL,
|
||
|
|
state TEXT NOT NULL
|
||
|
|
CHECK (state IN ('pending', 'delivered', 'blocked')),
|
||
|
|
receipt_json TEXT,
|
||
|
|
attempts INTEGER NOT NULL DEFAULT 0,
|
||
|
|
next_attempt REAL NOT NULL DEFAULT 0,
|
||
|
|
last_error TEXT,
|
||
|
|
blocked_reason TEXT,
|
||
|
|
created_at TEXT NOT NULL,
|
||
|
|
updated_at TEXT NOT NULL
|
||
|
|
);
|
||
|
|
CREATE TABLE IF NOT EXISTS resolution_ledger (
|
||
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
|
|
decision_id TEXT NOT NULL,
|
||
|
|
memo_id TEXT NOT NULL,
|
||
|
|
verb TEXT NOT NULL,
|
||
|
|
resolves TEXT,
|
||
|
|
status TEXT NOT NULL,
|
||
|
|
error_type TEXT,
|
||
|
|
payload_json TEXT,
|
||
|
|
next_attempt REAL,
|
||
|
|
created_at TEXT NOT NULL
|
||
|
|
);
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
|
||
|
|
def submit(self, envelope: dict) -> ForwardResult:
|
||
|
|
checked = validate_envelope(envelope)
|
||
|
|
key = checked["idempotency_key"]
|
||
|
|
body_digest = submission_digest(checked)
|
||
|
|
drafted_by = checked.get("drafted_by")
|
||
|
|
provenance = "asserted" if drafted_by else "absent"
|
||
|
|
now = self.clock()
|
||
|
|
early: tuple[str, str] | None = None
|
||
|
|
with self._connect() as db:
|
||
|
|
row = db.execute(
|
||
|
|
"SELECT * FROM pose_ledger WHERE idempotency_key=?", (key,)
|
||
|
|
).fetchone()
|
||
|
|
if row is None:
|
||
|
|
db.execute(
|
||
|
|
"""INSERT INTO pose_ledger (
|
||
|
|
idempotency_key, submission_digest, envelope_json, drafted_by,
|
||
|
|
drafted_by_provenance, provenance_note, state, attempts,
|
||
|
|
next_attempt, created_at, updated_at
|
||
|
|
) VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, 0, ?, ?)""",
|
||
|
|
(
|
||
|
|
key, body_digest, _dump(checked), drafted_by, provenance,
|
||
|
|
PROVENANCE_NOTE, _iso(now), _iso(now),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
elif row["submission_digest"] != body_digest:
|
||
|
|
db.execute(
|
||
|
|
"""UPDATE pose_ledger SET blocked_reason='idempotency_conflict',
|
||
|
|
updated_at=? WHERE idempotency_key=?""",
|
||
|
|
(_iso(now), key),
|
||
|
|
)
|
||
|
|
if row["state"] != "delivered":
|
||
|
|
db.execute(
|
||
|
|
"UPDATE pose_ledger SET state='blocked' WHERE idempotency_key=?",
|
||
|
|
(key,),
|
||
|
|
)
|
||
|
|
early = ("blocked", "idempotency_conflict")
|
||
|
|
elif row["state"] == "delivered":
|
||
|
|
early = ("replay", "replay")
|
||
|
|
elif row["state"] == "blocked":
|
||
|
|
early = ("blocked", row["blocked_reason"] or "blocked")
|
||
|
|
elif row["next_attempt"] > now:
|
||
|
|
early = ("pending", row["last_error"] or "waiting")
|
||
|
|
if early is not None:
|
||
|
|
return self._result(key, *early)
|
||
|
|
return self._attempt(key)
|
||
|
|
|
||
|
|
def run_due(self) -> ForwardResult | None:
|
||
|
|
now = self.clock()
|
||
|
|
with self._connect() as db:
|
||
|
|
row = db.execute(
|
||
|
|
"""SELECT idempotency_key FROM pose_ledger
|
||
|
|
WHERE state='pending' AND next_attempt<=?
|
||
|
|
ORDER BY created_at LIMIT 1""",
|
||
|
|
(now,),
|
||
|
|
).fetchone()
|
||
|
|
key = None if row is None else row["idempotency_key"]
|
||
|
|
if key is None:
|
||
|
|
return None
|
||
|
|
return self._attempt(key)
|
||
|
|
|
||
|
|
def pose(self, key: str) -> dict | None:
|
||
|
|
with self._connect() as db:
|
||
|
|
row = db.execute(
|
||
|
|
"SELECT * FROM pose_ledger WHERE idempotency_key=?", (key,)
|
||
|
|
).fetchone()
|
||
|
|
return dict(row) if row is not None else None
|
||
|
|
|
||
|
|
def observe_resolution(
|
||
|
|
self, *, decision_id: str, memo_id: str, verb: str, serves: dict | None, payload: dict | None
|
||
|
|
) -> str:
|
||
|
|
resolves = serves.get("resolves") if isinstance(serves, dict) else None
|
||
|
|
if not resolution_applies(verb, serves):
|
||
|
|
self._append_resolution(
|
||
|
|
decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
|
||
|
|
status="not_applicable", error_type=None, payload=None, retry_at=None,
|
||
|
|
)
|
||
|
|
return "not_applicable"
|
||
|
|
prior = self._latest_resolution(decision_id, memo_id)
|
||
|
|
if prior and prior["status"] in {"resolved", "already_resolved", "conflict"}:
|
||
|
|
return prior["status"]
|
||
|
|
if prior and prior["status"] == "failed":
|
||
|
|
if prior["next_attempt"] is not None and prior["next_attempt"] > self.clock():
|
||
|
|
return "failed"
|
||
|
|
if payload is None and prior["payload_json"]:
|
||
|
|
payload = json.loads(prior["payload_json"])
|
||
|
|
if not isinstance(payload, dict):
|
||
|
|
raise ValueError("resolution payload is required")
|
||
|
|
encoded = _dump(payload)
|
||
|
|
if len(encoded.encode("utf-8")) > 8192:
|
||
|
|
self._append_resolution(
|
||
|
|
decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
|
||
|
|
status="conflict", error_type="payload_unbounded", payload=None, retry_at=None,
|
||
|
|
)
|
||
|
|
return "conflict"
|
||
|
|
try:
|
||
|
|
status, _body = self.transport(
|
||
|
|
"POST",
|
||
|
|
f"{self.hub_base_url}/decisions/{decision_id}/resolve",
|
||
|
|
payload,
|
||
|
|
{},
|
||
|
|
)
|
||
|
|
status = int(status)
|
||
|
|
except OSError as exc:
|
||
|
|
self._append_resolution(
|
||
|
|
decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
|
||
|
|
status="failed", error_type=type(exc).__name__, payload=payload,
|
||
|
|
retry_at=self.clock() + _backoff(1),
|
||
|
|
)
|
||
|
|
return "failed"
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
self._append_resolution(
|
||
|
|
decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
|
||
|
|
status="failed", error_type="transport_invalid", payload=payload,
|
||
|
|
retry_at=self.clock() + _backoff(1),
|
||
|
|
)
|
||
|
|
return "failed"
|
||
|
|
existing = None
|
||
|
|
if status == 409:
|
||
|
|
try:
|
||
|
|
read_status, existing = self.transport(
|
||
|
|
"GET", f"{self.hub_base_url}/decisions/{decision_id}", {}, {}
|
||
|
|
)
|
||
|
|
if int(read_status) != 200 or not isinstance(existing, dict):
|
||
|
|
existing = None
|
||
|
|
raise OSError("resolution_unreadable")
|
||
|
|
except OSError as exc:
|
||
|
|
error = exc.__class__.__name__ if str(exc) != "resolution_unreadable" else "resolution_unreadable"
|
||
|
|
self._append_resolution(
|
||
|
|
decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
|
||
|
|
status="failed", error_type=error, payload=payload,
|
||
|
|
retry_at=self.clock() + _backoff(1),
|
||
|
|
)
|
||
|
|
return "failed"
|
||
|
|
outcome = correlate_resolution(memo_id, post_status=status, existing=existing)
|
||
|
|
self._append_resolution(
|
||
|
|
decision_id=decision_id, memo_id=memo_id, verb=verb, resolves=resolves,
|
||
|
|
status=outcome, error_type=None if outcome != "failed" else f"http_{status}",
|
||
|
|
payload=payload, retry_at=None if outcome != "failed" else self.clock() + _backoff(1),
|
||
|
|
)
|
||
|
|
return outcome
|
||
|
|
|
||
|
|
def run_due_resolution(self) -> str | None:
|
||
|
|
now = self.clock()
|
||
|
|
with self._connect() as db:
|
||
|
|
row = db.execute(
|
||
|
|
"""SELECT * FROM resolution_ledger
|
||
|
|
WHERE status='failed' AND next_attempt IS NOT NULL AND next_attempt<=?
|
||
|
|
ORDER BY id DESC LIMIT 1""",
|
||
|
|
(now,),
|
||
|
|
).fetchone()
|
||
|
|
pending = dict(row) if row is not None else None
|
||
|
|
if pending is None:
|
||
|
|
return None
|
||
|
|
payload = json.loads(pending["payload_json"]) if pending["payload_json"] else None
|
||
|
|
serves = {"hub_decision_id": pending["decision_id"], "resolves": pending["resolves"]}
|
||
|
|
return self.observe_resolution(
|
||
|
|
decision_id=pending["decision_id"],
|
||
|
|
memo_id=pending["memo_id"],
|
||
|
|
verb=pending["verb"],
|
||
|
|
serves=serves,
|
||
|
|
payload=payload,
|
||
|
|
)
|
||
|
|
|
||
|
|
def resolutions(self) -> list[dict]:
|
||
|
|
with self._connect() as db:
|
||
|
|
rows = db.execute(
|
||
|
|
"SELECT * FROM resolution_ledger ORDER BY id"
|
||
|
|
).fetchall()
|
||
|
|
return [dict(row) for row in rows]
|
||
|
|
|
||
|
|
def _attempt(self, key: str) -> ForwardResult:
|
||
|
|
with self._connect() as db:
|
||
|
|
row = db.execute(
|
||
|
|
"SELECT * FROM pose_ledger WHERE idempotency_key=?", (key,)
|
||
|
|
).fetchone()
|
||
|
|
current = dict(row) if row is not None else None
|
||
|
|
if current is None or current["state"] != "pending":
|
||
|
|
return self._result(key, current["state"] if current else "blocked", "not_pending")
|
||
|
|
envelope = json.loads(current["envelope_json"])
|
||
|
|
token, token_error = self._read_token()
|
||
|
|
if token_error:
|
||
|
|
self._mark_retry(key, token_error)
|
||
|
|
return self._result(key, "pending", token_error)
|
||
|
|
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||
|
|
try:
|
||
|
|
status, body = self.transport(
|
||
|
|
"POST", f"{self.intake_base_url}/v1/memos", envelope, headers
|
||
|
|
)
|
||
|
|
status = int(status)
|
||
|
|
except OSError as exc:
|
||
|
|
self._mark_retry(key, type(exc).__name__)
|
||
|
|
return self._result(key, "pending", type(exc).__name__)
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
self._mark_retry(key, "transport_invalid")
|
||
|
|
return self._result(key, "pending", "transport_invalid")
|
||
|
|
if status in (200, 201):
|
||
|
|
try:
|
||
|
|
receipt = bounded_receipt(body, envelope["memo"]["id"])
|
||
|
|
except ContractError as exc:
|
||
|
|
code = safe_error_code(exc.code, "receipt_mismatch")
|
||
|
|
self._mark_blocked(key, code)
|
||
|
|
return self._result(key, "blocked", code)
|
||
|
|
self._mark_delivered(key, receipt)
|
||
|
|
return self._result(key, "delivered", "delivered")
|
||
|
|
if status == 409 or (status not in _RETRYABLE and 400 <= status < 500):
|
||
|
|
code = safe_error_code(
|
||
|
|
body.get("error") if isinstance(body, dict) else None,
|
||
|
|
"conflict" if status == 409 else f"http_{status}",
|
||
|
|
)
|
||
|
|
self._mark_blocked(key, code)
|
||
|
|
return self._result(key, "blocked", code)
|
||
|
|
self._mark_retry(key, f"http_{status}")
|
||
|
|
return self._result(key, "pending", f"http_{status}")
|
||
|
|
|
||
|
|
def _read_token(self) -> tuple[str | None, str | None]:
|
||
|
|
if self.token_reader is None:
|
||
|
|
return None, "projected_token_absent"
|
||
|
|
try:
|
||
|
|
token = self.token_reader()
|
||
|
|
except Exception as exc:
|
||
|
|
return None, type(exc).__name__
|
||
|
|
if not isinstance(token, str) or not token.strip() or token != token.strip():
|
||
|
|
return None, "projected_token_absent"
|
||
|
|
if len(token) > 16384:
|
||
|
|
return None, "projected_token_absent"
|
||
|
|
return token, None
|
||
|
|
|
||
|
|
def _mark_retry(self, key: str, error: str) -> None:
|
||
|
|
now = self.clock()
|
||
|
|
with self._connect() as db:
|
||
|
|
attempts = db.execute(
|
||
|
|
"SELECT attempts FROM pose_ledger WHERE idempotency_key=?", (key,)
|
||
|
|
).fetchone()["attempts"] + 1
|
||
|
|
db.execute(
|
||
|
|
"""UPDATE pose_ledger SET state='pending', attempts=?, last_error=?,
|
||
|
|
next_attempt=?, updated_at=? WHERE idempotency_key=?""",
|
||
|
|
(attempts, error, now + _backoff(attempts), _iso(now), key),
|
||
|
|
)
|
||
|
|
|
||
|
|
def _mark_blocked(self, key: str, reason: str) -> None:
|
||
|
|
now = self.clock()
|
||
|
|
with self._connect() as db:
|
||
|
|
db.execute(
|
||
|
|
"""UPDATE pose_ledger SET state='blocked', blocked_reason=?, last_error=?,
|
||
|
|
updated_at=? WHERE idempotency_key=?""",
|
||
|
|
(reason, reason, _iso(now), key),
|
||
|
|
)
|
||
|
|
|
||
|
|
def _mark_delivered(self, key: str, receipt: dict) -> None:
|
||
|
|
now = self.clock()
|
||
|
|
with self._connect() as db:
|
||
|
|
db.execute(
|
||
|
|
"""UPDATE pose_ledger SET state='delivered', receipt_json=?, last_error=NULL,
|
||
|
|
blocked_reason=NULL, updated_at=? WHERE idempotency_key=?""",
|
||
|
|
(_dump(receipt), _iso(now), key),
|
||
|
|
)
|
||
|
|
|
||
|
|
def _result(self, key: str, status: str, detail: str) -> ForwardResult:
|
||
|
|
row = self.pose(key)
|
||
|
|
receipt = json.loads(row["receipt_json"]) if row and row["receipt_json"] else None
|
||
|
|
return ForwardResult(
|
||
|
|
status=status,
|
||
|
|
idempotency_key=key,
|
||
|
|
receipt=receipt,
|
||
|
|
drafted_by=row["drafted_by"] if row else None,
|
||
|
|
drafted_by_provenance=row["drafted_by_provenance"] if row else "absent",
|
||
|
|
provenance_note=row["provenance_note"] if row else PROVENANCE_NOTE,
|
||
|
|
detail=detail,
|
||
|
|
)
|
||
|
|
|
||
|
|
def _append_resolution(
|
||
|
|
self, *, decision_id, memo_id, verb, resolves, status, error_type, payload, retry_at
|
||
|
|
) -> None:
|
||
|
|
with self._connect() as db:
|
||
|
|
db.execute(
|
||
|
|
"""UPDATE resolution_ledger SET next_attempt=NULL
|
||
|
|
WHERE decision_id=? AND memo_id=? AND status='failed'""",
|
||
|
|
(decision_id, memo_id),
|
||
|
|
)
|
||
|
|
db.execute(
|
||
|
|
"""INSERT INTO resolution_ledger (
|
||
|
|
decision_id, memo_id, verb, resolves, status, error_type,
|
||
|
|
payload_json, next_attempt, created_at
|
||
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
|
|
(
|
||
|
|
decision_id, memo_id, verb, resolves, status, error_type,
|
||
|
|
_dump(payload) if payload is not None else None,
|
||
|
|
retry_at, _iso(self.clock()),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
def _latest_resolution(self, decision_id: str, memo_id: str) -> dict | None:
|
||
|
|
with self._connect() as db:
|
||
|
|
row = db.execute(
|
||
|
|
"""SELECT * FROM resolution_ledger
|
||
|
|
WHERE decision_id=? AND memo_id=? ORDER BY id DESC LIMIT 1""",
|
||
|
|
(decision_id, memo_id),
|
||
|
|
).fetchone()
|
||
|
|
return dict(row) if row is not None else None
|
||
|
|
|
||
|
|
@contextmanager
|
||
|
|
def _connect(self) -> Iterator[sqlite3.Connection]:
|
||
|
|
db = sqlite3.connect(self.path)
|
||
|
|
db.row_factory = sqlite3.Row
|
||
|
|
try:
|
||
|
|
db.execute("PRAGMA journal_mode=WAL")
|
||
|
|
yield db
|
||
|
|
db.commit()
|
||
|
|
except Exception:
|
||
|
|
db.rollback()
|
||
|
|
raise
|
||
|
|
finally:
|
||
|
|
db.close()
|
||
|
|
|
||
|
|
|
||
|
|
def _base_url(value: str) -> str:
|
||
|
|
if (
|
||
|
|
not isinstance(value, str)
|
||
|
|
or not (value.startswith("http://") or value.startswith("https://"))
|
||
|
|
):
|
||
|
|
raise ValueError("base_url is required")
|
||
|
|
if "127.0.0.1" in value or "localhost" in value:
|
||
|
|
raise ValueError("base_url must be an injected test host")
|
||
|
|
return value.rstrip("/")
|
||
|
|
|
||
|
|
|
||
|
|
def _backoff(attempts: int) -> float:
|
||
|
|
return float(min(3600, 2 ** min(attempts, 11)))
|
||
|
|
|
||
|
|
|
||
|
|
def _dump(value: dict) -> str:
|
||
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||
|
|
|
||
|
|
|
||
|
|
def _iso(timestamp: float) -> str:
|
||
|
|
return datetime.fromtimestamp(timestamp, timezone.utc).isoformat()
|