state-hub/api/services/memo_forwarding.py

"""Pose, replay, and conflict ledger with injected transport.

Not imported by the API runtime. The private dispatcher composes this ledger
under a process lock; dev tests retain the injectable adapter. 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 MemoForwarder:
    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, *, dispatch: bool = True) -> 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)
        if not dispatch:
            return self._result(key, "pending", "queued")
        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()


# Compatibility for the original injectable development fixture.
DevIntakeFixture = MemoForwarder