Implement durable recurring backup alert delivery candidate
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m16s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
This commit is contained in:
tegwick 2026-10-31 10:19:16 +01:00
parent ce6f2bdacf
commit 267d502d93
18 changed files with 944 additions and 1 deletions

View file

@ -0,0 +1,3 @@
# Non-secret candidate endpoints. Verify native identity/routing before installation.
STATE_HUB_BASE=http://10.43.68.154:8000
ALERTMANAGER_BASE=http://10.43.37.119:9093

View file

@ -0,0 +1,43 @@
{
"global": {
"resolve_timeout": "5m"
},
"route": {
"receiver": "pending-operator-channel",
"group_by": ["namespace", "alertname"],
"group_wait": "30s",
"group_interval": "5m",
"repeat_interval": "4h",
"routes": [
{
"matchers": [
"application=\"coulomb-social\"",
"owner=\"platform-root\"",
"alertname=\"CoulombBackupProblem\"",
"condition=~\"backup_failed|pending_backup|backup_stale\""
],
"group_by": ["alertname", "event_id"],
"receiver": "coulomb-backup-platform-root-email"
}
]
},
"receivers": [
{"name": "pending-operator-channel"},
{
"name": "coulomb-backup-platform-root-email",
"email_configs": [
{
"to": "bernd.worsch@gmail.com",
"from": "platform@coulomb.social",
"smarthost": "smtp.ionos.de:587",
"auth_username": "platform@coulomb.social",
"auth_password_file": "/etc/alertmanager/secrets/telemetry-alert-smtp/password",
"require_tls": true,
"send_resolved": true,
"headers": {"Subject": "[{{ .Status }}] Coulomb backup alert for platform-root"},
"text": "{{ range .Alerts }}Condition: {{ .Labels.condition }}\nVerification test: {{ .Labels.verification }}\nEvent: {{ .Labels.event_id }}\n{{ .Annotations.description }}\n\nA resolved notification indicates the producer lease ended or an explicit clear was submitted; verify the backup receipts before treating it as recovery.\n{{ end }}"
}
]
}
]
}

View file

@ -0,0 +1,19 @@
[Unit]
Description=Retain and deliver coulomb-social backup failure notification for %i
After=network-online.target
Wants=network-online.target
[Service]
Type=oneshot
User=root
UMask=0077
StateDirectory=coulomb-social-backup-alerts
StateDirectoryMode=0700
EnvironmentFile=/etc/coulomb-social-backup/alerting.env
ExecStart=/usr/bin/python3 /opt/coulomb-social-backup/notify-backup-alert.py --execute --failed-unit %i --state-dir /var/lib/coulomb-social-backup-alerts --hub-base ${STATE_HUB_BASE} --alertmanager-base ${ALERTMANAGER_BASE}
TimeoutStartSec=5min
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/var/lib/coulomb-social-backup-alerts
ProtectHome=true

View file

@ -1,14 +1,20 @@
[Unit]
Description=Check coulomb-social pending captures, paired backup, mirror and capacity evidence
OnFailure=coulomb-social-backup-alert@%n.service
[Service]
Type=oneshot
User=root
UMask=0077
StateDirectory=coulomb-social-backup-alerts
StateDirectoryMode=0700
EnvironmentFile=/etc/coulomb-social-backup/alerting.env
# Candidate limits: owner admission and received-alert proof precede activation.
ExecStart=/usr/bin/python3 /opt/coulomb-social-backup/check-paired-backup-health.py --mirror-receipt /var/lib/coulomb-social-backup/last-mirror.json --minimum-free-bytes 8589934592 --maximum-state-bytes 8589934592
ExecStart=/usr/bin/python3 /opt/coulomb-social-backup/observe-backup-alerts.py --execute --state-dir /var/lib/coulomb-social-backup --mirror-receipt /var/lib/coulomb-social-backup/last-mirror.json --alert-state-dir /var/lib/coulomb-social-backup-alerts --minimum-free-bytes 8589934592 --maximum-state-bytes 8589934592 --hub-base ${STATE_HUB_BASE} --alertmanager-base ${ALERTMANAGER_BASE}
TimeoutStartSec=5min
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/var/lib/coulomb-social-backup-alerts
ProtectHome=true

View file

@ -2,6 +2,7 @@
Description=Coherent encrypted coulomb-social database and content backup
After=network-online.target k3s.service
Wants=network-online.target
OnFailure=coulomb-social-backup-alert@%n.service
[Service]
Type=oneshot

View file

@ -0,0 +1,23 @@
{
"schema": "coulomb.backup-alert-isolated-smtp.v1",
"observed_at": "2026-10-31T09:14:32.318517+00:00",
"real_alertmanager_api_and_route_readback": true,
"synthetic_smtp_received": 1,
"event_id_in_email": true,
"loopback_only_smtp": true,
"external_email_sent": false,
"production_route_activated": false,
"transport_handoff": {
"status": "active_email_route_verified",
"accepted_at": "2026-10-31T09:14:32.318517+00:00",
"email_delivery_proven": false
},
"fixture_overrides": [
"loopback SMTP without TLS/auth",
"example.invalid addresses",
"zero group wait",
"one second group interval"
],
"rehearsal_source_sha256": "988c0bfc5397fee1093a44a65c30528f9426abd6130fbfdfb178a374b6f2a1eb",
"consumer_source_sha256": "07522437d4276b92f0a44149adc6c9e81bb18dbe59ce0f1f9e58a48347700d0e"
}

View file

