tegwick
93b6ba2aae
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
100 lines
4.4 KiB
Python
100 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Redacted ledger failure notices to the platform-root State Hub inbox."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
|
|
from ledger_archive import Refused, checkpoint, private_directory, private_file
|
|
|
|
UNITS = {"email-connect-ledger-backup.service", "email-connect-ledger-freshness.service"}
|
|
TEST_UNIT = "email-connect-ledger-alert-test.service"
|
|
RECIPIENT = "platform-root"
|
|
|
|
|
|
def post(base, message):
|
|
request = Request(base.rstrip("/") + "/messages/", data=json.dumps(message).encode(),
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
with urlopen(request, timeout=15) as response:
|
|
result = json.loads(response.read(65536))
|
|
# A queue receipt is pending, never evidence of central inbox delivery.
|
|
if (result.get("to_agent") != RECIPIENT or result.get("from_agent") != "email-connect"
|
|
or result.get("subject") != message["subject"]
|
|
or result.get("body") != message["body"]):
|
|
raise Refused("inbox_receipt_not_verified")
|
|
uuid.UUID(result["id"])
|
|
return result["id"]
|
|
|
|
|
|
def notify(state, base, unit=None, test=False, sender=post):
|
|
if unit not in UNITS | {None, TEST_UNIT} or (test and unit is None):
|
|
raise Refused("invalid_notification_unit")
|
|
test = test or unit == TEST_UNIT
|
|
state.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
private_directory(state)
|
|
fd = os.open(state / "notify.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, "w") as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
if unit:
|
|
event = str(uuid.uuid4())
|
|
record = {
|
|
"event_id": event, "unit": unit, "test": test,
|
|
"observed_at": datetime.now(UTC).isoformat(), "status": "pending",
|
|
"message": {
|
|
"from_agent": "email-connect", "to_agent": RECIPIENT,
|
|
"subject": "email-connect ledger " + ("notification test" if test else "backup failure"),
|
|
"body": json.dumps({
|
|
"event_id": event, "unit": unit, "test": test,
|
|
"workplan": "EMAIL-WP-0007", "host": "railiance01",
|
|
"action": "Confirm inbox delivery only." if test else
|
|
"Inspect receiver readiness, backup pending evidence and archive freshness; "
|
|
"do not automatically restore, resend or discard a pending capture.",
|
|
}, sort_keys=True),
|
|
},
|
|
}
|
|
checkpoint(state / (event + ".json"), record)
|
|
delivered = []
|
|
for path in sorted(state.glob("*.json")):
|
|
private_file(path)
|
|
record = json.loads(path.read_text())
|
|
if record["status"] != "pending":
|
|
continue
|
|
message = record["message"]
|
|
if message["from_agent"] != "email-connect" or message["to_agent"] != RECIPIENT:
|
|
raise Refused("invalid_pending_notification")
|
|
# Delivery is at least once: if the hub accepted a request but the
|
|
# reply was lost, replay may duplicate its stable event_id.
|
|
message_id = sender(base, message)
|
|
record.update(status="delivered", message_id=message_id,
|
|
delivered_at=datetime.now(UTC).isoformat())
|
|
checkpoint(path, record)
|
|
delivered.append(message_id)
|
|
return {"status": "inbox_delivery_verified", "recipient": RECIPIENT,
|
|
"message_ids": delivered, "human_acknowledged": False}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--unit", choices=sorted(UNITS | {TEST_UNIT}))
|
|
parser.add_argument("--test", action="store_true")
|
|
parser.add_argument("--state-dir", type=Path,
|
|
default=Path("/var/lib/email-connect-backup/notifications"))
|
|
parser.add_argument("--api-base", default="http://10.43.68.154:8000")
|
|
args = parser.parse_args()
|
|
print(json.dumps(notify(args.state_dir, args.api_base, args.unit, args.test)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as error:
|
|
print(json.dumps({"status": "pending_notification_requires_replay",
|
|
"failure_type": type(error).__name__}))
|
|
raise SystemExit(1)
|