tegwick
ff59199c3b
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029c-c003-7283-acc9-930b434e6cae
101 lines
4.2 KiB
Python
101 lines
4.2 KiB
Python
"""State Hub decision write-back (INFD-WP-0005, FrictionlessDecisionIntake §8).
|
|
|
|
When a human's terminal disposition (accept or decline) answers a memo that
|
|
serves a hub decision with ``resolves: on-answer``, resolve that decision with
|
|
the memo id in the rationale. ``POST /decisions/{id}/resolve`` is the hub's
|
|
sanctioned write; it reflects an act already taken here and decides nothing.
|
|
|
|
Off unless ``INFD_HUB_WRITEBACK=1``: when off, the payload is built and
|
|
returned as ``skipped`` so the evidence exists, but nothing is sent. The
|
|
transport is injected; this module opens no socket by itself.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Callable, Mapping
|
|
|
|
FLAG = "INFD_HUB_WRITEBACK"
|
|
RESOLVING_VERBS = frozenset({"accept", "decline"})
|
|
_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
|
|
|
|
#: (method, url, json body) -> (status, parsed json or {})
|
|
Transport = Callable[[str, str, dict], tuple[int, dict]]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Answer:
|
|
"""A human's recorded terminal disposition on one memo."""
|
|
|
|
memo_id: str
|
|
memo_version: int
|
|
verb: str
|
|
disposition_id: str
|
|
decided_at: str
|
|
reviewer_display: str
|
|
approval_id: str | None
|
|
approval_binding_digest: str | None
|
|
view_hash: str
|
|
actor_kind: str = "person"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WritebackResult:
|
|
status: str # not_applicable | skipped | resolved | already_resolved | conflict | failed
|
|
decision_id: str | None
|
|
payload: dict | None = None
|
|
detail: str = ""
|
|
|
|
|
|
def enabled(env: Mapping[str, str] | None = None) -> bool:
|
|
return (os.environ if env is None else env).get(FLAG) == "1"
|
|
|
|
|
|
def build_resolution(answer: Answer) -> dict:
|
|
if answer.actor_kind != "person":
|
|
raise ValueError("only a human disposition can resolve a hub decision")
|
|
if answer.verb not in RESOLVING_VERBS:
|
|
raise ValueError(f"verb {answer.verb!r} does not resolve a decision")
|
|
rationale = (
|
|
f"{answer.verb} in informed-decision memo {answer.memo_id} v{answer.memo_version}; "
|
|
f"approval {answer.approval_id or 'none'}; binding {answer.approval_binding_digest or 'none'}; "
|
|
f"view_hash {answer.view_hash}; disposition {answer.disposition_id} at {answer.decided_at}"
|
|
)
|
|
return {
|
|
"decided_by": f"{answer.reviewer_display} via informed-decision",
|
|
"rationale": rationale,
|
|
"write_log": True,
|
|
}
|
|
|
|
|
|
def write_back(answer: Answer, serves: Mapping | None, *, base_url: str,
|
|
transport: Transport, env: Mapping[str, str] | None = None) -> WritebackResult:
|
|
if not serves or serves.get("resolves") != "on-answer" or answer.verb not in RESOLVING_VERBS:
|
|
return WritebackResult("not_applicable", (serves or {}).get("hub_decision_id"))
|
|
decision_id = serves.get("hub_decision_id", "")
|
|
if not _UUID_RE.match(decision_id or ""):
|
|
raise ValueError("hub_decision_id must be a lowercase UUID")
|
|
payload = build_resolution(answer)
|
|
if not enabled(env):
|
|
return WritebackResult("skipped", decision_id, payload, f"{FLAG} is not 1")
|
|
url = f"{base_url.rstrip('/')}/decisions/{decision_id}/resolve"
|
|
try:
|
|
status, body = transport("POST", url, payload)
|
|
except OSError as exc:
|
|
return WritebackResult("failed", decision_id, payload, type(exc).__name__)
|
|
if 200 <= status < 300:
|
|
return WritebackResult("resolved", decision_id, payload)
|
|
if status == 409:
|
|
# Already resolved: ours if the hub rationale names this memo.
|
|
try:
|
|
read_status, current = transport("GET", f"{base_url.rstrip('/')}/decisions/{decision_id}", {})
|
|
except OSError as exc:
|
|
return WritebackResult("failed", decision_id, payload, type(exc).__name__)
|
|
if read_status != 200 or not isinstance(current, dict) or not isinstance(current.get('rationale'), str):
|
|
return WritebackResult("failed", decision_id, payload, "resolution_unreadable")
|
|
if f"memo {answer.memo_id} " in (current.get("rationale") or ""):
|
|
return WritebackResult("already_resolved", decision_id, payload)
|
|
return WritebackResult("conflict", decision_id, payload, json.dumps(body)[:200])
|
|
return WritebackResult("failed", decision_id, payload, f"http_{status}")
|