@ -0,0 +1,51 @@
import datetime, importlib.util, json, pathlib, socket, socketserver, subprocess, tempfile, threading, time, urllib.request
ROOT=pathlib.Path('/home/worsch/coulomb-social')
spec=importlib.util.spec_from_file_location('alerts',ROOT/'scripts/backup_alerts.py');alerts=importlib.util.module_from_spec(spec);spec.loader.exec_module(alerts)
received=[]
class SMTP(socketserver.StreamRequestHandler):
def handle(self):
self.wfile.write(b'220 local synthetic SMTP\r\n');self.wfile.flush()
data=False;body=[]
while line:=self.rfile.readline():
if data:
if line==b'.\r\n':received.append(b''.join(body).decode());data=False;self.wfile.write(b'250 received\r\n')
else:body.append(line)
elif line.upper().startswith((b'EHLO',b'HELO')):self.wfile.write(b'250 localhost\r\n')
elif line.upper().startswith((b'MAIL',b'RCPT',b'RSET',b'NOOP')):self.wfile.write(b'250 OK\r\n')
elif line.upper().startswith(b'DATA'):data=True;body=[];self.wfile.write(b'354 send data\r\n')
elif line.upper().startswith(b'QUIT'):self.wfile.write(b'221 bye\r\n');self.wfile.flush();return
else:self.wfile.write(b'500 unsupported\r\n')
self.wfile.flush()
with tempfile.TemporaryDirectory(prefix='csoc-native-alert-') as temp, socketserver.ThreadingTCPServer(('127.0.0.1',0),SMTP) as smtp:
threading.Thread(target=smtp.serve_forever,daemon=True).start()
root=pathlib.Path(temp)
cid=subprocess.check_output(['docker','create','--network','none','sha256:690c7b525f4367aa91f73e2f91c632206d32e97c6384bdbf2fb7a861b420340d'],text=True).strip()
try:subprocess.run(['docker','cp',cid+':/bin/alertmanager',str(root/'alertmanager')],check=True,capture_output=True)
finally:subprocess.run(['docker','rm',cid],check=True,capture_output=True)
config=json.loads((ROOT/'deploy/backup/alertmanager-config.pending.json').read_text())
email=config['receivers'][1]['email_configs'][0]
email.pop('auth_username');email.pop('auth_password_file');email['to']='receiver@example.invalid';email['from']='sender@example.invalid';email['smarthost']='127.0.0.1:'+str(smtp.server_address[1]);email['require_tls']=False
config['route']['group_wait']='0s';config['route']['group_interval']='1s'
(root/'config.json').write_text(json.dumps(config))
with socket.socket() as sock:sock.bind(('127.0.0.1',0));port=sock.getsockname()[1]
endpoint='http://127.0.0.1:'+str(port)
process=subprocess.Popen([str(root/'alertmanager'),'--config.file='+str(root/'config.json'),'--storage.path='+str(root/'data'),'--web.listen-address=127.0.0.1:'+str(port),'--cluster.listen-address='],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
try:
for i in range(100):
if process.poll() is not None:raise RuntimeError('isolated_alertmanager_stopped')
try:urllib.request.urlopen(endpoint+'/-/ready',timeout=1).close();break
except Exception:time.sleep(.1)
else:raise RuntimeError('isolated_alertmanager_not_ready')
now=datetime.datetime.now(datetime.timezone.utc)
event={'event_id':'8f002752-1b8d-4aba-a67e-a76bf5768d41','condition':'backup_stale','test':True,'observed_at':now.isoformat()}
try:result=alerts.deliver_alertmanager(endpoint,event,now)
except Exception:
print(json.dumps(alerts.request_json(endpoint+"/api/v2/alerts")));raise
for i in range(100):
if received:break
time.sleep(.1)
assert len(received)==1 and event['event_id'] in received[0] and 'receiver@example.invalid' in received[0]
proof={'schema':'coulomb.backup-alert-isolated-smtp.v1','observed_at':now.isoformat(),'real_alertmanager_api_and_route_readback':True,'synthetic_smtp_received':1,'event_id_in_email':True,'loopback_only_smtp':True,'external_email_sent':False,'production_route_activated':False,'transport_handoff':result,'fixture_overrides':['loopback SMTP without TLS/auth','example.invalid addresses','zero group wait','one second group interval']}
(ROOT/'docs/release/backup-alert-isolated-smtp-2026-10-31.json').write_text(json.dumps(proof,indent=2)+'\n');print(json.dumps(proof))
finally:
process.terminate();process.wait(timeout=10);smtp.shutdown()

View file

@ -0,0 +1,107 @@
# Recurring backup alert integration — source candidate
Both destinations are accepted and their one-off tests received: platform-root's
State Hub inbox and bernd.worsch@gmail.com. The founder confirmed Gmail inbox
receipt; the exact proof Job and SMTP receipt were independently verified.
That does not activate recurring delivery.
The consumer now has a durable outbox and source wiring for recurring delivery.
It reuses State Hub and the existing Alertmanager/SMTP projection; it adds no
mail service, mailbox credential or public endpoint. Deployment remains pending.
## Consumer behavior
`scripts/backup_alerts.py` locks a private journal, writes and fsyncs each event
before a send, and checkpoints each channel separately. A failed State Hub call
does not prevent the email handoff, and vice versa. Replays retain the UUID.
State Hub delivery requires the exact central message receipt; a queue receipt
is insufficient. A lost HTTP reply can produce a duplicate inbox message with
the same event ID: this is at-least-once delivery, not exactly-once delivery.
Alertmanager submission uses its [v2 API](https://raw.githubusercontent.com/prometheus/alertmanager/main/api/v2/openapi.yaml).
The consumer reads back exact labels/times and the expected email receiver for
active events. Millisecond timestamps match the native API's serialization.
Acceptance is an email-transport handoff, never a recipient acknowledgment.
Alertmanager owns SMTP retry/grouping/silencing. No SMTP password enters this
consumer process, State Hub, output or repository.
`observe-backup-alerts.py` calls the existing complete health check. Pending and
stale reasons create their respective alert episodes; other evidence/capacity
failures create `backup_failed`. Repeated unhealthy observations refresh the same
episode; a later recurrence after recovery gets a new UUID. The existing hourly
freshness unit drives observation and replay. Its exit status measures whether
monitoring/delivery succeeded; backup health remains explicit in its JSON.
Both the paired-backup and freshness units have an OnFailure template handler.
The handler validates systemd's MONITOR_UNIT, MONITOR_SERVICE_RESULT and
MONITOR_INVOCATION_ID, and uses the failed invocation UUID for replay. A later
healthy report clears explicit failed-run events only if its fully verified
paired capture is newer than the failure. An old healthy backup cannot erase a
new failed-run alert. No automatic restore, capture replay or deletion is added.
Active alerts have a three-hour lease, refreshed hourly. Resolved emails explicitly
say that lease expiry or an explicit clear is not recovery proof. Independent
producer/notification-debt monitoring remains an operating requirement: a dead
producer cannot monitor itself. Native delivery and receipt tests must include
producer interruption, notification backlog and channel outage cases.
The separate private state directory is `/var/lib/coulomb-social-backup-alerts`.
Records are limited to 16 KiB each and 1,024 events; pressure at 75% produces a
health alert. Confirmed records are retained, not automatically deleted. This
bounds committed record count, not filesystem allocation or abandoned temporary
files after a crash. Reviewed archival/capacity handling is still required. A
full journal refuses new events while explicit replay of existing events remains
available. Processing rotates attempts to prevent bounded service executions
from permanently starving later records.
## Email route and ownership
`deploy/backup/alertmanager-config.pending.json` is a consumer route proposal,
not a live Helm values file. It selects only this application's three backup
conditions and platform-root owner, leaving unrelated alerts on the existing
pending receiver. SMTP uses the existing `telemetry-alert-smtp/password`
projection, sender platform@coulomb.social, verified TLS and the chosen Gmail
recipient. It includes resolved notifications so short incidents can still be
reported. The pending generic telemetry acknowledgment service is not required
or activated by this backup route; recipient receipt is not audit acknowledgment.
The observed live service is `telemetry-alertmanager`, ClusterIP 10.43.37.119,
port 9093; GET /api/v2/receivers returned only pending-operator-channel.
The route is therefore **not active**. `alerting.env.example` records candidate
private endpoints and contains no credential. Recheck their identities before
installation. rapp-telemetry owns the exact Helm merge, existing-secret mount,
current release diff, route activation and rollback. Preserve all other admitted
routes and mounted secrets when preparing that exact owner operation.
## Verification and remaining execution
Full `make check`: 460 passed, zero failures/errors/skips, including the mandatory
browser journeys. The 14 new outbox/producer checks cover independent failures,
queue receipts, replay, recovery, journal pressure and systemd invocation binding.
`backup-alert-recurring-sources-2026-10-31.json` binds the checked sources.
Native amtool validated the exact route configuration and all three positive
conditions plus four unrelated labelsets. Its container had no network.
`backup-alert-route-validation-2026-10-31.json` binds the image and config hash.
The actual target systemd 255 verified all five candidate units from temporary
files, then removed those files. Return code was zero; warnings concerned two
unrelated unreadable identity timers. Nothing was installed or enabled.
`backup-alert-unit-validation-2026-10-31.json` binds the exact unit files.
An isolated real Alertmanager process accepted/read back a synthetic event and
delivered an email to a loopback-only SMTP fixture. Its overrides replaced
recipient/sender with example.invalid addresses, SMTP with a local listener
without authentication/TLS, and grouping delays with short fixture values.
This proves API, routing, template and message correlation, not live TLS/custody.
The previous founder test proves the live SMTP projection separately. Receipt
and reproduction source are `backup-alert-isolated-smtp-2026-10-31.json` and
`.rehearsal.txt`. The local process, copied binary and state were removed.
Before activation, prepare and review the exact owner route/mount change and
rebuilt operating bundle, install consumer units disabled, and verify live
failed/pending/stale test events in both channels. Include channel failure,
restart/replay and producer-death detection evidence. Preserve stable test IDs
and do not replay the already completed one-off email Job. The existing capacity,
rolling retention, historical-copy disposition and service-contract obligations
remain required before the already approved schedule is enabled. WP-0012-T03
remains in progress; this candidate does not complete WP-0013/14/15 acceptance.

View file

@ -0,0 +1,26 @@
{
"schema": "coulomb.backup-alert-source-validation.v1",
"full_gate": "make check",
"results": {
"tests": 460,
"errors": 0,
"failures": 0,
"skipped": 0
},
"sources": {
"scripts/backup_alerts.py": "07522437d4276b92f0a44149adc6c9e81bb18dbe59ce0f1f9e58a48347700d0e",
"scripts/notify-backup-alert.py": "0bd53bc5ad22d8bcb0326fb18a17737744b06416fe785dad077308e45f051d5a",
"scripts/observe-backup-alerts.py": "17caf4fa65283242f618b039ff1d018801c9890ea2a9178912cbdd4b50ee2664",
"scripts/check-paired-backup-health.py": "2d40ec6e4c59ea7fe1d46065d197b9a87ca6811f023cc6f3e20e4184a3490566",
"tests/test_backup_alert_outbox.py": "eb1c3263999592d7bed70a6b092a65edc660558d1036409d3473ceb60d36cdf5",
"deploy/backup/alertmanager-config.pending.json": "03622039baa7ba953a72263df666bd81b9cf9469ccc6977c21fcfd23d4bc90c3",
"deploy/backup/alerting.env.example": "509844b5c8ef618b812d52f32e19c3c559623636dca62f7b16f437aa90cc0458",
"deploy/backup/coulomb-social-backup-alert@.service": "4bfa019d29b1d0de278f03553602a5e824de50ed99ebcf3b5035cf4d5b06fe98",
"deploy/backup/coulomb-social-backup-freshness.service": "8d6278c2cbca7f48b659c239c1d1e181fa4b9cf023fe7429850e294c6cd84d0e",
"deploy/backup/coulomb-social-paired-backup.service": "8b01b788bb08f453fdd4807104185c89411e0b0b4dedee9240f14f110ceb5513",
"deploy/backup/coulomb-social-backup-freshness.timer": "75d1d0721c07fdd0703910c06bb8a678396440e7d95411a37ee7de4938f3ac31",
"deploy/backup/coulomb-social-paired-backup.timer": "5e87235e66f89a64a972562d6f163f62420628fc44fb7fd8d603a96c347dfbb9"
},
"installed": false,
"recurring_alerts_activated": false
}

View file

@ -0,0 +1,12 @@
{
"schema": "coulomb.backup-alert-route-validation.v1",
"observed_at": "2026-10-31T09:12:50.612789+00:00",
"validator_image_id": "sha256:690c7b525f4367aa91f73e2f91c632206d32e97c6384bdbf2fb7a861b420340d",
"config_sha256": "03622039baa7ba953a72263df666bd81b9cf9469ccc6977c21fcfd23d4bc90c3",
"native_config_check": true,
"native_positive_routes": 3,
"native_unrelated_routes_excluded": 4,
"network": "none",
"live_configuration_changed": false,
"notification_sent": false
}

View file

@ -0,0 +1,15 @@
{
"returncode": 0,
"output": "net-kingdom-identity-backup.timer: Failed to open /etc/systemd/system/net-kingdom-identity-backup.timer: Permission denied\nnet-kingdom-identity-backup-health.timer: Failed to open /etc/systemd/system/net-kingdom-identity-backup-health.timer: Permission denied\n",
"units": 5,
"installed": false,
"schema": "coulomb.backup-alert-unit-validation.v1",
"observed_at": "2026-10-31T09:13:11.864616+00:00",
"source_sha256": {
"deploy/backup/coulomb-social-backup-alert@.service": "4bfa019d29b1d0de278f03553602a5e824de50ed99ebcf3b5035cf4d5b06fe98",
"deploy/backup/coulomb-social-backup-freshness.service": "8d6278c2cbca7f48b659c239c1d1e181fa4b9cf023fe7429850e294c6cd84d0e",
"deploy/backup/coulomb-social-paired-backup.service": "8b01b788bb08f453fdd4807104185c89411e0b0b4dedee9240f14f110ceb5513",
"deploy/backup/coulomb-social-backup-freshness.timer": "75d1d0721c07fdd0703910c06bb8a678396440e7d95411a37ee7de4938f3ac31",
"deploy/backup/coulomb-social-paired-backup.timer": "5e87235e66f89a64a972562d6f163f62420628fc44fb7fd8d603a96c347dfbb9"
}
}

View file

@ -1,5 +1,10 @@
# Backup operating source delivery review
**Recurring-alert update:** source integration is now prepared in
`backup-alert-recurring-integration-2026-10-31.md`. Both sealed bundles below
exclude the new outbox, observer, failure handler and route proposal. Neither is
a complete current installation bundle; rebuild and review before installation.
**Superseded after verified revision19:** the bundle below is historical and must
not be installed. Current preparation is
`backup-operating-delivery-revision19-2026-10-31.json`: archive060620da,

266
scripts/backup_alerts.py Normal file
View file

@ -0,0 +1,266 @@
"""Durable consumer outbox for the existing State Hub and Alertmanager owners.
No credentials or SMTP implementation. HTTP acceptance by Alertmanager is a
handoff, never proof of email delivery. Active events must be refreshed hourly;
their three-hour leases cannot replace independent monitoring of this producer.
"""
from __future__ import annotations
import fcntl
import json
import os
import stat
import tempfile
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path
from urllib.parse import urlencode
from urllib.request import Request, urlopen
CONDITIONS = {"backup_failed", "pending_backup", "backup_stale"}
RECEIVER = "coulomb-backup-platform-root-email"
MAX_EVENTS = 1024
MAX_BYTES = 16384
class Refused(ValueError):
pass
def private(path, directory=False):
info = path.lstat()
valid = stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode)
if (not valid or info.st_uid != os.geteuid() or info.st_mode & 0o077
or (not directory and info.st_nlink != 1)):
raise Refused("unsafe_alert_state")
def checkpoint(path, record):
data = json.dumps(record, sort_keys=True).encode()
if len(data) > MAX_BYTES:
raise Refused("alert_record_too_large")
fd, name = tempfile.mkstemp(prefix=".checkpoint-", dir=path.parent)
try:
with os.fdopen(fd, "wb") as stream:
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
os.replace(name, path)
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
try:
os.fsync(directory)
finally:
os.close(directory)
finally:
Path(name).unlink(missing_ok=True)
def timestamp(value):
parsed = datetime.fromisoformat(value)
if parsed.tzinfo is None:
raise Refused("timezone_required")
return parsed.astimezone(UTC)
def labels(record):
return {"alertname": "CoulombBackupProblem", "application": "coulomb-social",
"owner": "platform-root", "instance": "railiance01",
"condition": record["condition"], "event_id": record["event_id"],
"verification": "true" if record["test"] else "false"}
def message(record):
return {
"from_agent": "coulomb-social", "to_agent": "platform-root",
"subject": "Coulomb backup " + ("TEST " if record["test"] else "")
+ record["condition"],
"body": json.dumps({
"event_id": record["event_id"], "condition": record["condition"],
"observed_at": record["observed_at"], "test": record["test"],
"application": "coulomb-social", "host": "railiance01",
"workplan": "CSOC-WP-0012",
"action": "Inspect backup/freshness service and retained receipts. "
"Do not automatically restore, delete or discard a pending capture.",
}, sort_keys=True),
}
def request_json(url, payload=None):
request = Request(url, data=None if payload is None else json.dumps(payload).encode(),
headers={"Content-Type": "application/json"})
with urlopen(request, timeout=15) as response:
if response.status not in (200, 201):
raise Refused("unexpected_http_status")
data = response.read(65537)
if len(data) > 65536:
raise Refused("oversized_transport_response")
return json.loads(data) if data else None
def deliver_hub(base, record):
expected = message(record)
receipt = request_json(base.rstrip("/") + "/messages/", expected)
if not isinstance(receipt, dict) or any(receipt.get(k) != v for k, v in expected.items()):
raise Refused("central_inbox_receipt_required")
return str(uuid.UUID(receipt["id"]))
def deliver_alertmanager(base, record, now):
# Alertmanager's v2 API returns millisecond precision. Submit that same
# precision so exact readback does not falsely reject a successful handoff.
end = timestamp(record["resolved_at"]) if record.get("resolved_at") else (
now + timedelta(hours=3))
end = end.isoformat(timespec="milliseconds")
start = timestamp(record["observed_at"]).isoformat(timespec="milliseconds")
alert = {"labels": labels(record), "startsAt": start, "endsAt": end,
"annotations": {"summary": "Coulomb backup " + record["condition"],
"description": message(record)["body"]}}
endpoint = base.rstrip("/") + "/api/v2/alerts"
request_json(endpoint, [alert])
if not record.get("resolved_at"):
query = urlencode({"filter": 'event_id="' + record["event_id"] + '"'})
active = request_json(endpoint + "?" + query)
if not isinstance(active, list) or not any(
item.get("labels") == alert["labels"]
and timestamp(item["startsAt"]) == timestamp(alert["startsAt"])
and timestamp(item["endsAt"]) == timestamp(end)
and {r["name"] for r in item.get("receivers", [])} == {RECEIVER}
for item in active
):
raise Refused("reviewed_email_route_readback_required")
return {"status": "resolution_accepted" if record.get("resolved_at")
else "active_email_route_verified", "accepted_at": now.isoformat(),
"email_delivery_proven": False}
def load(path):
private(path)
if path.stat().st_size > MAX_BYTES:
raise Refused("alert_record_too_large")
with path.open() as stream:
record = json.load(stream)
if (record.get("schema") != "coulomb.backup-alert-event.v1"
or str(uuid.UUID(record["event_id"])) + ".json" != path.name
or record["condition"] not in CONDITIONS or type(record["test"]) is not bool):
raise Refused("invalid_alert_record")
timestamp(record["observed_at"])
if record.get("resolved_at"):
if timestamp(record["resolved_at"]) < timestamp(record["observed_at"]):
raise Refused("resolution_before_event")
return record
def drain(state, hub_base, alertmanager_base, *, event=None, resolve=None, now=None,
observations=None, recovered_capture=None,
hub_sender=deliver_hub, email_sender=deliver_alertmanager):
"""Serialize producers, checkpoint before sends, and attempt channels independently.
event is {event_id, condition, test}; a repeated ID must match. resolve names
an existing UUID. No automatic deletion: a full journal refuses new events
but still allows replay of existing events. Archive disposition is explicit.
"""
now = now or datetime.now(UTC)
if now.tzinfo is None:
raise Refused("timezone_required")
if event:
event = dict(event)
event["event_id"] = str(uuid.UUID(event["event_id"]))
if set(event) != {"event_id", "condition", "test"} or (
event["condition"] not in CONDITIONS or type(event["test"]) is not bool
):
raise Refused("invalid_event")
if resolve:
resolve = str(uuid.UUID(resolve))
if observations is not None and (set(observations) != CONDITIONS
or any(type(v) is not bool for v in observations.values())):
raise Refused("complete_health_observations_required")
if recovered_capture:
recovered_capture = timestamp(recovered_capture)
if observations is None or any(observations.values()) or recovered_capture > now:
raise Refused("healthy_new_capture_required")
state.mkdir(mode=0o700, parents=True, exist_ok=True)
private(state, directory=True)
if state.resolve() != state.absolute():
raise Refused("symlinked_alert_state")
fd = os.open(state / "notify.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
with os.fdopen(fd, "w") as lock:
private(state / "notify.lock")
fcntl.flock(lock, fcntl.LOCK_EX)
paths = sorted(state.glob("*.json"))
records = {p.stem: (p, load(p)) for p in paths}
if event:
key = event["event_id"]
if key in records:
if any(records[key][1][k] != v for k, v in event.items()):
raise Refused("event_id_reused_with_different_content")
else:
if len(records) >= MAX_EVENTS:
raise Refused("alert_journal_full_requires_archival")
record = {"schema": "coulomb.backup-alert-event.v1", **event,
"origin": "explicit",
"observed_at": now.isoformat(), "hub_message_id": None,
"alertmanager": None}
path = state / (key + ".json")
checkpoint(path, record)
records[key] = (path, record)
if observations is not None:
for condition, firing in sorted(observations.items()):
active = [(p, r) for p, r in records.values()
if r.get("origin") == "health" and r["condition"] == condition
and not r["test"] and not r.get("resolved_at")]
if firing and not active:
if len(records) >= MAX_EVENTS:
raise Refused("alert_journal_full_requires_archival")
key = str(uuid.uuid4())
record = {"schema": "coulomb.backup-alert-event.v1", "event_id": key,
"condition": condition, "test": False, "origin": "health",
"observed_at": now.isoformat(), "hub_message_id": None,
"alertmanager": None}
path = state / (key + ".json")
checkpoint(path, record)
records[key] = (path, record)
elif not firing:
for path, record in active:
record["resolved_at"] = now.isoformat()
checkpoint(path, record)
if recovered_capture:
for path, record in records.values():
if (record.get("origin") == "explicit" and not record["test"]
and record["condition"] == "backup_failed"
and not record.get("resolved_at")
and timestamp(record["observed_at"]) < recovered_capture):
record["resolved_at"] = now.isoformat()
checkpoint(path, record)
if resolve:
path, record = records[resolve]
if not record.get("resolved_at"):
record["resolved_at"] = now.isoformat()
checkpoint(path, record)
failures = []
# Rotate unfinished work across bounded systemd invocations: a slow or
# unreachable channel must not permanently starve later events.
ordered = sorted(records.items(), key=lambda item: item[1][1].get("last_attempt_at", ""))
for key, (path, record) in ordered:
if (record.get("hub_message_id")
and (record.get("alertmanager") or {}).get("status") == "resolution_accepted"):
continue
record["last_attempt_at"] = now.isoformat()
checkpoint(path, record)
if not record.get("hub_message_id"):
try:
record["hub_message_id"] = hub_sender(hub_base, record)
checkpoint(path, record)
except Exception:
failures.append({"event_id": key, "channel": "state_hub"})
accepted = record.get("alertmanager") or {}
if accepted.get("status") == "resolution_accepted":
continue
try:
record["alertmanager"] = email_sender(alertmanager_base, record, now)
checkpoint(path, record)
except Exception:
failures.append({"event_id": key, "channel": "email_handoff"})
return {"status": "pending_channel_retry" if failures else "channel_handoffs_verified",
"failures": failures, "events": len(records), "email_delivery_proven": False,
"automatic_monitoring_active": False}

View file

@ -182,6 +182,7 @@ def check(state, mirror_receipt, now, *, min_free_bytes, max_state_bytes,
captured = validate_primary(receipt)
age = (now - captured).total_seconds()
report["capture_age_seconds"] = round(age, 3)
report["captured_at"] = captured.isoformat()
if age < 0:
reasons.append("capture_timestamp_in_future")
elif age > 26 * 3600:

View file

@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""Queue/replay backup alerts through State Hub and the reviewed email route."""
import argparse
import json
import os
from pathlib import Path
from backup_alerts import drain
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--execute", action="store_true")
parser.add_argument("--state-dir", type=Path, required=True)
parser.add_argument("--hub-base", required=True)
parser.add_argument("--alertmanager-base", required=True)
parser.add_argument("--event-id")
parser.add_argument("--condition", choices=["backup_failed", "pending_backup", "backup_stale"])
parser.add_argument("--test", action="store_true")
parser.add_argument("--resolve")
parser.add_argument("--failed-unit", choices=["coulomb-social-paired-backup.service",
"coulomb-social-backup-freshness.service"])
args = parser.parse_args()
if args.failed_unit:
if (args.event_id or args.condition or args.test or args.resolve
or os.environ.get("MONITOR_UNIT") != args.failed_unit
or not os.environ.get("MONITOR_INVOCATION_ID")
or os.environ.get("MONITOR_SERVICE_RESULT") in (None, "success")):
parser.error("A matching failed systemd invocation is required")
args.event_id = os.environ["MONITOR_INVOCATION_ID"]
args.condition = "backup_failed"
if bool(args.event_id) != bool(args.condition) or (args.test and not args.event_id):
parser.error("event-id and condition are required together; test requires an event")
if not args.execute:
print(json.dumps({"status": "prepared_only", "notification_sent": False}))
return 0
event = {"event_id": args.event_id, "condition": args.condition,
"test": args.test} if args.event_id else None
try:
result = drain(args.state_dir, args.hub_base, args.alertmanager_base,
event=event, resolve=args.resolve)
except Exception as error:
result = {"status": "alert_delivery_requires_review", "failure_type": type(error).__name__,
"email_delivery_proven": False}
print(json.dumps(result))
return int(result["status"] != "channel_handoffs_verified")
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,61 @@
#!/usr/bin/env python3
"""Observe backup health, retain alert episodes, and refresh both delivery channels."""
import argparse
import importlib.util
import json
from datetime import UTC, datetime
from pathlib import Path
from backup_alerts import MAX_EVENTS, drain
spec = importlib.util.spec_from_file_location(
"backup_health", Path(__file__).with_name("check-paired-backup-health.py"))
health = importlib.util.module_from_spec(spec)
spec.loader.exec_module(health)
PENDING = {"multiple_pending_captures", "pending_requires_review", "backup_stalled"}
def observations(report):
reasons = set(report["reasons"])
return {"pending_backup": bool(reasons & PENDING), "backup_stale": "backup_stale" in reasons,
"backup_failed": bool(reasons - PENDING - {"backup_stale"})}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--execute", action="store_true")
parser.add_argument("--state-dir", type=Path, required=True)
parser.add_argument("--mirror-receipt", type=Path, required=True)
parser.add_argument("--alert-state-dir", type=Path, required=True)
parser.add_argument("--minimum-free-bytes", type=int, required=True)
parser.add_argument("--maximum-state-bytes", type=int, required=True)
parser.add_argument("--hub-base", required=True)
parser.add_argument("--alertmanager-base", required=True)
args = parser.parse_args()
now = datetime.now(UTC)
report = health.check(args.state_dir, args.mirror_receipt, now,
min_free_bytes=args.minimum_free_bytes,
max_state_bytes=args.maximum_state_bytes)
if sum(1 for _ in args.alert_state_dir.glob("*.json")) >= MAX_EVENTS * 3 // 4:
report["reasons"] = sorted(set(report["reasons"]) | {"alert_journal_near_capacity"})
report["status"] = "unhealthy"
if not args.execute:
print(json.dumps(report))
return int(report["status"] != "healthy")
try:
delivery = drain(args.alert_state_dir, args.hub_base, args.alertmanager_base,
now=now, observations=observations(report),
recovered_capture=report.get("captured_at")
if report["status"] == "healthy" else None)
except Exception as error:
delivery = {"status": "alert_delivery_requires_review",
"failure_type": type(error).__name__}
print(json.dumps({"health": report, "delivery": delivery}))
# The unit measures successful monitoring, not backup health. A delivered
# unhealthy observation must not trigger a second generic failure every hour.
return int(delivery["status"] != "channel_handoffs_verified")
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,236 @@
"""Exercise interrupted/partial delivery without contacting live services."""
import importlib.util
import json
import sys
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path
import pytest
SOURCE = Path(__file__).parents[1] / "scripts/backup_alerts.py"
spec = importlib.util.spec_from_file_location("backup_alerts_test", SOURCE)
alerts = importlib.util.module_from_spec(spec)
spec.loader.exec_module(alerts)
NOW = datetime(2026, 10, 31, 10, tzinfo=UTC)
EVENT = {"event_id": str(uuid.uuid4()), "condition": "backup_failed", "test": True}
def fail(*args):
raise TimeoutError("sensitive remote error must never be retained")
def email_ok(base, record, now):
return {"status": "resolution_accepted" if record.get("resolved_at")
else "active_email_route_verified", "accepted_at": now.isoformat(),
"email_delivery_proven": False}
def run(state, **kwargs):
options = {"now": NOW, "hub_sender": lambda *a: str(uuid.uuid4()),
"email_sender": email_ok}
options.update(kwargs)
return alerts.drain(state, "http://hub.invalid", "http://mail.invalid", **options)
def test_partial_failure_preserves_event_and_does_not_starve_other_channel(tmp_path):
state = tmp_path / "outbox"
result = run(state, event=EVENT, hub_sender=fail)
assert result["failures"] == [{"event_id": EVENT["event_id"], "channel": "state_hub"}]
record = json.loads((state / (EVENT["event_id"] + ".json")).read_text())
assert record["alertmanager"]["status"] == "active_email_route_verified"
assert record["hub_message_id"] is None
assert "sensitive" not in json.dumps(record)
assert result["email_delivery_proven"] is False
assert run(state, event=EVENT)["status"] == "channel_handoffs_verified"
assert len(list(state.glob("*.json"))) == 1
def test_replay_does_not_repeat_confirmed_hub_but_refreshes_email_lease(tmp_path):
state = tmp_path / "outbox"
run(state, event=EVENT, email_sender=fail)
calls = []
def email(base, record, now):
calls.append(record["event_id"])
return email_ok(base, record, now)
assert run(state, hub_sender=fail, email_sender=email)["failures"] == []
assert run(state, hub_sender=fail, email_sender=email)["failures"] == []
assert calls == [EVENT["event_id"], EVENT["event_id"]]
def test_resolution_retry_retains_original_event_and_stops_after_acceptance(tmp_path):
state = tmp_path / "outbox"
run(state, event=EVENT)
later = NOW + timedelta(hours=1)
run(state, resolve=EVENT["event_id"], now=later, email_sender=fail)
path = state / (EVENT["event_id"] + ".json")
record = json.loads(path.read_text())
assert record["resolved_at"] == later.isoformat()
assert run(state, now=later, hub_sender=fail)["failures"] == []
assert run(state, now=later, hub_sender=fail, email_sender=fail)["failures"] == []
assert path.exists() # Resolution does not discard evidence.
def test_reused_id_and_unsafe_journal_refuse_before_send(tmp_path):
state = tmp_path / "outbox"
run(state, event=EVENT)
with pytest.raises(alerts.Refused, match="reused"):
run(state, event={**EVENT, "condition": "backup_stale"})
path = state / (EVENT["event_id"] + ".json")
path.chmod(0o644)
with pytest.raises(alerts.Refused, match="unsafe"):
run(state, hub_sender=fail, email_sender=fail)
def test_full_journal_preserves_pending_and_existing_replay(tmp_path, monkeypatch):
state = tmp_path / "outbox"
monkeypatch.setattr(alerts, "MAX_EVENTS", 1)
run(state, event=EVENT, email_sender=fail)
with pytest.raises(alerts.Refused, match="journal_full"):
run(state, event={**EVENT, "event_id": str(uuid.uuid4())})
assert run(state)["failures"] == []
def test_hub_queue_receipt_is_not_delivery_and_exact_body_is_required(monkeypatch):
record = {**EVENT, "observed_at": NOW.isoformat()}
monkeypatch.setattr(alerts, "request_json", lambda *a: {"status": "queued", "id": "local"})
with pytest.raises(alerts.Refused, match="central_inbox"):
alerts.deliver_hub("http://hub.invalid", record)
expected = alerts.message(record)
uid = str(uuid.uuid4())
monkeypatch.setattr(alerts, "request_json", lambda *a: {**expected, "id": uid})
assert alerts.deliver_hub("http://hub.invalid", record) == uid
def test_email_route_readback_must_match_not_just_http_success(monkeypatch):
record = {**EVENT, "observed_at": NOW.isoformat()}
calls = []
receiver = "pending-operator-channel"
def request(url, payload=None):
calls.append((url, payload))
if payload is not None:
return None
return [{**calls[-2][1][0], "receivers": [{"name": receiver}]}]
monkeypatch.setattr(alerts, "request_json", request)
with pytest.raises(alerts.Refused, match="email_route"):
alerts.deliver_alertmanager("http://mail.invalid", record, NOW)
receiver = alerts.RECEIVER
result = alerts.deliver_alertmanager("http://mail.invalid", record, NOW)
assert result["email_delivery_proven"] is False
assert alerts.timestamp(calls[-2][1][0]["endsAt"]) == NOW + timedelta(hours=3)
assert calls[-2][1][0]["labels"]["event_id"] == EVENT["event_id"]
def test_native_api_millisecond_precision_is_used_for_exact_readback(monkeypatch):
now = NOW.replace(microsecond=123456)
record = {**EVENT, "observed_at": now.isoformat()}
posted = []
def request(url, payload=None):
if payload:
posted.extend(payload)
return None
return [{**posted[0], "receivers": [{"name": alerts.RECEIVER}]}]
monkeypatch.setattr(alerts, "request_json", request)
assert alerts.deliver_alertmanager("http://mail.invalid", record, now)["status"] == (
"active_email_route_verified")
assert posted[0]["startsAt"].endswith(".123+00:00")
assert posted[0]["endsAt"].endswith(".123+00:00")
def test_checkpoint_failure_before_first_send_leaves_no_unjournaled_send(tmp_path, monkeypatch):
calls = []
monkeypatch.setattr(alerts, "checkpoint", fail)
with pytest.raises(TimeoutError):
run(tmp_path / "outbox", event=EVENT, hub_sender=lambda *a: calls.append(a),
email_sender=lambda *a: calls.append(a))
assert calls == []
def test_health_episode_deduplicates_then_resolves_and_recurrence_is_new(tmp_path):
state = tmp_path / "outbox"
observations = {c: c == "pending_backup" for c in alerts.CONDITIONS}
run(state, observations=observations)
run(state, observations=observations, now=NOW + timedelta(hours=1))
assert len(list(state.glob("*.json"))) == 1
run(state, observations=dict.fromkeys(alerts.CONDITIONS, False),
now=NOW + timedelta(hours=2))
first = json.loads(next(state.glob("*.json")).read_text())
assert first["alertmanager"]["status"] == "resolution_accepted"
run(state, observations=observations, now=NOW + timedelta(hours=3))
assert len(list(state.glob("*.json"))) == 2
def test_service_failure_needs_a_newer_healthy_paired_capture_to_clear(tmp_path):
state = tmp_path / "outbox"
event = {**EVENT, "test": False}
run(state, event=event)
healthy = dict.fromkeys(alerts.CONDITIONS, False)
path = state / (EVENT["event_id"] + ".json")
run(state, observations=healthy, recovered_capture=(NOW - timedelta(hours=1)).isoformat())
assert "resolved_at" not in json.loads(path.read_text())
run(state, observations=healthy, recovered_capture=(NOW + timedelta(minutes=1)).isoformat(),
now=NOW + timedelta(hours=1))
assert json.loads(path.read_text())["alertmanager"]["status"] == "resolution_accepted"
def test_unknown_health_and_future_capture_cannot_clear_incidents(tmp_path):
with pytest.raises(alerts.Refused, match="complete_health"):
run(tmp_path / "outbox", observations={"backup_stale": False})
with pytest.raises(alerts.Refused, match="healthy_new_capture"):
run(tmp_path / "outbox", observations=dict.fromkeys(alerts.CONDITIONS, False),
recovered_capture=(NOW + timedelta(hours=1)).isoformat())
def cli_module(monkeypatch, name):
monkeypatch.syspath_prepend(str(SOURCE.parent))
module_spec = importlib.util.spec_from_file_location(name, SOURCE.with_name(name + ".py"))
module = importlib.util.module_from_spec(module_spec)
module_spec.loader.exec_module(module)
return module
def test_journal_pressure_emits_failure_before_full_without_claiming_recovery(
tmp_path, monkeypatch, capsys,
):
monitor = cli_module(monkeypatch, "observe-backup-alerts")
monkeypatch.setattr(monitor, "MAX_EVENTS", 4)
for i in range(3):
(tmp_path / f"{i}.json").write_text("synthetic")
monkeypatch.setattr(monitor.health, "check", lambda *a, **kw: {
"status": "healthy", "reasons": [], "captured_at": NOW.isoformat()})
calls = []
def drain(*args, **kwargs):
calls.append(kwargs)
return {"status": "channel_handoffs_verified"}
monkeypatch.setattr(monitor, "drain", drain)
monkeypatch.setattr(sys, "argv", ["observe", "--execute", "--state-dir", str(tmp_path),
"--mirror-receipt", str(tmp_path / "unused"), "--alert-state-dir", str(tmp_path),
"--minimum-free-bytes", "1", "--maximum-state-bytes", "1",
"--hub-base", "http://hub.invalid", "--alertmanager-base", "http://mail.invalid"])
assert monitor.main() == 0 # Monitoring succeeded, while backup health is unhealthy.
assert calls[0]["observations"]["backup_failed"] is True
assert calls[0]["recovered_capture"] is None
assert json.loads(capsys.readouterr().out)["health"]["status"] == "unhealthy"
def test_failure_handler_requires_matching_systemd_invocation(tmp_path, monkeypatch):
handler = cli_module(monkeypatch, "notify-backup-alert")
unit = "coulomb-social-paired-backup.service"
monkeypatch.setattr(sys, "argv", ["notify", "--execute", "--failed-unit", unit,
"--state-dir", str(tmp_path), "--hub-base", "http://hub.invalid",
"--alertmanager-base", "http://mail.invalid"])
monkeypatch.setenv("MONITOR_UNIT", "other.service")
monkeypatch.setenv("MONITOR_INVOCATION_ID", EVENT["event_id"].replace("-", ""))
monkeypatch.setenv("MONITOR_SERVICE_RESULT", "exit-code")
with pytest.raises(SystemExit):
handler.main()
calls = []
def drain(*a, **kw):
calls.append(kw)
return {"status": "channel_handoffs_verified"}
monkeypatch.setattr(handler, "drain", drain)
monkeypatch.setenv("MONITOR_UNIT", unit)
assert handler.main() == 0
assert calls[0]["event"] == {**EVENT, "event_id": EVENT["event_id"].replace("-", ""),
"test": False}

View file

@ -480,6 +480,23 @@ handling and native received-signal evidence are still required, alongside the
existing operating/service-contract obligations. Destination and one-off test
acceptance are complete and should not be requested again.
2026-10-31 recurring alert implementation: durable per-channel outbox now connects
the candidate hourly health observer and systemd failure handlers to State Hub
and the existing Alertmanager email owner. Stable event IDs, independent channel
retry, restart persistence, route readback and newer-paired-point failure clearing
are implemented; no credentials are added to the consumer. Three-hour alert
leases require hourly refresh and independent producer monitoring. A dedicated
route proposal preserves unrelated routing. Native amtool config/positive and
negative route checks pass; native systemd verifies all five candidate units.
An isolated real Alertmanager delivered the synthetic event to loopback SMTP;
this caught and fixed the API's millisecond timestamp precision requirement.
Review: docs/release/backup-alert-recurring-integration-2026-10-31.md.
Live Alertmanager still has only the pending receiver. Owner route/mount review,
current operating bundle rebuild/install, native dual-channel incident/replay and
producer-death proof remain required. Neither old sealed operating bundle is a
complete current installer. All runtime/timer/capacity/retention and other service
acceptance gates remain unchanged; no new external alert was sent this turn.
## Completion
Finish when every task above has its stated evidence and acceptance, or hand off any newly discovered residual to a live work record first. Cross-owner dependencies remain their existing canonical records.