331 lines
18 KiB
Python
331 lines
18 KiB
Python
"""Durable, account-bound human release gate inbox."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from src.private_state import connect_private_sqlite
|
|
|
|
|
|
_HASH = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$")
|
|
_PROJECT = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
|
|
_STATES = {"pending", "released", "held", "superseded"}
|
|
_CHECKLIST = {"exact_hash", "artifacts_reviewed", "provenance_reviewed"}
|
|
|
|
|
|
class GateConflict(RuntimeError):
|
|
"""The gate or idempotency revision no longer matches."""
|
|
|
|
|
|
class GateValidationError(ValueError):
|
|
"""The producer or reviewer payload is not safe to persist."""
|
|
|
|
|
|
class GateNotFound(LookupError):
|
|
"""No gate exists for this account."""
|
|
|
|
|
|
def _canonical(value: object) -> str:
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
|
|
|
|
|
def _fingerprint(value: object) -> str:
|
|
return hashlib.sha256(_canonical(value).encode()).hexdigest()
|
|
|
|
|
|
class HumanGateStore:
|
|
def __init__(self, path: str | Path, *, clock: Callable[[], float]):
|
|
self.path = Path(path)
|
|
self.clock = clock
|
|
self._initialize()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
connection = connect_private_sqlite(self.path, timeout=2.0)
|
|
connection.row_factory = sqlite3.Row
|
|
return connection
|
|
|
|
def _initialize(self) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS human_gates (
|
|
id TEXT PRIMARY KEY, login TEXT NOT NULL, source TEXT NOT NULL,
|
|
project TEXT NOT NULL, candidate_hash TEXT NOT NULL,
|
|
title TEXT NOT NULL, priority INTEGER NOT NULL,
|
|
payload_json TEXT NOT NULL, state TEXT NOT NULL,
|
|
revision INTEGER NOT NULL, created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL, superseded_by TEXT,
|
|
decision_reason TEXT NOT NULL DEFAULT '',
|
|
override_reason TEXT NOT NULL DEFAULT '',
|
|
checklist_json TEXT NOT NULL DEFAULT '{}',
|
|
UNIQUE(login, source, project, candidate_hash)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS human_gates_queue
|
|
ON human_gates(login, state, priority DESC, created_at, id);
|
|
CREATE TABLE IF NOT EXISTS human_gate_intake_keys (
|
|
login TEXT NOT NULL, idempotency_key TEXT NOT NULL,
|
|
fingerprint TEXT NOT NULL, gate_id TEXT NOT NULL,
|
|
PRIMARY KEY(login, idempotency_key)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS human_gate_history (
|
|
sequence INTEGER PRIMARY KEY AUTOINCREMENT, gate_id TEXT NOT NULL,
|
|
action TEXT NOT NULL, at REAL NOT NULL, details_json TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS human_gate_receipts (
|
|
receipt_id TEXT PRIMARY KEY, login TEXT NOT NULL,
|
|
idempotency_key TEXT NOT NULL, fingerprint TEXT NOT NULL,
|
|
gate_id TEXT NOT NULL, receipt_json TEXT NOT NULL,
|
|
UNIQUE(login, idempotency_key)
|
|
);
|
|
"""
|
|
)
|
|
|
|
@staticmethod
|
|
def _login(value: str) -> str:
|
|
value = str(value).strip().lower()
|
|
if not value or len(value) > 255:
|
|
raise GateValidationError("login is required")
|
|
return value
|
|
|
|
@staticmethod
|
|
def _key(value: str) -> str:
|
|
value = str(value).strip()
|
|
if not value or len(value) > 128:
|
|
raise GateValidationError("Idempotency key is required")
|
|
return value
|
|
|
|
@staticmethod
|
|
def _candidate(raw: dict) -> dict:
|
|
if not isinstance(raw, dict):
|
|
raise GateValidationError("Candidate must be an object")
|
|
source = str(raw.get("source", "")).strip()
|
|
project = str(raw.get("project", "")).strip()
|
|
candidate_hash = str(raw.get("candidate_hash", "")).strip()
|
|
title = " ".join(str(raw.get("title", "")).split())
|
|
priority = raw.get("priority", 0)
|
|
if not source or len(source) > 128:
|
|
raise GateValidationError("source is invalid")
|
|
if not _PROJECT.fullmatch(project):
|
|
raise GateValidationError("project is invalid")
|
|
if not _HASH.fullmatch(candidate_hash):
|
|
raise GateValidationError("candidate hash is invalid")
|
|
if not title or len(title) > 300:
|
|
raise GateValidationError("title is invalid")
|
|
if isinstance(priority, bool) or not isinstance(priority, int) or not 0 <= priority <= 100:
|
|
raise GateValidationError("priority is invalid")
|
|
normalized = {
|
|
"source": source, "project": project, "candidate_hash": candidate_hash,
|
|
"title": title, "priority": priority,
|
|
"artifacts": HumanGateStore._references(raw.get("artifacts", []), "name"),
|
|
"links": HumanGateStore._references(raw.get("links", []), "label"),
|
|
"checks": HumanGateStore._checks(raw.get("checks", [])),
|
|
"score": raw.get("score") if isinstance(raw.get("score"), dict) else {},
|
|
"provenance": raw.get("provenance") if isinstance(raw.get("provenance"), dict) else {},
|
|
}
|
|
if len(_canonical(normalized)) > 100_000:
|
|
raise GateValidationError("Candidate payload is too large")
|
|
return normalized
|
|
|
|
@staticmethod
|
|
def _references(raw: object, label: str) -> list[dict]:
|
|
if not isinstance(raw, list) or len(raw) > 50:
|
|
raise GateValidationError("references are invalid")
|
|
result = []
|
|
for item in raw:
|
|
if not isinstance(item, dict):
|
|
raise GateValidationError("reference is invalid")
|
|
name, url = str(item.get(label, "")).strip(), str(item.get("url", "")).strip()
|
|
if not name or len(name) > 200 or not url.startswith("https://") or len(url) > 2048:
|
|
raise GateValidationError("reference is invalid")
|
|
result.append({label: name, "url": url})
|
|
return result
|
|
|
|
@staticmethod
|
|
def _checks(raw: object) -> list[dict]:
|
|
if not isinstance(raw, list) or len(raw) > 100:
|
|
raise GateValidationError("checks are invalid")
|
|
result = []
|
|
for item in raw:
|
|
if not isinstance(item, dict):
|
|
raise GateValidationError("check is invalid")
|
|
name, state = str(item.get("name", "")).strip(), item.get("state")
|
|
if not name or len(name) > 200 or state not in {"success", "failure", "pending", "skipped"}:
|
|
raise GateValidationError("check is invalid")
|
|
result.append({"name": name, "state": state, "required": bool(item.get("required", True))})
|
|
return result
|
|
|
|
def _history(self, connection: sqlite3.Connection, gate_id: str) -> list[dict]:
|
|
return [
|
|
{"sequence": row[0], "action": row[1], "at": row[2], **json.loads(row[3])}
|
|
for row in connection.execute(
|
|
"SELECT sequence, action, at, details_json FROM human_gate_history WHERE gate_id=? ORDER BY sequence",
|
|
(gate_id,),
|
|
)
|
|
]
|
|
|
|
def _present(self, connection: sqlite3.Connection, row: sqlite3.Row, *, history: bool = False) -> dict:
|
|
payload = json.loads(row["payload_json"])
|
|
item = {
|
|
"id": row["id"], **payload, "state": row["state"], "revision": row["revision"],
|
|
"created_at": row["created_at"], "updated_at": row["updated_at"],
|
|
"superseded_by": row["superseded_by"],
|
|
}
|
|
if row["state"] in {"released", "held"}:
|
|
item.update({
|
|
"reason": row["decision_reason"], "override_reason": row["override_reason"],
|
|
"checklist": json.loads(row["checklist_json"]),
|
|
})
|
|
if history:
|
|
item["history"] = self._history(connection, row["id"])
|
|
return item
|
|
|
|
def has_intake_key(self, login: str, idempotency_key: str) -> bool:
|
|
with self._connect() as connection:
|
|
return connection.execute(
|
|
"SELECT 1 FROM human_gate_intake_keys WHERE login=? AND idempotency_key=?",
|
|
(self._login(login), self._key(idempotency_key)),
|
|
).fetchone() is not None
|
|
|
|
def intake(self, login: str, raw: dict, *, idempotency_key: str) -> dict:
|
|
login, key, payload = self._login(login), self._key(idempotency_key), self._candidate(raw)
|
|
fingerprint = _fingerprint(payload)
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
prior = connection.execute(
|
|
"SELECT fingerprint, gate_id FROM human_gate_intake_keys WHERE login=? AND idempotency_key=?",
|
|
(login, key),
|
|
).fetchone()
|
|
if prior:
|
|
if prior["fingerprint"] != fingerprint:
|
|
raise GateConflict("Idempotency key was already used for another candidate")
|
|
row = connection.execute("SELECT * FROM human_gates WHERE id=? AND login=?", (prior["gate_id"], login)).fetchone()
|
|
return self._present(connection, row, history=True)
|
|
existing = connection.execute(
|
|
"SELECT * FROM human_gates WHERE login=? AND source=? AND project=? AND candidate_hash=?",
|
|
(login, payload["source"], payload["project"], payload["candidate_hash"]),
|
|
).fetchone()
|
|
if existing:
|
|
old_payload = json.loads(existing["payload_json"])
|
|
immutable_old = {key: value for key, value in old_payload.items() if key != "checks"}
|
|
immutable_new = {key: value for key, value in payload.items() if key != "checks"}
|
|
if immutable_old != immutable_new:
|
|
raise GateConflict("Candidate hash is already bound to different facts")
|
|
if old_payload != payload:
|
|
if existing["state"] == "superseded":
|
|
raise GateConflict("Candidate hash was superseded by a newer candidate")
|
|
now = float(self.clock())
|
|
revision = existing["revision"] + 1
|
|
reopened = existing["state"] in {"released", "held"}
|
|
state = "pending" if reopened else existing["state"]
|
|
action = "reopened" if reopened else "updated"
|
|
connection.execute(
|
|
"UPDATE human_gates SET payload_json=?, state=?, revision=?, updated_at=?, decision_reason='', override_reason='', checklist_json='{}' WHERE id=?",
|
|
(_canonical(payload), state, revision, now, existing["id"]),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO human_gate_history(gate_id,action,at,details_json) VALUES (?,?,?,?)",
|
|
(existing["id"], action, now, _canonical({"candidate_hash": payload["candidate_hash"]})),
|
|
)
|
|
existing = connection.execute(
|
|
"SELECT * FROM human_gates WHERE id=? AND login=?",
|
|
(existing["id"], login),
|
|
).fetchone()
|
|
connection.execute("INSERT INTO human_gate_intake_keys VALUES (?,?,?,?)", (login, key, fingerprint, existing["id"]))
|
|
return self._present(connection, existing, history=True)
|
|
now, gate_id = float(self.clock()), str(uuid.uuid4())
|
|
old_rows = connection.execute(
|
|
"SELECT id, revision FROM human_gates WHERE login=? AND source=? AND project=? AND state='pending'",
|
|
(login, payload["source"], payload["project"]),
|
|
).fetchall()
|
|
connection.execute(
|
|
"INSERT INTO human_gates(id,login,source,project,candidate_hash,title,priority,payload_json,state,revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?, 'pending',1,?,?)",
|
|
(gate_id, login, payload["source"], payload["project"], payload["candidate_hash"], payload["title"], payload["priority"], _canonical(payload), now, now),
|
|
)
|
|
connection.execute("INSERT INTO human_gate_history(gate_id,action,at,details_json) VALUES (?,?,?,?)", (gate_id, "intake", now, _canonical({"candidate_hash": payload["candidate_hash"]})))
|
|
for old in old_rows:
|
|
connection.execute("UPDATE human_gates SET state='superseded', revision=?, updated_at=?, superseded_by=? WHERE id=? AND state='pending'", (old["revision"] + 1, now, gate_id, old["id"]))
|
|
connection.execute("INSERT INTO human_gate_history(gate_id,action,at,details_json) VALUES (?,?,?,?)", (old["id"], "superseded", now, _canonical({"superseded_by": gate_id, "candidate_hash": payload["candidate_hash"]})))
|
|
connection.execute("INSERT INTO human_gate_intake_keys VALUES (?,?,?,?)", (login, key, fingerprint, gate_id))
|
|
row = connection.execute("SELECT * FROM human_gates WHERE id=?", (gate_id,)).fetchone()
|
|
return self._present(connection, row, history=True)
|
|
|
|
def list(self, login: str, *, state: str = "pending", limit: int = 100) -> dict:
|
|
login = self._login(login)
|
|
if state not in _STATES and state != "all":
|
|
raise GateValidationError("state is invalid")
|
|
limit = min(max(int(limit), 1), 100)
|
|
with self._connect() as connection:
|
|
pending_count = connection.execute("SELECT COUNT(*) FROM human_gates WHERE login=? AND state='pending'", (login,)).fetchone()[0]
|
|
where, args = ("login=?", [login]) if state == "all" else ("login=? AND state=?", [login, state])
|
|
rows = connection.execute(f"SELECT * FROM human_gates WHERE {where} ORDER BY CASE WHEN state='pending' THEN 0 ELSE 1 END, priority DESC, created_at, id LIMIT ?", (*args, limit)).fetchall()
|
|
return {"pending_count": pending_count, "items": [self._present(connection, row) for row in rows]}
|
|
|
|
def detail(self, login: str, gate_id: str) -> dict:
|
|
with self._connect() as connection:
|
|
row = connection.execute("SELECT * FROM human_gates WHERE login=? AND id=?", (self._login(login), gate_id)).fetchone()
|
|
if row is None:
|
|
raise GateNotFound("Gate not found")
|
|
return self._present(connection, row, history=True)
|
|
|
|
def decide(self, login: str, gate_id: str, *, expected_revision: int, decision: str, reason: str, override_reason: str, checklist: dict, idempotency_key: str) -> dict:
|
|
login, key = self._login(login), self._key(idempotency_key)
|
|
reason, override_reason = str(reason).strip(), str(override_reason).strip()
|
|
if decision not in {"release", "hold"}:
|
|
raise GateValidationError("decision is invalid")
|
|
if decision == "hold" and not reason:
|
|
raise GateValidationError("Hold reason is required")
|
|
if decision == "release" and (
|
|
not isinstance(checklist, dict)
|
|
or set(checklist) != _CHECKLIST
|
|
or not all(value is True for value in checklist.values())
|
|
):
|
|
raise GateValidationError("A complete release checklist is required")
|
|
if decision == "hold":
|
|
checklist = {}
|
|
request = {"gate_id": gate_id, "expected_revision": expected_revision, "decision": decision, "reason": reason, "override_reason": override_reason, "checklist": checklist}
|
|
fingerprint = _fingerprint(request)
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
prior = connection.execute("SELECT fingerprint, receipt_json FROM human_gate_receipts WHERE login=? AND idempotency_key=?", (login, key)).fetchone()
|
|
if prior:
|
|
if prior["fingerprint"] != fingerprint:
|
|
raise GateConflict("Idempotency key was already used for another decision")
|
|
return json.loads(prior["receipt_json"])
|
|
row = connection.execute("SELECT * FROM human_gates WHERE login=? AND id=?", (login, gate_id)).fetchone()
|
|
if row is None:
|
|
raise GateNotFound("Gate not found")
|
|
if row["state"] != "pending" or row["revision"] != expected_revision:
|
|
raise GateConflict("Gate revision is stale")
|
|
payload = json.loads(row["payload_json"])
|
|
unmet = [check["name"] for check in payload["checks"] if check["required"] and check["state"] != "success"]
|
|
if decision == "release" and unmet and not override_reason:
|
|
raise GateValidationError("An explicit override reason is required for unmet checks")
|
|
now, receipt_id, revision = float(self.clock()), str(uuid.uuid4()), row["revision"] + 1
|
|
state = "released" if decision == "release" else "held"
|
|
connection.execute("UPDATE human_gates SET state=?,revision=?,updated_at=?,decision_reason=?,override_reason=?,checklist_json=? WHERE id=?", (state, revision, now, reason, override_reason, _canonical(checklist), gate_id))
|
|
receipt = {"receipt_id": receipt_id, "gate_id": gate_id, "candidate_hash": row["candidate_hash"], "state": state, "revision": revision, "decided_at": now, "reason": reason, "override_reason": override_reason, "checklist": checklist, "unmet_required_checks": unmet}
|
|
connection.execute("INSERT INTO human_gate_history(gate_id,action,at,details_json) VALUES (?,?,?,?)", (gate_id, state, now, _canonical({"receipt_id": receipt_id, "reason": reason, "override_reason": override_reason, "unmet_required_checks": unmet})))
|
|
connection.execute("INSERT INTO human_gate_receipts VALUES (?,?,?,?,?,?)", (receipt_id, login, key, fingerprint, gate_id, _canonical(receipt)))
|
|
return receipt
|
|
|
|
def has_receipt_key(self, login: str, idempotency_key: str) -> bool:
|
|
with self._connect() as connection:
|
|
return connection.execute(
|
|
"SELECT 1 FROM human_gate_receipts WHERE login=? AND idempotency_key=?",
|
|
(self._login(login), self._key(idempotency_key)),
|
|
).fetchone() is not None
|
|
|
|
def receipt(self, login: str, receipt_id: str) -> dict:
|
|
with self._connect() as connection:
|
|
row = connection.execute("SELECT receipt_json FROM human_gate_receipts WHERE login=? AND receipt_id=?", (self._login(login), receipt_id)).fetchone()
|
|
if row is None:
|
|
raise GateNotFound("Receipt not found")
|
|
return json.loads(row[0])
|