From e3a7332b9150e58f54298972de5a44959cd319b8 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sun, 11 Oct 2026 00:14:45 -0400 Subject: [PATCH 1/5] feat: prototype generation-bound failure threshold recovery --- README.md | 6 ++ alertmanager_sink.py | 13 +++- chaski.py | 11 +++ emitter.py | 46 ++++++++++++- failure_family_adapter.py | 119 +++++++++++++++++++++++++++++++++ test_failure_family_adapter.py | 30 +++++++++ test_resolution.py | 90 +++++++++++++++++++++++++ 7 files changed, 310 insertions(+), 5 deletions(-) create mode 100644 failure_family_adapter.py create mode 100644 test_failure_family_adapter.py create mode 100644 test_resolution.py diff --git a/README.md b/README.md index 6d3cbdc..9cebd9d 100644 --- a/README.md +++ b/README.md @@ -173,3 +173,9 @@ reconciliation retain their existing behavior. This flag does not move the separate ROOT event-rule consumer to SSE. Closing a network reader can wait for its bounded 45-second socket timeout. Source tests are not permission to enable this flag, replay production history or claim natural scheduled acceptance. + +### Repeated command failures + +`failure_family_adapter.py` consumes the `dp failure-families` JSON report and a directly typed Quipu `ConfigSetting` containing a single JSON description. Policy fields are `default_threshold`, `window_seconds`, `default_owner`, optional `family_thresholds` (hashed family ID to threshold), and `tool_owners` (tool name to crew owner). Invalid/missing/ambiguous policy or failed aggregation exits nonzero, leaving the emitter UNKNOWN. Ownership comes from this graph policy, with an explicit crew fallback; human recipients are refused. + +Configure the emitter with `from: BELOW`, `to: ABOVE`, `resolve_event: recovered`, `generation_event_ids: true`, and `missing_is_from: true`. Keep `baseline_emits: false`: first sighting above threshold is a baseline, not an observed crossing. A complete successful rolling-window inventory can resolve absent families; partial observations and failed runs cannot. Unsupported program shapes are a separate disclosed measurement population, not silently attributed to a known family. Each actual crossing gets a generation-bound identity, stable across retries and restarts. Recovery refers to the original frozen firing labels and cannot overtake its receipt. The existing Alertmanager sink retains its finite TTL; scheduled delivery and producer-liveness monitoring are still deployment requirements. This adapter is not scheduled or enabled by the repository itself. diff --git a/alertmanager_sink.py b/alertmanager_sink.py index 7acb3ff..1e06212 100644 --- a/alertmanager_sink.py +++ b/alertmanager_sink.py @@ -90,7 +90,7 @@ def labels(self, event: dict) -> dict: "alertname": self.alert["alertname"], "severity": self.alert["severity"], "emitter": self.emitter_label, - "event_id": event["event_id"], + "event_id": event.get("resolution_of") or event["event_id"], "item": event["item"], }) kind = event.get("work_kind") @@ -105,6 +105,8 @@ def labels(self, event: dict) -> dict: def payload(self, event: dict) -> dict: start = event["observed_at"] end = max(start, self._clock()) + self.ttl_s + if event.get("resolution_of"): + end = event["resolved_at"] fields = {k: ("" if v is None else v) for k, v in event.items()} summary = self.alert.get("summary", "{event} for {item}").format_map(_Missing(fields)) annotations = {"summary": summary} @@ -114,12 +116,17 @@ def payload(self, event: dict) -> dict: "startsAt": _iso(start), "endsAt": _iso(end)} def _received(self, event: dict) -> bool: - q = urllib.parse.urlencode([("filter", f'event_id="{event["event_id"]}"'), + identity = event.get("resolution_of") or event["event_id"] + q = urllib.parse.urlencode([("filter", f'event_id="{identity}"'), ("filter", f'emitter="{self.emitter_label}"')]) status, raw = self._request("GET", f"/api/v2/alerts?{q}") if status // 100 != 2 or not raw: raise ConnectionError(f"read-back: HTTP {status}, {len(raw)} bytes") - return any(a.get("labels", {}).get("event_id") == event["event_id"] for a in json.loads(raw)) + rows = json.loads(raw) + if not isinstance(rows, list) or any(not isinstance(a, dict) for a in rows): + raise ConnectionError("read-back: malformed alert inventory") + active = any(a.get("labels", {}).get("event_id") == identity for a in rows) + return not active if event.get("resolution_of") else active def deliver(self, event: dict) -> None: status, _ = self._request("POST", "/api/v2/alerts", [self.payload(event)]) diff --git a/chaski.py b/chaski.py index b4e1d26..e7ef377 100644 --- a/chaski.py +++ b/chaski.py @@ -342,6 +342,14 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: need = [f for f in ("severity", "action", "condition", "comment") if not g.get(f)] if need: raise RuleError(f"emitter {d['label']}: graph sink needs the Reaction fields {need}") + if "resolve_event" in d and (not isinstance(d["resolve_event"], str) + or not d["resolve_event"].strip()): + raise RuleError("resolve_event must be a nonempty event name") + for option in ("generation_event_ids", "missing_is_from"): + if option in d and type(d[option]) is not bool: + raise RuleError(f"{option} must be a boolean") + if d.get("missing_is_from") and not (d.get("resolve_event") and d.get("generation_event_ids")): + raise RuleError("complete-inventory recovery requires resolve_event and generation_event_ids") e = emitter.Emitter(label=d["label"], command=list(d.get("command", [])), interval_s=parse_duration(d["schedule"]), owner=d["owner"], event=d.get("event", "unblocked"), timeout_s=int(d.get("timeout_s", 300)), @@ -349,6 +357,9 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: from_verdict=d.get("from", emitter.BLOCKED), to_verdict=d.get("to", emitter.UNBLOCKED), baseline_emits=bool(d.get("baseline_emits", False)), + resolve_event=d.get("resolve_event"), + generation_event_ids=bool(d.get("generation_event_ids", False)), + missing_is_from=bool(d.get("missing_is_from", False)), change_driven=d.get("trigger", "schedule") == "changes", externally_triggered=d.get("trigger", "schedule") == "external") if d.get("trigger", "schedule") not in {"schedule", "changes", "external"}: diff --git a/emitter.py b/emitter.py index 56f166e..47a9826 100644 --- a/emitter.py +++ b/emitter.py @@ -44,6 +44,7 @@ from __future__ import annotations import json +import hashlib import logging import re import sqlite3 @@ -117,6 +118,9 @@ class Emitter: from_verdict: str = BLOCKED to_verdict: str = UNBLOCKED baseline_emits: bool = False + resolve_event: str | None = None + generation_event_ids: bool = False + missing_is_from: bool = False change_driven: bool = False externally_triggered: bool = False @@ -179,11 +183,22 @@ def observe(conn: sqlite3.Connection, emitter: Emitter, records: list[dict], now they were. """ _validate(emitter, records) + if emitter.missing_is_from and scope is None: + # Opt-in only for complete inventories of a bounded rolling window. + # Failed adapters never reach observe; partial scope never clears absence. + present_keys = {r[emitter.key] for r in records} + records = records + [{emitter.key: item, "verdict": emitter.from_verdict} + for (item,) in conn.execute( + "SELECT item FROM verdicts WHERE emitter=? AND tracked=1", + (emitter.label,)).fetchall() if item not in present_keys] if scope is not None and any(r[emitter.key] not in scope for r in records): raise ProtocolError("adapter record outside its partial scope") queued: list[str] = [] def emit(r: dict, item: str, generation: int) -> None: + if emitter.generation_event_ids: + r = {**r, "event_id": hashlib.sha256( + f"{emitter.label}:{r['event_id']}:{generation}".encode()).hexdigest()} payload = { "event_id": r["event_id"], "event": emitter.event, "emitter": emitter.label, "item": item, "generation": generation, @@ -234,7 +249,7 @@ def emit(r: dict, item: str, generation: int) -> None: # Same verdict, NEW transition: it went away and came back # between two runs (re-blocked and unblocked again; re-verified # and lapsed again). The adapter's id says so; the verdict cannot. - if (verdict == emitter.to_verdict and last_event is not None + if (not emitter.generation_event_ids and verdict == emitter.to_verdict and last_event is not None and r["event_id"] != last_event): generation += 2 conn.execute("UPDATE verdicts SET generation=? WHERE emitter=? AND item=?", @@ -247,6 +262,26 @@ def emit(r: dict, item: str, generation: int) -> None: (verdict, generation, r.get("evidence"), now, emitter.label, item)) if last == emitter.from_verdict and verdict == emitter.to_verdict: emit(r, item, generation) + elif (emitter.resolve_event and last == emitter.to_verdict + and verdict == emitter.from_verdict and last_event): + # Resolve the exact frozen firing identity, never a new label set. + old = conn.execute("SELECT payload FROM outbox WHERE event_id=?", + (last_event,)).fetchone() + if old is None: + raise ProtocolError("resolution has no retained firing payload") + payload = json.loads(old[0]) + resolution_id = hashlib.sha256( + (last_event + ":resolved").encode()).hexdigest() + payload.update(event_id=resolution_id, event=emitter.resolve_event, + resolution_of=last_event, resolved_at=now, + generation=generation) + cur = conn.execute( + "INSERT OR IGNORE INTO outbox" + " (event_id,emitter,item,generation,payload,created) VALUES (?,?,?,?,?,?)", + (resolution_id, emitter.label, item, generation, + json.dumps(payload, sort_keys=True), now)) + if cur.rowcount: + queued.append(resolution_id) for (item,) in conn.execute( "SELECT item FROM verdicts WHERE emitter=? AND tracked=1", (emitter.label,)).fetchall(): if item not in present and (scope is None or item in scope): @@ -332,10 +367,17 @@ def deliver_pending(conn: sqlite3.Connection, sink: Sink, now: float, emitter: s " AND next_attempt <= ? ORDER BY created LIMIT ?", (emitter, now, limit)).fetchall() ok = failed = 0 for event_id, payload, attempts in rows: + event = json.loads(payload) + if event.get("resolution_of"): + prior = conn.execute("SELECT delivered_at FROM outbox WHERE event_id=?", + (event["resolution_of"],)).fetchone() + if prior is None or prior[0] is None: + # A delayed firing must never be delivered AFTER its resolution. + continue if budget is not None and not budget.try_take(now): break # the rest wait for a later slot; nothing is charged to them try: - sink.deliver(json.loads(payload)) + sink.deliver(event) except Exception as exc: # noqa: BLE001 — any failure is a retry failed += 1 backoff = min(MAX_BACKOFF_S, BASE_BACKOFF_S * 2 ** attempts) diff --git a/failure_family_adapter.py b/failure_family_adapter.py new file mode 100644 index 0000000..35ab55d --- /dev/null +++ b/failure_family_adapter.py @@ -0,0 +1,119 @@ +"""Threshold adapter over desire-path's own bounded command-family aggregation. + +Policy is a governed configuration node's single JSON description, read over +Quipu's public query API. It is never guessed from a missing or failed read. +Output includes hashed family identity and numerical evidence only. +""" +import argparse +import datetime as dt +import json +import re +import subprocess +import urllib.request + +FAMILY = re.compile(r"[0-9a-f]{64}\Z") +OWNER = re.compile(r"[a-z][a-z0-9_-]{0,63}\Z") + + +def validate_policy(policy): + if not isinstance(policy, dict): + raise ValueError("policy must be an object") + for key, low, high in (("default_threshold", 1, 10000), ("window_seconds", 60, 86400)): + value = policy.get(key) + if type(value) is not int or not low <= value <= high: + raise ValueError("invalid numerical threshold policy") + overrides = policy.get("family_thresholds", {}) + if not isinstance(overrides, dict) or len(overrides) > 1000: + raise ValueError("invalid threshold overrides") + for family, value in overrides.items(): + if not FAMILY.fullmatch(family) or type(value) is not int or not 1 <= value <= 10000: + raise ValueError("invalid family override") + owners = policy.get("tool_owners", {}) + if not isinstance(owners, dict): + raise ValueError("invalid tool ownership") + for owner in [policy.get("default_owner", ""), *owners.values()]: + if not isinstance(owner, str) or not OWNER.fullmatch(owner) or owner in {"human", "stiwi"}: + raise ValueError("policy requires crew ownership") + return policy + + +def policy_from_graph(server, iri): + if not re.fullmatch(r"https?://[^<>\s\"{}]+", iri): + raise ValueError("invalid governed policy IRI") + # The writer enforces governance; the reader also requires one directly + # asserted ConfigSetting type before consuming its JSON configuration. + namespace = iri.rsplit("/", 1)[0] + "/" + type_query = f'SELECT ?type WHERE {{ <{iri}> a ?type . FILTER(?type = <{namespace}ConfigSetting>) }}' + typed_request = urllib.request.Request(server.rstrip("/") + "/query", + data=json.dumps({"query": type_query}).encode(), headers={ + "Content-Type": "application/json", "X-Quipu-Client": "chaski"}) + with urllib.request.urlopen(typed_request, timeout=15) as response: + typed = json.load(response) + if "error" in typed or typed.get("truncated") or not typed.get("rows"): + raise ValueError("governed policy type UNKNOWN") + query = f'SELECT ?value WHERE {{ <{iri}> ?value }}' + request = urllib.request.Request(server.rstrip("/") + "/query", + data=json.dumps({"query": query}).encode(), headers={ + "Content-Type": "application/json", "X-Quipu-Client": "chaski"}) + with urllib.request.urlopen(request, timeout=15) as response: + data = json.load(response) + if "error" in data or data.get("truncated") or not isinstance(data.get("rows"), list): + raise ValueError("policy query UNKNOWN") + if len(data["rows"]) != 1: + raise ValueError("policy missing or ambiguous") + return validate_policy(json.loads(data["rows"][0]["value"])) + + +def verdicts(report, policy): + validate_policy(policy) + if not isinstance(report, dict) or report.get("window_seconds") != policy["window_seconds"]: + raise ValueError("aggregation window mismatch") + for key in ("unknown", "excluded"): + if type(report.get(key)) is not int or report[key] < 0: + raise ValueError("missing coverage accounting") + families = report.get("families") + if not isinstance(families, list) or len(families) > 10000: + raise ValueError("invalid family inventory") + records, seen = [], set() + for family in families: + identifier = family.get("id") + count = family.get("current_window") + tool = family.get("tool") + if (not isinstance(identifier, str) or not FAMILY.fullmatch(identifier) + or identifier in seen or type(count) is not int or count < 0 + or not isinstance(tool, str)): + raise ValueError("invalid or duplicate family") + seen.add(identifier) + threshold = policy.get("family_thresholds", {}).get(identifier, policy["default_threshold"]) + owner = policy.get("tool_owners", {}).get(tool, policy["default_owner"]) + records.append({"item": identifier, "verdict": "ABOVE" if count >= threshold else "BELOW", + "event_id": identifier, "owner": owner, + "evidence": f"count={count}; threshold={threshold}; window={policy['window_seconds']}s; unsupported_records={report['unknown']}"}) + return records + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--quipu-server", required=True) + parser.add_argument("--policy-iri", required=True) + parser.add_argument("--dp", default="dp") + parser.add_argument("--db", required=True) + args = parser.parse_args() + policy = policy_from_graph(args.quipu_server, args.policy_iri) + end = dt.datetime.now(dt.timezone.utc) + # Historical identities remain visible after their hour expires. Complete + # successful inventories permit the emitter's opt-in missing_is_from; + # any adapter failure never clears an existing firing. + start = end - dt.timedelta(days=7) + result = subprocess.run([args.dp, "--db", args.db, "--json", "failure-families", + "--since", start.isoformat(), "--until", end.isoformat(), + "--window", f"{policy['window_seconds']}s"], capture_output=True, text=True, + timeout=60, check=False) + if result.returncode: + # Do not echo dp stderr: it may contain an operational path or raw input. + raise RuntimeError("desire-path aggregation failed") + print(json.dumps(verdicts(json.loads(result.stdout), policy))) + + +if __name__ == "__main__": + main() diff --git a/test_failure_family_adapter.py b/test_failure_family_adapter.py new file mode 100644 index 0000000..b9a3478 --- /dev/null +++ b/test_failure_family_adapter.py @@ -0,0 +1,30 @@ +import unittest +from failure_family_adapter import verdicts, validate_policy + + +class AdapterTests(unittest.TestCase): + def setUp(self): + self.policy={"default_threshold":5,"window_seconds":3600,"default_owner":"crew","family_thresholds":{},"tool_owners":{"Read":"reader"}} + self.family="a"*64 + + def report(self,count): + return {"window_seconds":3600,"unknown":3,"excluded":2,"families":[{"id":self.family,"tool":"Read","current_window":count}]} + + def test_boundary_and_override(self): + self.assertEqual(verdicts(self.report(4),self.policy)[0]["verdict"],"BELOW") + above=verdicts(self.report(5),self.policy)[0] + self.assertEqual(above["verdict"],"ABOVE") + self.assertEqual(above["owner"],"reader") + self.policy["family_thresholds"][self.family]=10 + self.assertEqual(verdicts(self.report(5),self.policy)[0]["verdict"],"BELOW") + + def test_unknown_policy_and_inventory(self): + for key,value in (("default_owner","human"),("default_threshold",True),("window_seconds",0),("family_thresholds",{self.family:False})): + with self.assertRaises(ValueError):validate_policy({**self.policy,key:value}) + for report in ({}, {**self.report(0),"unknown":None},self.report(-1)): + with self.assertRaises(ValueError):verdicts(report,self.policy) + duplicate=self.report(1);duplicate["families"]*=2 + with self.assertRaises(ValueError):verdicts(duplicate,self.policy) + + +if __name__ == "__main__": unittest.main() diff --git a/test_resolution.py b/test_resolution.py new file mode 100644 index 0000000..e4d847d --- /dev/null +++ b/test_resolution.py @@ -0,0 +1,90 @@ +"""Recovery retains exact alert identity and cannot overtake an unreceived firing.""" +import json +import unittest + +import emitter as em +from alertmanager_sink import AlertmanagerSink + + +class RecoveryTests(unittest.TestCase): + def setUp(self): + self.db = em.connect(":memory:") + self.e = em.Emitter(label="family", command=[], interval_s=60, owner="crew", + from_verdict="BELOW", to_verdict="ABOVE", + resolve_event="recovered") + + def tearDown(self): + self.db.close() + + def observe(self, verdict, now): + r = {"item": "family-1", "verdict": verdict, "event_id": "crossing-1", + "owner": "crew"} + return em.observe(self.db, self.e, [r], now) + + def test_unknown_and_recovery(self): + self.assertEqual(self.observe("BELOW", 100), []) + self.assertEqual(self.observe("ABOVE", 200), ["crossing-1"]) + self.assertEqual(self.observe("ABOVE", 210), []) + self.assertEqual(self.observe("UNKNOWN", 220), []) + resolved = self.observe("BELOW", 300) + self.assertEqual(len(resolved), 1) + self.assertEqual(self.observe("BELOW", 310), []) + firing = json.loads(self.db.execute("select payload from outbox where event_id='crossing-1'").fetchone()[0]) + recovery = json.loads(self.db.execute("select payload from outbox where event_id=?", resolved).fetchone()[0]) + sink = AlertmanagerSink("https://example.invalid", {"alertname":"Failures", "severity":"warning"}, "family", clock=lambda: 400) + self.assertEqual(sink.labels(firing), sink.labels(recovery)) + self.assertEqual(sink.payload(firing)["startsAt"], sink.payload(recovery)["startsAt"]) + self.assertEqual(recovery["resolved_at"], 300) + self.assertLess(sink.payload(recovery)["endsAt"], sink.payload(firing)["endsAt"]) + + def test_resolution_waits_for_firing_receipt(self): + self.observe("BELOW", 100); self.observe("ABOVE", 200); self.observe("BELOW", 300) + class Sink: + seen = [] + fail = True + def deliver(self, event): + if self.fail: raise ConnectionError("unreceived") + self.seen.append(event) + sink=Sink() + self.assertEqual(em.deliver_pending(self.db,sink,400,self.e.label), (0,1)) + self.assertEqual(sink.seen, []) + sink.fail=False + self.assertEqual(em.deliver_pending(self.db,sink,500,self.e.label), (2,0)) + self.assertNotIn("resolution_of",sink.seen[0]) + self.assertEqual(sink.seen[1]["resolution_of"],"crossing-1") + + def test_generation_identity_and_complete_missing(self): + self.e.generation_event_ids = True + self.e.missing_is_from = True + self.observe("BELOW",100) + first=self.observe("ABOVE",200) + self.assertEqual(len(first),1) + self.assertEqual(self.observe("ABOVE",210),[]) + # Partial absence is not a complete zero count. + self.assertEqual(em.observe(self.db,self.e,[],220,scope=set()),[]) + recovered=em.observe(self.db,self.e,[],300) + self.assertEqual(len(recovered),1) + second=self.observe("ABOVE",400) + self.assertEqual(len(second),1) + self.assertNotEqual(first,second) + + def test_resolution_receipt_requires_valid_readback(self): + self.observe("BELOW",100); self.observe("ABOVE",200) + resolved=self.observe("BELOW",300) + event=json.loads(self.db.execute("select payload from outbox where event_id=?",resolved).fetchone()[0]) + sink=AlertmanagerSink("https://example.invalid",{"alertname":"Failures","severity":"warning"},"family") + requests=[] + def request(method,path,body=None): + requests.append((method,path,body)) + return (200,b"[]") + sink._request=request + sink.deliver(event) + self.assertEqual(requests[0][2][0]["labels"]["event_id"],"crossing-1") + self.assertIn("crossing-1",requests[1][1]) + sink._request=lambda method,path,body=None:(200,b'[{"labels":{"event_id":"crossing-1"}}]') + with self.assertRaises(ConnectionError):sink.deliver(event) + sink._request=lambda method,path,body=None:(200,b'{}') + with self.assertRaises(ConnectionError):sink.deliver(event) + + +if __name__ == "__main__": unittest.main() From d6f34ed4fc6fa270f73e623b918d1c4c0c916e46 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sun, 11 Oct 2026 18:16:15 -0400 Subject: [PATCH 2/5] feat: run scheduled failure-family alerts with retained leases --- alertmanager_sink.py | 13 +++ chaski.py | 13 ++- docs/local-failure-families.md | 28 +++++++ emitter.py | 120 +++++++++++++++++++++++++- failure_family_adapter.py | 82 ++++++++++++++---- graph_sink.py | 20 +++++ once.py | 148 +++++++++++++++++++++++++++++++++ test_failure_family_adapter.py | 89 ++++++++++++++++---- test_once.py | 76 +++++++++++++++++ test_renewal.py | 123 +++++++++++++++++++++++++++ test_resolution.py | 100 +++++++++++++--------- 11 files changed, 732 insertions(+), 80 deletions(-) create mode 100644 docs/local-failure-families.md create mode 100644 once.py create mode 100644 test_once.py create mode 100644 test_renewal.py diff --git a/alertmanager_sink.py b/alertmanager_sink.py index 1e06212..f6619b6 100644 --- a/alertmanager_sink.py +++ b/alertmanager_sink.py @@ -135,6 +135,12 @@ def deliver(self, event: dict) -> None: if not self._received(event): raise ConnectionError(f"{event['event_id']}: accepted but not active on read-back") + def renew(self, event: dict) -> None: + """Receipt-backed same-label lease refresh; no new domain event.""" + if event.get("resolution_of"): + raise ValueError("cannot renew recovery") + self.deliver(event) + class _Missing(dict): """str.format_map that leaves an unknown field visible instead of raising.""" @@ -165,3 +171,10 @@ def setup(self) -> None: def deliver(self, event: dict) -> None: for s in self.sinks: s.deliver(event) + + def renew(self, event: dict) -> None: + recipients = [s for s in self.sinks if callable(getattr(s, "renew", None))] + if not recipients: + raise ValueError("no lease-aware recipient") + for sink in recipients: + sink.renew(event) diff --git a/chaski.py b/chaski.py index e7ef377..d6aed02 100644 --- a/chaski.py +++ b/chaski.py @@ -350,6 +350,16 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: raise RuleError(f"{option} must be a boolean") if d.get("missing_is_from") and not (d.get("resolve_event") and d.get("generation_event_ids")): raise RuleError("complete-inventory recovery requires resolve_event and generation_event_ids") + renew = d.get("renew_interval_s", 0) + if type(renew) is not int or renew < 0: + raise RuleError("renew_interval_s must be a nonnegative integer") + if renew and (d.get("trigger", "schedule") != "schedule" or "alertmanager" not in kinds + or not d.get("resolve_event") or not d.get("generation_event_ids")): + raise RuleError("renewal requires a scheduled generation-bound recovering alert emitter") + if renew: + ttl = d["sink"]["alertmanager"].get("ttl_s", 86400) + if type(ttl) is not int or not parse_duration(d["schedule"]) <= renew < ttl: + raise RuleError("renewal interval must fit schedule and finite alert TTL") e = emitter.Emitter(label=d["label"], command=list(d.get("command", [])), interval_s=parse_duration(d["schedule"]), owner=d["owner"], event=d.get("event", "unblocked"), timeout_s=int(d.get("timeout_s", 300)), @@ -361,7 +371,8 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: generation_event_ids=bool(d.get("generation_event_ids", False)), missing_is_from=bool(d.get("missing_is_from", False)), change_driven=d.get("trigger", "schedule") == "changes", - externally_triggered=d.get("trigger", "schedule") == "external") + externally_triggered=d.get("trigger", "schedule") == "external", + renew_interval_s=renew) if d.get("trigger", "schedule") not in {"schedule", "changes", "external"}: raise RuleError("emitter trigger must be schedule, changes or external") if e.externally_triggered and d.get("baseline_emits") is not True: diff --git a/docs/local-failure-families.md b/docs/local-failure-families.md new file mode 100644 index 0000000..b09593c --- /dev/null +++ b/docs/local-failure-families.md @@ -0,0 +1,28 @@ +# Scheduled local failure-family collector + +The adapter reads a directly typed ConfigSetting with one JSON threshold and +crew-ownership policy. Missing or ambiguous policy is UNKNOWN. Use a desire-path +release whose local failure-families command opens existing databases read-only. +The adapter isolates its child HOME so operator remote-store configuration cannot +redirect the explicit local measurement. It never copies the database or exports +raw commands, arguments or error messages. + +The once.py entry runs one scheduled emitter tick without a listener. Its state +and metrics parents must be provisioned beforehand. A process lock prevents +overlap; a persistent write budget retains its thirty-second floor across +restarts. Adapter counters and graph setup receipts also persist. Setup reuse +requires directly typed read-back; changed metadata, retraction or graph-delivery +failure requires repair. Startup failure publishes UNKNOWN when the metrics path +is writable. + +Opt-in renewal requires a scheduled Alertmanager emitter with generation IDs and +recovery configured. Set renew_interval_s at least as high as the schedule +interval and strictly below the alert TTL. Each renewal requires fresh positive +source evidence and a received firing, and sends the original frozen alert only +to the lease-aware sink. It creates no additional graph, command or outbox event. +UNKNOWN, missing, recovered and old generations cannot renew. Failed renewals +retain backoff across restarts. Lease expiry is not evidence of recovery. + +Monitor both measurement status and freshness, along with pending outbox and +renewal failures. Source measurement success does not prove receiver delivery. +This source installs no schedule, changes no routing and authorizes no test page. diff --git a/emitter.py b/emitter.py index 47a9826..cbb8794 100644 --- a/emitter.py +++ b/emitter.py @@ -66,6 +66,24 @@ DELIVER_PER_TICK = 5 SCHEMA = """ +CREATE TABLE IF NOT EXISTS renewal_receipts ( + event_id TEXT PRIMARY KEY, + last_received REAL, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt REAL NOT NULL DEFAULT 0, + last_error TEXT +); +CREATE TABLE IF NOT EXISTS runner_status ( + emitter TEXT PRIMARY KEY, + last_attempt REAL NOT NULL, + last_success REAL NOT NULL, + ok INTEGER NOT NULL, + unknown INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS delivery_budget ( + scope TEXT PRIMARY KEY, + last_taken REAL NOT NULL +); CREATE TABLE IF NOT EXISTS verdicts ( emitter TEXT NOT NULL, item TEXT NOT NULL, @@ -123,6 +141,8 @@ class Emitter: missing_is_from: bool = False change_driven: bool = False externally_triggered: bool = False + renew_interval_s: int = 0 + persist_runs: bool = False @property def verdicts(self) -> set[str]: @@ -394,6 +414,75 @@ def deliver_pending(conn: sqlite3.Connection, sink: Sink, now: float, emitter: s return ok, failed +def renew_active(conn: sqlite3.Connection, sink: Sink, now: float, definition: Emitter, + positive: set[str], budget: WriteBudget | None = None, + limit: int = DELIVER_PER_TICK) -> tuple[int, int]: + """Refresh only received firings with fresh positive source evidence. + + Renewal sends the original payload only to a lease-aware receiver. It creates + no transition/outbox/graph event. Missing, UNKNOWN, recovery, pending firing + and old generations cannot authorize a refresh. + """ + if not definition.renew_interval_s or not positive: + return 0, 0 + renewal = getattr(sink, "renew", None) + if not callable(renewal): + raise ProtocolError("renewal requires a lease-aware sink") + rows = conn.execute( + "SELECT v.item, o.event_id, o.payload, COALESCE(r.attempts,0) " + "FROM verdicts v JOIN outbox o ON o.event_id=v.last_event " + "LEFT JOIN renewal_receipts r ON r.event_id=o.event_id " + "WHERE v.emitter=? AND v.tracked=1 AND v.verdict=? " + "AND v.item IN (SELECT value FROM json_each(?)) " + "AND o.delivered_at IS NOT NULL AND COALESCE(r.next_attempt,0)<=? " + "AND COALESCE(r.last_received,o.delivered_at)<=? " + "ORDER BY COALESCE(r.last_received,o.delivered_at) LIMIT ?", + (definition.label, definition.to_verdict, json.dumps(sorted(positive)), now, + now-definition.renew_interval_s, limit)).fetchall() + ok = failed = 0 + for item, identity, payload, attempts in rows: + if item not in positive: + continue + event = json.loads(payload) + if event.get("resolution_of"): + raise ProtocolError("active generation points to recovery") + if budget is not None and not budget.try_take(now): + break + try: + renewal(event) + except Exception as exc: + failed += 1 + backoff = min(MAX_BACKOFF_S, BASE_BACKOFF_S * 2 ** min(attempts, 20)) + conn.execute( + "INSERT INTO renewal_receipts(event_id,attempts,next_attempt,last_error) " + "VALUES(?,1,?,?) ON CONFLICT(event_id) DO UPDATE SET " + "attempts=attempts+1,next_attempt=excluded.next_attempt,last_error=excluded.last_error", + (identity, now+backoff, str(exc)[:300])) + continue + conn.execute( + "INSERT INTO renewal_receipts(event_id,last_received,attempts,next_attempt) " + "VALUES(?,?,1,0) ON CONFLICT(event_id) DO UPDATE SET " + "last_received=excluded.last_received,attempts=attempts+1,next_attempt=0,last_error=NULL", + (identity, now)) + ok += 1 + return ok, failed + + +class PersistentWriteBudget(WriteBudget): + """A local one-shot keeps the same global sink budget across restarts.""" + def __init__(self, conn: sqlite3.Connection, interval_s: float): + super().__init__(interval_s) + self.conn = conn + + def try_take(self, now: float) -> bool: + row = self.conn.execute( + "INSERT INTO delivery_budget(scope,last_taken) VALUES('global',?) " + "ON CONFLICT(scope) DO UPDATE SET last_taken=excluded.last_taken " + "WHERE delivery_budget.last_taken<=excluded.last_taken-?", + (now, self.interval_s)) + return row.rowcount == 1 + + class Runner: """Schedules one emitter's adapter runs and deliveries, and counts them. @@ -405,15 +494,19 @@ class Runner: def __init__(self, emitter: Emitter, conn: sqlite3.Connection, sink: Sink, budget: WriteBudget | None = None): self.emitter, self.conn, self.sink, self.budget = emitter, conn, sink, budget - self.last_attempt = 0.0 - self.last_success = 0.0 - self.runs = {"ok": 0, "unknown": 0} + saved = (conn.execute("SELECT last_attempt,last_success,ok,unknown FROM runner_status " + "WHERE emitter=?", (emitter.label,)).fetchone() + if emitter.persist_runs else None) + self.last_attempt = saved[0] if saved else 0.0 + self.last_success = saved[1] if saved else 0.0 + self.runs = {"ok": saved[2] if saved else 0, "unknown": saved[3] if saved else 0} self.queued_total = 0 self._snap_lock = threading.Lock() self._snapshot: dict = {} self._publish() def tick(self, now: float) -> None: + positive = set() if not self.emitter.externally_triggered and now - self.last_attempt >= self.emitter.interval_s: self.last_attempt = now try: @@ -421,10 +514,20 @@ def tick(self, now: float) -> None: self.queued_total += len(observe(self.conn, self.emitter, records, now)) self.runs["ok"] += 1 self.last_success = now + positive = {r[self.emitter.key] for r in records + if r["verdict"] == self.emitter.to_verdict} except Exception as exc: # noqa: BLE001 — a failed run is UNKNOWN for every item self.runs["unknown"] += 1 LOG.warning("emitter %s UNKNOWN (no verdict changed): %s", self.emitter.label, exc) deliver_pending(self.conn, self.sink, now, self.emitter.label, self.budget, limit=DELIVER_PER_TICK) + renew_active(self.conn, self.sink, now, self.emitter, positive, self.budget) + if self.emitter.persist_runs: + self.conn.execute( + "INSERT INTO runner_status VALUES(?,?,?,?,?) ON CONFLICT(emitter) DO UPDATE SET " + "last_attempt=excluded.last_attempt,last_success=excluded.last_success," + "ok=excluded.ok,unknown=excluded.unknown", + (self.emitter.label,self.last_attempt,self.last_success, + self.runs["ok"],self.runs["unknown"])) self._publish() def _publish(self) -> None: @@ -434,7 +537,14 @@ def _publish(self) -> None: delivered = self.conn.execute( "SELECT COUNT(*) FROM outbox WHERE emitter=? AND delivered_at IS NOT NULL", (self.emitter.label,)).fetchone()[0] - snap = {"runs": dict(self.runs), "last_success": self.last_success, + renewal_pending, renewal_failures = self.conn.execute( + "SELECT COUNT(r.last_received),COALESCE(SUM(CASE WHEN r.last_error IS NOT NULL THEN 1 ELSE 0 END),0) " + "FROM renewal_receipts r JOIN outbox o ON o.event_id=r.event_id " + "JOIN verdicts v ON v.last_event=o.event_id " + "WHERE v.emitter=? AND v.tracked=1 AND v.verdict=?", + (self.emitter.label,self.emitter.to_verdict)).fetchone() + snap = {"renewal_receipts": renewal_pending, "renewal_failures": renewal_failures, + "runs": dict(self.runs), "last_success": self.last_success, "queued_total": self.queued_total, "pending": pending, "oldest_created": oldest, "delivered": delivered} with self._snap_lock: @@ -449,6 +559,8 @@ def metrics(self, esc) -> list[str]: lines += [ f'chaski_emitter_last_success_timestamp_seconds{{emitter="{label}"}} {s["last_success"]:.0f}', f'chaski_emitter_events_queued_total{{emitter="{label}"}} {s["queued_total"]}', + f'chaski_emitter_renewal_failures{{emitter="{label}"}} {s["renewal_failures"]}', + f'chaski_emitter_renewal_receipts{{emitter="{label}"}} {s["renewal_receipts"]}', f'chaski_emitter_outbox_pending{{emitter="{label}"}} {s["pending"]}', f'chaski_emitter_outbox_delivered_total{{emitter="{label}"}} {s["delivered"]}', f'chaski_emitter_outbox_oldest_pending_age_seconds{{emitter="{label}"}} ' diff --git a/failure_family_adapter.py b/failure_family_adapter.py index 35ab55d..e0812ea 100644 --- a/failure_family_adapter.py +++ b/failure_family_adapter.py @@ -4,9 +4,14 @@ Quipu's public query API. It is never guessed from a missing or failed read. Output includes hashed family identity and numerical evidence only. """ + import argparse import datetime as dt import json +import os +from pathlib import Path +import shutil +import tempfile import re import subprocess import urllib.request @@ -43,18 +48,22 @@ def policy_from_graph(server, iri): # The writer enforces governance; the reader also requires one directly # asserted ConfigSetting type before consuming its JSON configuration. namespace = iri.rsplit("/", 1)[0] + "/" - type_query = f'SELECT ?type WHERE {{ <{iri}> a ?type . FILTER(?type = <{namespace}ConfigSetting>) }}' - typed_request = urllib.request.Request(server.rstrip("/") + "/query", - data=json.dumps({"query": type_query}).encode(), headers={ - "Content-Type": "application/json", "X-Quipu-Client": "chaski"}) + type_query = f"SELECT ?type WHERE {{ <{iri}> a ?type . FILTER(?type = <{namespace}ConfigSetting>) }}" + typed_request = urllib.request.Request( + server.rstrip("/") + "/query", + data=json.dumps({"query": type_query}).encode(), + headers={"Content-Type": "application/json", "X-Quipu-Client": "chaski"}, + ) with urllib.request.urlopen(typed_request, timeout=15) as response: typed = json.load(response) if "error" in typed or typed.get("truncated") or not typed.get("rows"): raise ValueError("governed policy type UNKNOWN") - query = f'SELECT ?value WHERE {{ <{iri}> ?value }}' - request = urllib.request.Request(server.rstrip("/") + "/query", - data=json.dumps({"query": query}).encode(), headers={ - "Content-Type": "application/json", "X-Quipu-Client": "chaski"}) + query = f"SELECT ?value WHERE {{ <{iri}> ?value }}" + request = urllib.request.Request( + server.rstrip("/") + "/query", + data=json.dumps({"query": query}).encode(), + headers={"Content-Type": "application/json", "X-Quipu-Client": "chaski"}, + ) with urllib.request.urlopen(request, timeout=15) as response: data = json.load(response) if "error" in data or data.get("truncated") or not isinstance(data.get("rows"), list): @@ -79,16 +88,29 @@ def verdicts(report, policy): identifier = family.get("id") count = family.get("current_window") tool = family.get("tool") - if (not isinstance(identifier, str) or not FAMILY.fullmatch(identifier) - or identifier in seen or type(count) is not int or count < 0 - or not isinstance(tool, str)): + if ( + not isinstance(identifier, str) + or not FAMILY.fullmatch(identifier) + or identifier in seen + or type(count) is not int + or count < 0 + or not isinstance(tool, str) + ): raise ValueError("invalid or duplicate family") seen.add(identifier) threshold = policy.get("family_thresholds", {}).get(identifier, policy["default_threshold"]) owner = policy.get("tool_owners", {}).get(tool, policy["default_owner"]) - records.append({"item": identifier, "verdict": "ABOVE" if count >= threshold else "BELOW", - "event_id": identifier, "owner": owner, - "evidence": f"count={count}; threshold={threshold}; window={policy['window_seconds']}s; unsupported_records={report['unknown']}"}) + records.append( + { + "item": identifier, + "verdict": "ABOVE" if count >= threshold else "BELOW", + "event_id": identifier, + "owner": owner, + "evidence": (f"count={count}; threshold={threshold}; " + f"window={policy['window_seconds']}s; " + f"unsupported_records={report['unknown']}"), + } + ) return records @@ -105,10 +127,34 @@ def main(): # successful inventories permit the emitter's opt-in missing_is_from; # any adapter failure never clears an existing firing. start = end - dt.timedelta(days=7) - result = subprocess.run([args.dp, "--db", args.db, "--json", "failure-families", - "--since", start.isoformat(), "--until", end.isoformat(), - "--window", f"{policy['window_seconds']}s"], capture_output=True, text=True, - timeout=60, check=False) + # Explicit paths survive a clean child HOME. Operator remote-store config + # must not silently redirect this local, read-only measurement. + binary = shutil.which(args.dp) + if binary is None: + raise RuntimeError("desire-path executable unavailable") + database = str(Path(args.db).resolve()) + with tempfile.TemporaryDirectory(prefix="failure-family-reader-") as home: + child_env = dict(os.environ, HOME=home, XDG_CONFIG_HOME=home) + result = subprocess.run( + [ + str(Path(binary).resolve()), + "--db", + database, + "--json", + "failure-families", + "--since", + start.isoformat(), + "--until", + end.isoformat(), + "--window", + f"{policy['window_seconds']}s", + ], + env=child_env, + capture_output=True, + text=True, + timeout=60, + check=False, + ) if result.returncode: # Do not echo dp stderr: it may contain an operational path or raw input. raise RuntimeError("desire-path aggregation failed") diff --git a/graph_sink.py b/graph_sink.py index c8c586b..4f6aac9 100644 --- a/graph_sink.py +++ b/graph_sink.py @@ -147,6 +147,26 @@ def _received(self, event: dict) -> bool: raise ConnectionError(f"read-back returned no rows field: {str(out)[:200]}") return any(str(r.get("e")) == event["event_id"] for r in rows) + def setup_present(self) -> bool: + """Direct type read-back for a one-shot's retained setup receipt.""" + query = (f"SELECT ?type WHERE {{ <{self.reaction_iri()}> a ?type . " + f"FILTER(?type = <{self.ns}Reaction>) }}") + req = urllib.request.Request( + self.base + "/query", data=json.dumps({"query": query}).encode(), + headers={"Content-Type": "application/json", "User-Agent": USER_AGENT, + "X-Quipu-Client": CLIENT_LABEL}) + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + out = json.load(resp) + rows = out.get("rows") + if out.get("error") or out.get("truncated") or not isinstance(rows, list): + raise ConnectionError("setup read-back UNKNOWN") + expected = {self.ns + "Reaction"} + if self.ns == NS: + expected.add("aegis:Reaction") + if any(not isinstance(row, dict) or row.get("type") not in expected for row in rows): + raise ConnectionError("setup read-back has unexpected type") + return bool(rows) + def setup_pending(self) -> bool: """The Reaction every firing points at is owed once per process. The emitter spends a write-budget slot on it, separately from any event.""" diff --git a/once.py b/once.py new file mode 100644 index 0000000..3a3243f --- /dev/null +++ b/once.py @@ -0,0 +1,148 @@ +"""One bounded scheduled emitter tick; no HTTP listener or graph event poll.""" + +from __future__ import annotations + +import argparse +import fcntl +import hashlib +import re +import os +from pathlib import Path +import tempfile +import time + +import chaski +import emitter + + +class RetainedSetup: + """Reuse an exact setup receipt only while the owned receiver is typed. + + Wrapping only the graph child keeps Alertmanager failures from erasing a + valid graph setup. A graph delivery failure drops the local cache; retry + repairs its metadata under the existing idempotent snapshot key. + """ + + def __init__(self, conn, receiver): + self.conn, self.receiver = conn, receiver + self.signature = hashlib.sha256(receiver.reaction_turtle().encode()).hexdigest() + self.key = receiver.reaction_iri() + conn.execute("CREATE TABLE IF NOT EXISTS receiver_setup (receiver TEXT PRIMARY KEY, signature TEXT NOT NULL)") + + def setup_pending(self): + row = self.conn.execute("SELECT signature FROM receiver_setup WHERE receiver=?", (self.key,)).fetchone() + if row and row[0] == self.signature and self.receiver.setup_present(): + self.receiver._reaction_written = True + return False + return True + + def setup(self): + self.receiver.setup() + if not self.receiver.setup_present(): + raise ConnectionError("setup accepted but not typed on read-back") + self.conn.execute( + "INSERT INTO receiver_setup VALUES(?,?) ON CONFLICT(receiver) DO UPDATE SET signature=excluded.signature", + (self.key, self.signature), + ) + + def deliver(self, event): + try: + self.receiver.deliver(event) + except Exception: + self.conn.execute("DELETE FROM receiver_setup WHERE receiver=?", (self.key,)) + raise + + +def retain_setup(conn, sink): + import alertmanager_sink + import graph_sink + + if isinstance(sink, graph_sink.QuipuFiringSink): + return RetainedSetup(conn, sink) + if isinstance(sink, alertmanager_sink.ChainSink): + return alertmanager_sink.ChainSink([retain_setup(conn, child) for child in sink.sinks]) + return sink + + +def write_metrics(path: Path, lines: list[str]) -> None: + with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as tmp: + temp = Path(tmp.name) + tmp.write("\n".join(lines) + "\n") + try: + temp.chmod(0o644) + temp.replace(path) + finally: + temp.unlink(missing_ok=True) + + +def publish(path: Path, runner: emitter.Runner) -> None: + # The regular exporter can scrape these fixed-emitter metrics. No family, + # command, error or argument is a metric label. + lines = runner.metrics(lambda value: value.replace("\\", "\\\\").replace('"', '\\"')) + label = runner.emitter.label + measured = int(runner.last_success > 0 and runner.last_success >= runner.last_attempt) + lines += [ + f'chaski_local_measurement_ok{{emitter="{label}"}} {measured}', + f'chaski_local_last_attempt_seconds{{emitter="{label}"}} {runner.last_attempt}', + ] + write_metrics(path, lines) + + +def run_once(config: Path, label: str, state: Path, quipu: str, metrics: Path, write_interval: int = 30) -> int: + if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]{0,127}", label): + raise ValueError("invalid one-shot emitter label") + definitions = [(e, sink) for e, sink in chaski.load_emitters(config) if e.label == label] + if len(definitions) != 1: + raise ValueError("select exactly one configured emitter") + definition, sink_config = definitions[0] + definition.persist_runs = True + if definition.change_driven or definition.externally_triggered: + raise ValueError("one-shot requires a scheduled adapter") + if write_interval < 30: + raise ValueError("one-shot sink budget has a thirty-second floor") + # Lock before opening/migrating state or contacting any sink. The unit's + # state and metrics parents are provisioned by the reviewed installation. + with state.with_suffix(state.suffix + ".lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + conn = emitter.connect(state) + try: + sink = retain_setup(conn, chaski.make_sink(definition, sink_config, quipu)) + runner = emitter.Runner(definition, conn, sink, emitter.PersistentWriteBudget(conn, write_interval)) + runner.tick(time.time()) + publish(metrics, runner) + # A successful empty adapter is measurable; failure remains UNKNOWN. + return 0 if runner.last_success > 0 and runner.last_success >= runner.last_attempt else 2 + finally: + conn.close() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--emitters", type=Path, required=True) + parser.add_argument("--label", required=True) + parser.add_argument("--state", type=Path, required=True) + parser.add_argument("--quipu", required=True) + parser.add_argument("--metrics", type=Path, required=True) + parser.add_argument("--write-interval", type=int, default=30) + args = parser.parse_args() + # Supply bounded error classes rather than exception text containing raw + # adapter output, operational paths or credential-file names. + try: + return run_once(args.emitters, args.label, args.state, args.quipu, args.metrics, args.write_interval) + except Exception as exc: + # A concurrent holder owns its measurement. Other startup failures must + # not leave a previous successful status looking like a fresh result. + if not isinstance(exc, BlockingIOError) and re.fullmatch( + r"[A-Za-z][A-Za-z0-9_.-]{0,127}", args.label): + try: + write_metrics(args.metrics, [ + f'chaski_local_measurement_ok{{emitter="{args.label}"}} 0', + f'chaski_local_last_attempt_seconds{{emitter="{args.label}"}} {time.time()}']) + except OSError: + pass # Missing metrics is caught by the separate scrape/liveness rule. + print(f"UNKNOWN one-shot: {type(exc).__name__}", file=os.sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test_failure_family_adapter.py b/test_failure_family_adapter.py index b9a3478..4acff3b 100644 --- a/test_failure_family_adapter.py +++ b/test_failure_family_adapter.py @@ -4,27 +4,82 @@ class AdapterTests(unittest.TestCase): def setUp(self): - self.policy={"default_threshold":5,"window_seconds":3600,"default_owner":"crew","family_thresholds":{},"tool_owners":{"Read":"reader"}} - self.family="a"*64 + self.policy = { + "default_threshold": 5, + "window_seconds": 3600, + "default_owner": "crew", + "family_thresholds": {}, + "tool_owners": {"Read": "reader"}, + } + self.family = "a" * 64 - def report(self,count): - return {"window_seconds":3600,"unknown":3,"excluded":2,"families":[{"id":self.family,"tool":"Read","current_window":count}]} + def report(self, count): + return { + "window_seconds": 3600, + "unknown": 3, + "excluded": 2, + "families": [{"id": self.family, "tool": "Read", "current_window": count}], + } def test_boundary_and_override(self): - self.assertEqual(verdicts(self.report(4),self.policy)[0]["verdict"],"BELOW") - above=verdicts(self.report(5),self.policy)[0] - self.assertEqual(above["verdict"],"ABOVE") - self.assertEqual(above["owner"],"reader") - self.policy["family_thresholds"][self.family]=10 - self.assertEqual(verdicts(self.report(5),self.policy)[0]["verdict"],"BELOW") + self.assertEqual(verdicts(self.report(4), self.policy)[0]["verdict"], "BELOW") + above = verdicts(self.report(5), self.policy)[0] + self.assertEqual(above["verdict"], "ABOVE") + self.assertEqual(above["owner"], "reader") + self.policy["family_thresholds"][self.family] = 10 + self.assertEqual(verdicts(self.report(5), self.policy)[0]["verdict"], "BELOW") def test_unknown_policy_and_inventory(self): - for key,value in (("default_owner","human"),("default_threshold",True),("window_seconds",0),("family_thresholds",{self.family:False})): - with self.assertRaises(ValueError):validate_policy({**self.policy,key:value}) - for report in ({}, {**self.report(0),"unknown":None},self.report(-1)): - with self.assertRaises(ValueError):verdicts(report,self.policy) - duplicate=self.report(1);duplicate["families"]*=2 - with self.assertRaises(ValueError):verdicts(duplicate,self.policy) + for key, value in ( + ("default_owner", "human"), + ("default_threshold", True), + ("window_seconds", 0), + ("family_thresholds", {self.family: False}), + ): + with self.assertRaises(ValueError): + validate_policy({**self.policy, key: value}) + for report in ({}, {**self.report(0), "unknown": None}, self.report(-1)): + with self.assertRaises(ValueError): + verdicts(report, self.policy) + duplicate = self.report(1) + duplicate["families"] *= 2 + with self.assertRaises(ValueError): + verdicts(duplicate, self.policy) -if __name__ == "__main__": unittest.main() +if __name__ == "__main__": + unittest.main() + + +def test_local_reader_ignores_operator_remote_config(tmp_path, monkeypatch, capsys): + import json + from pathlib import Path + from types import SimpleNamespace + import failure_family_adapter as adapter + + operator = tmp_path / "operator" + (operator / ".dp").mkdir(parents=True) + config = operator / ".dp" / "config.toml" + config.write_text('store_mode = "remote"\nremote_url = "https://example.invalid"\n') + monkeypatch.setenv("HOME", str(operator)) + policy = {"default_threshold": 5, "window_seconds": 3600, "default_owner": "crew"} + monkeypatch.setattr(adapter, "policy_from_graph", lambda *args: policy) + monkeypatch.setattr(adapter.shutil, "which", lambda value: "/usr/bin/true") + seen = [] + def child(command, **kwargs): + home = Path(kwargs["env"]["HOME"]) + assert home != operator + assert not (home / ".dp" / "config.toml").exists() + assert Path(command[2]).is_absolute() + seen.append(home) + return SimpleNamespace(returncode=0, stdout=json.dumps( + {"window_seconds": 3600, "unknown": 0, "excluded": 0, "families": []})) + monkeypatch.setattr(adapter.subprocess, "run", child) + monkeypatch.setattr(adapter.os.sys, "argv", + ["adapter", "--quipu-server", "https://example.invalid", + "--policy-iri", "https://example.invalid/policy", + "--db", str(tmp_path / "existing.db")]) + adapter.main() + assert capsys.readouterr().out.strip() == "[]" + assert not seen[0].exists() + assert config.read_text().startswith('store_mode = "remote"') diff --git a/test_once.py b/test_once.py new file mode 100644 index 0000000..865a64c --- /dev/null +++ b/test_once.py @@ -0,0 +1,76 @@ +"""Restart and retained-setup controls for the bounded scheduled entry.""" + +import pytest + +import emitter +import once + + +class Receiver: + def __init__(self): + self.present = False + self.writes = 0 + self.content = "metadata" + self._reaction_written = False + + def reaction_turtle(self): + return self.content + + def reaction_iri(self): + return "https://example.invalid/reaction" + + def setup_present(self): + return self.present + + def setup(self): + self.writes += 1 + self.present = True + + def deliver(self, event): + raise ConnectionError("controlled failure") + + +def test_setup_restart_reuses_receipt_but_retraction_and_change_repair(): + conn = emitter.connect(":memory:") + receiver = Receiver() + first = once.RetainedSetup(conn, receiver) + assert first.setup_pending() + first.setup() + restarted = once.RetainedSetup(conn, receiver) + assert not restarted.setup_pending() + assert receiver.writes == 1 + receiver.present = False + assert restarted.setup_pending() + restarted.setup() + receiver.content = "changed metadata" + changed = once.RetainedSetup(conn, receiver) + assert changed.setup_pending() + changed.setup() + assert receiver.writes == 3 + + +def test_unverified_setup_and_failed_graph_delivery_drop_cache(): + conn = emitter.connect(":memory:") + receiver = Receiver() + wrapper = once.RetainedSetup(conn, receiver) + wrapper.setup() + with pytest.raises(ConnectionError): + wrapper.deliver({}) + assert conn.execute("SELECT COUNT(*) FROM receiver_setup").fetchone()[0] == 0 + receiver.setup = lambda: None + receiver.present = False + with pytest.raises(ConnectionError): + wrapper.setup() + assert conn.execute("SELECT COUNT(*) FROM receiver_setup").fetchone()[0] == 0 + + +def test_startup_failure_replaces_previous_green_with_unknown(tmp_path, monkeypatch): + metrics = tmp_path / "metrics.prom" + metrics.write_text('chaski_local_measurement_ok{emitter="local"} 1\n') + monkeypatch.setattr(once.os.sys, "argv", + ["once", "--emitters", str(tmp_path / "absent.yaml"), + "--label", "local", "--state", str(tmp_path / "state.db"), + "--quipu", "https://example.invalid", + "--metrics", str(metrics)]) + assert once.main() == 2 + assert 'chaski_local_measurement_ok{emitter="local"} 0' in metrics.read_text() diff --git a/test_renewal.py b/test_renewal.py new file mode 100644 index 0000000..931f4e6 --- /dev/null +++ b/test_renewal.py @@ -0,0 +1,123 @@ +"""Lease refresh must require fresh evidence and preserve event identity.""" + +from unittest.mock import patch + +import emitter + + +class Sink: + def __init__(self): + self.deliveries = [] + self.renewals = [] + self.fail = False + + def deliver(self, event): + self.deliveries.append(dict(event)) + + def renew(self, event): + if self.fail: + raise ConnectionError("controlled failure") + self.renewals.append(dict(event)) + + +def setup(): + definition = emitter.Emitter( + label="lease", + command=[], + interval_s=5, + owner="crew", + from_verdict="BELOW", + to_verdict="ABOVE", + resolve_event="recovered", + generation_event_ids=True, + renew_interval_s=30, + persist_runs=True, + ) + conn = emitter.connect(":memory:") + sink = Sink() + return definition, conn, sink, emitter.Runner(definition, conn, sink) + + +def tick(runner, now, verdict=None, fail=False): + records = [] if verdict is None else [{"item": "a", "event_id": "source-a", "verdict": verdict, "owner": "crew"}] + with patch.object( + emitter, "run_adapter", side_effect=ConnectionError("unknown") if fail else None, return_value=records + ): + runner.tick(now) + + +def test_refresh_same_identity_without_new_transition(): + _, conn, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + tick(runner, 100, "ABOVE") + assert sink.renewals == sink.deliveries + assert conn.execute("SELECT COUNT(*) FROM outbox").fetchone()[0] == 1 + + +def test_unknown_missing_and_skipped_measurements_never_renew(): + _, _, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + tick(runner, 100, "UNKNOWN") + tick(runner, 200, fail=True) + tick(runner, 201, "ABOVE") # interval prevents a new measurement + tick(runner, 300) + assert sink.renewals == [] + + +def test_recovery_and_new_generation_cannot_refresh_old_identity(): + _, _, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + old = sink.deliveries[0]["event_id"] + tick(runner, 100, "BELOW") + assert sink.deliveries[1]["resolution_of"] == old + tick(runner, 200, "ABOVE") + new = sink.deliveries[2]["event_id"] + assert new != old + tick(runner, 300, "ABOVE") + assert [r["event_id"] for r in sink.renewals] == [new] + + +def test_pending_firing_cannot_be_renewed(): + definition, conn, sink, _ = setup() + emitter.observe(conn, definition, [{"item": "a", "verdict": "BELOW"}], 10) + emitter.observe(conn, definition, [{"item": "a", "verdict": "ABOVE", "event_id": "source-a"}], 20) + assert emitter.renew_active(conn, sink, 100, definition, {"a"}) == (0, 0) + assert sink.renewals == [] + + +def test_restart_retains_unknown_and_renewal_backoff(): + definition, conn, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + sink.fail = True + tick(runner, 100, "ABOVE") + tick(runner, 110, fail=True) + restarted = emitter.Runner(definition, conn, sink) + assert restarted.runs == {"ok": 3, "unknown": 1} + sink.fail = False + tick(restarted, 120, "ABOVE") + assert sink.renewals == [] + tick(restarted, 140, "ABOVE") + assert len(sink.renewals) == 1 + + +def test_persistent_global_budget_survives_runner_restart(): + _, conn, _, _ = setup() + assert emitter.PersistentWriteBudget(conn, 30).try_take(100) + restarted = emitter.PersistentWriteBudget(conn, 30) + assert not restarted.try_take(110) + assert restarted.try_take(130) + + +def test_nonpositive_old_row_cannot_starve_eligible_refresh(): + definition, conn, sink, _ = setup() + before = [{"item": item, "verdict": "BELOW"} for item in ("a", "b")] + above = [{"item": item, "verdict": "ABOVE", "event_id": "source-" + item} for item in ("a", "b")] + emitter.observe(conn, definition, before, 10) + emitter.observe(conn, definition, above, 20) + emitter.deliver_pending(conn, sink, 20, definition.label) + assert emitter.renew_active(conn, sink, 100, definition, {"b"}, limit=1) == (1, 0) + assert sink.renewals[0]["item"] == "b" diff --git a/test_resolution.py b/test_resolution.py index e4d847d..3b036e2 100644 --- a/test_resolution.py +++ b/test_resolution.py @@ -1,4 +1,5 @@ """Recovery retains exact alert identity and cannot overtake an unreceived firing.""" + import json import unittest @@ -9,16 +10,21 @@ class RecoveryTests(unittest.TestCase): def setUp(self): self.db = em.connect(":memory:") - self.e = em.Emitter(label="family", command=[], interval_s=60, owner="crew", - from_verdict="BELOW", to_verdict="ABOVE", - resolve_event="recovered") + self.e = em.Emitter( + label="family", + command=[], + interval_s=60, + owner="crew", + from_verdict="BELOW", + to_verdict="ABOVE", + resolve_event="recovered", + ) def tearDown(self): self.db.close() def observe(self, verdict, now): - r = {"item": "family-1", "verdict": verdict, "event_id": "crossing-1", - "owner": "crew"} + r = {"item": "family-1", "verdict": verdict, "event_id": "crossing-1", "owner": "crew"} return em.observe(self.db, self.e, [r], now) def test_unknown_and_recovery(self): @@ -31,60 +37,74 @@ def test_unknown_and_recovery(self): self.assertEqual(self.observe("BELOW", 310), []) firing = json.loads(self.db.execute("select payload from outbox where event_id='crossing-1'").fetchone()[0]) recovery = json.loads(self.db.execute("select payload from outbox where event_id=?", resolved).fetchone()[0]) - sink = AlertmanagerSink("https://example.invalid", {"alertname":"Failures", "severity":"warning"}, "family", clock=lambda: 400) + sink = AlertmanagerSink( + "https://example.invalid", {"alertname": "Failures", "severity": "warning"}, "family", clock=lambda: 400 + ) self.assertEqual(sink.labels(firing), sink.labels(recovery)) self.assertEqual(sink.payload(firing)["startsAt"], sink.payload(recovery)["startsAt"]) self.assertEqual(recovery["resolved_at"], 300) self.assertLess(sink.payload(recovery)["endsAt"], sink.payload(firing)["endsAt"]) def test_resolution_waits_for_firing_receipt(self): - self.observe("BELOW", 100); self.observe("ABOVE", 200); self.observe("BELOW", 300) + self.observe("BELOW", 100) + self.observe("ABOVE", 200) + self.observe("BELOW", 300) + class Sink: seen = [] fail = True + def deliver(self, event): - if self.fail: raise ConnectionError("unreceived") + if self.fail: + raise ConnectionError("unreceived") self.seen.append(event) - sink=Sink() - self.assertEqual(em.deliver_pending(self.db,sink,400,self.e.label), (0,1)) + + sink = Sink() + self.assertEqual(em.deliver_pending(self.db, sink, 400, self.e.label), (0, 1)) self.assertEqual(sink.seen, []) - sink.fail=False - self.assertEqual(em.deliver_pending(self.db,sink,500,self.e.label), (2,0)) - self.assertNotIn("resolution_of",sink.seen[0]) - self.assertEqual(sink.seen[1]["resolution_of"],"crossing-1") + sink.fail = False + self.assertEqual(em.deliver_pending(self.db, sink, 500, self.e.label), (2, 0)) + self.assertNotIn("resolution_of", sink.seen[0]) + self.assertEqual(sink.seen[1]["resolution_of"], "crossing-1") def test_generation_identity_and_complete_missing(self): self.e.generation_event_ids = True self.e.missing_is_from = True - self.observe("BELOW",100) - first=self.observe("ABOVE",200) - self.assertEqual(len(first),1) - self.assertEqual(self.observe("ABOVE",210),[]) + self.observe("BELOW", 100) + first = self.observe("ABOVE", 200) + self.assertEqual(len(first), 1) + self.assertEqual(self.observe("ABOVE", 210), []) # Partial absence is not a complete zero count. - self.assertEqual(em.observe(self.db,self.e,[],220,scope=set()),[]) - recovered=em.observe(self.db,self.e,[],300) - self.assertEqual(len(recovered),1) - second=self.observe("ABOVE",400) - self.assertEqual(len(second),1) - self.assertNotEqual(first,second) + self.assertEqual(em.observe(self.db, self.e, [], 220, scope=set()), []) + recovered = em.observe(self.db, self.e, [], 300) + self.assertEqual(len(recovered), 1) + second = self.observe("ABOVE", 400) + self.assertEqual(len(second), 1) + self.assertNotEqual(first, second) def test_resolution_receipt_requires_valid_readback(self): - self.observe("BELOW",100); self.observe("ABOVE",200) - resolved=self.observe("BELOW",300) - event=json.loads(self.db.execute("select payload from outbox where event_id=?",resolved).fetchone()[0]) - sink=AlertmanagerSink("https://example.invalid",{"alertname":"Failures","severity":"warning"},"family") - requests=[] - def request(method,path,body=None): - requests.append((method,path,body)) - return (200,b"[]") - sink._request=request + self.observe("BELOW", 100) + self.observe("ABOVE", 200) + resolved = self.observe("BELOW", 300) + event = json.loads(self.db.execute("select payload from outbox where event_id=?", resolved).fetchone()[0]) + sink = AlertmanagerSink("https://example.invalid", {"alertname": "Failures", "severity": "warning"}, "family") + requests = [] + + def request(method, path, body=None): + requests.append((method, path, body)) + return (200, b"[]") + + sink._request = request sink.deliver(event) - self.assertEqual(requests[0][2][0]["labels"]["event_id"],"crossing-1") - self.assertIn("crossing-1",requests[1][1]) - sink._request=lambda method,path,body=None:(200,b'[{"labels":{"event_id":"crossing-1"}}]') - with self.assertRaises(ConnectionError):sink.deliver(event) - sink._request=lambda method,path,body=None:(200,b'{}') - with self.assertRaises(ConnectionError):sink.deliver(event) + self.assertEqual(requests[0][2][0]["labels"]["event_id"], "crossing-1") + self.assertIn("crossing-1", requests[1][1]) + sink._request = lambda method, path, body=None: (200, b'[{"labels":{"event_id":"crossing-1"}}]') + with self.assertRaises(ConnectionError): + sink.deliver(event) + sink._request = lambda method, path, body=None: (200, b"{}") + with self.assertRaises(ConnectionError): + sink.deliver(event) -if __name__ == "__main__": unittest.main() +if __name__ == "__main__": + unittest.main() From ad1bb0cf13e1d581a81f966eaf1ca2b9682fe6d8 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sun, 11 Oct 2026 18:21:26 -0400 Subject: [PATCH 3/5] fix: use received-firing recovery and verify exact renewed leases --- alertmanager_sink.py | 34 +++++++++- chaski.py | 14 ++--- emitter.py | 33 +++------- test_renewal.py | 32 +++++++++- test_renewal_receiver.py | 53 ++++++++++++++++ test_resolution.py | 130 +++++++-------------------------------- 6 files changed, 149 insertions(+), 147 deletions(-) create mode 100644 test_renewal_receiver.py diff --git a/alertmanager_sink.py b/alertmanager_sink.py index 250cbc3..97558d9 100644 --- a/alertmanager_sink.py +++ b/alertmanager_sink.py @@ -136,10 +136,38 @@ def deliver(self, event: dict) -> None: raise ConnectionError(f"{event['event_id']}: accepted but not active on read-back") def renew(self, event: dict) -> None: - """Receipt-backed same-label lease refresh; no new domain event.""" - if event.get("resolution_of"): + """Refresh the receiver's exact prior labels, never current policy labels.""" + if event.get("resolution_of") or event.get("prior"): raise ValueError("cannot renew recovery") - self.deliver(event) + alerts = self._prior_alerts(event) + if len(alerts) != 1: + raise ConnectionError("renewal prior identity absent or ambiguous") + prior = alerts[0] + if not all(isinstance(value, str) for value in prior["labels"].values()): + raise ConnectionError("malformed renewal labels") + start = prior.get("startsAt") + if not isinstance(start, str): + raise ConnectionError("missing renewal start") + parsed = dt.datetime.fromisoformat(start.replace("Z", "+00:00")) + if parsed.tzinfo is None or parsed.timestamp() > self._clock(): + raise ConnectionError("invalid renewal start") + end = _iso(self._clock() + self.ttl_s) + payload = {"labels": prior["labels"], "startsAt": start, "endsAt": end, + "annotations": prior.get("annotations", {})} + status, _ = self._request("POST", "/api/v2/alerts", [payload]) + if status // 100 != 2: + raise ConnectionError("renewal POST not accepted") + received = self._prior_alerts(event) + if len(received) != 1 or received[0]["labels"] != payload["labels"]: + raise ConnectionError("renewal exact identity not received") + observed_end = received[0].get("endsAt") + if not isinstance(observed_end, str): + raise ConnectionError("missing renewal expiry receipt") + expiry = dt.datetime.fromisoformat(observed_end.replace("Z", "+00:00")) + requested = dt.datetime.fromisoformat(end.replace("Z", "+00:00")) + if expiry.tzinfo is None or expiry < requested or received[0].get("startsAt") != start: + raise ConnectionError("renewal lease not received") + def delivery_scope(self) -> dict: return {"kind": "alertmanager", "base": self.base, "emitter": self.emitter_label} diff --git a/chaski.py b/chaski.py index 213d1a4..ec4a22c 100644 --- a/chaski.py +++ b/chaski.py @@ -342,19 +342,18 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: need = [f for f in ("severity", "action", "condition", "comment") if not g.get(f)] if need: raise RuleError(f"emitter {d['label']}: graph sink needs the Reaction fields {need}") - if "resolve_event" in d and (not isinstance(d["resolve_event"], str) - or not d["resolve_event"].strip()): - raise RuleError("resolve_event must be a nonempty event name") + if "resolve_event" in d: + raise RuleError("use the received-firing resolve_recovery protocol") for option in ("generation_event_ids", "missing_is_from"): if option in d and type(d[option]) is not bool: raise RuleError(f"{option} must be a boolean") - if d.get("missing_is_from") and not (d.get("resolve_event") and d.get("generation_event_ids")): - raise RuleError("complete-inventory recovery requires resolve_event and generation_event_ids") + if d.get("missing_is_from") and not (d.get("resolve_recovery") and d.get("generation_event_ids")): + raise RuleError("complete-inventory recovery requires resolve_recovery and generation_event_ids") renew = d.get("renew_interval_s", 0) if type(renew) is not int or renew < 0: raise RuleError("renew_interval_s must be a nonnegative integer") if renew and (d.get("trigger", "schedule") != "schedule" or "alertmanager" not in kinds - or not d.get("resolve_event") or not d.get("generation_event_ids")): + or not d.get("resolve_recovery") or not d.get("generation_event_ids")): raise RuleError("renewal requires a scheduled generation-bound recovering alert emitter") if renew: ttl = d["sink"]["alertmanager"].get("ttl_s", 86400) @@ -363,8 +362,6 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: recovery = d.get("resolve_recovery", False) if not isinstance(recovery, bool) or (recovery and ("command" in kinds or d.get("trigger") == "external")): raise RuleError("recovery must be boolean and requires a verdict emitter without a command sink") - if recovery and d.get("resolve_event"): - raise RuleError("select one recovery protocol") e = emitter.Emitter(label=d["label"], command=list(d.get("command", [])), interval_s=parse_duration(d["schedule"]), owner=d["owner"], event=d.get("event", "unblocked"), timeout_s=int(d.get("timeout_s", 300)), @@ -372,7 +369,6 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]: from_verdict=d.get("from", emitter.BLOCKED), to_verdict=d.get("to", emitter.UNBLOCKED), baseline_emits=bool(d.get("baseline_emits", False)), - resolve_event=d.get("resolve_event"), generation_event_ids=bool(d.get("generation_event_ids", False)), missing_is_from=bool(d.get("missing_is_from", False)), resolve_recovery=recovery, diff --git a/emitter.py b/emitter.py index b8e4faa..3a44fde 100644 --- a/emitter.py +++ b/emitter.py @@ -151,7 +151,6 @@ class Emitter: from_verdict: str = BLOCKED to_verdict: str = UNBLOCKED baseline_emits: bool = False - resolve_event: str | None = None generation_event_ids: bool = False missing_is_from: bool = False change_driven: bool = False @@ -306,26 +305,6 @@ def emit(r: dict, item: str, generation: int) -> None: (verdict, generation, r.get("evidence"), now, emitter.label, item)) if last == emitter.from_verdict and verdict == emitter.to_verdict: emit(r, item, generation) - elif (emitter.resolve_event and last == emitter.to_verdict - and verdict == emitter.from_verdict and last_event): - # Resolve the exact frozen firing identity, never a new label set. - old = conn.execute("SELECT payload FROM outbox WHERE event_id=?", - (last_event,)).fetchone() - if old is None: - raise ProtocolError("resolution has no retained firing payload") - payload = json.loads(old[0]) - resolution_id = hashlib.sha256( - (last_event + ":resolved").encode()).hexdigest() - payload.update(event_id=resolution_id, event=emitter.resolve_event, - resolution_of=last_event, resolved_at=now, - generation=generation) - cur = conn.execute( - "INSERT OR IGNORE INTO outbox" - " (event_id,emitter,item,generation,payload,created) VALUES (?,?,?,?,?,?)", - (resolution_id, emitter.label, item, generation, - json.dumps(payload, sort_keys=True), now)) - if cur.rowcount: - queued.append(resolution_id) for (item,) in conn.execute( "SELECT item FROM verdicts WHERE emitter=? AND tracked=1", (emitter.label,)).fetchall(): if item not in present and (scope is None or item in scope): @@ -577,7 +556,7 @@ def renew_active(conn: sqlite3.Connection, sink: Sink, now: float, definition: E if not callable(renewal): raise ProtocolError("renewal requires a lease-aware sink") rows = conn.execute( - "SELECT v.item, o.event_id, o.payload, COALESCE(r.attempts,0) " + "SELECT v.item, o.event_id, o.payload, COALESCE(r.attempts,0), o.delivery_scope " "FROM verdicts v JOIN outbox o ON o.event_id=v.last_event " "LEFT JOIN renewal_receipts r ON r.event_id=o.event_id " "WHERE v.emitter=? AND v.tracked=1 AND v.verdict=? " @@ -588,7 +567,7 @@ def renew_active(conn: sqlite3.Connection, sink: Sink, now: float, definition: E (definition.label, definition.to_verdict, json.dumps(sorted(positive)), now, now-definition.renew_interval_s, limit)).fetchall() ok = failed = 0 - for item, identity, payload, attempts in rows: + for item, identity, payload, attempts, bound_scope in rows: if item not in positive: continue event = json.loads(payload) @@ -597,6 +576,9 @@ def renew_active(conn: sqlite3.Connection, sink: Sink, now: float, definition: E if budget is not None and not budget.try_take(now): break try: + current_scope = getattr(sink, "delivery_scope", lambda: None)() + if bound_scope is None or current_scope != json.loads(bound_scope): + raise ProtocolError("renewal receiver scope changed or unproven") renewal(event) except Exception as exc: failed += 1 @@ -648,7 +630,10 @@ def __init__(self, emitter: Emitter, conn: sqlite3.Connection, sink: Sink, self.last_attempt = saved[0] if saved else 0.0 self.last_success = saved[1] if saved else 0.0 self.runs = {"ok": saved[2] if saved else 0, "unknown": saved[3] if saved else 0} - self.queued_total = 0 + self.queued_total = (conn.execute( + "SELECT (SELECT COUNT(*) FROM outbox WHERE emitter=?)" + " + (SELECT COUNT(*) FROM recovery_outbox WHERE emitter=?)", + (emitter.label, emitter.label)).fetchone()[0] if emitter.persist_runs else 0) self._snap_lock = threading.Lock() self._snapshot: dict = {} self._publish() diff --git a/test_renewal.py b/test_renewal.py index 931f4e6..ebe9e42 100644 --- a/test_renewal.py +++ b/test_renewal.py @@ -11,6 +11,15 @@ def __init__(self): self.renewals = [] self.fail = False + def delivery_scope(self): + return {"kind": "test", "receiver": "lease"} + + def prepare_recovery(self, event): + return {"delivery_scope": self.delivery_scope()} + + def resolve(self, event): + self.deliveries.append(dict(event)) + def deliver(self, event): self.deliveries.append(dict(event)) @@ -28,7 +37,7 @@ def setup(): owner="crew", from_verdict="BELOW", to_verdict="ABOVE", - resolve_event="recovered", + resolve_recovery=True, generation_event_ids=True, renew_interval_s=30, persist_runs=True, @@ -72,7 +81,7 @@ def test_recovery_and_new_generation_cannot_refresh_old_identity(): tick(runner, 20, "ABOVE") old = sink.deliveries[0]["event_id"] tick(runner, 100, "BELOW") - assert sink.deliveries[1]["resolution_of"] == old + assert sink.deliveries[1]["prior"]["event_id"] == old tick(runner, 200, "ABOVE") new = sink.deliveries[2]["event_id"] assert new != old @@ -121,3 +130,22 @@ def test_nonpositive_old_row_cannot_starve_eligible_refresh(): emitter.deliver_pending(conn, sink, 20, definition.label) assert emitter.renew_active(conn, sink, 100, definition, {"b"}, limit=1) == (1, 0) assert sink.renewals[0]["item"] == "b" + + +def test_changed_receiver_scope_never_refreshes_prior_firing(): + _, _, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + sink.delivery_scope = lambda: {"kind": "test", "receiver": "other"} + tick(runner, 100, "ABOVE") + assert sink.renewals == [] + assert runner._snapshot["renewal_failures"] == 1 + + +def test_one_shot_queued_counter_does_not_reset_on_restart(): + definition, conn, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + tick(runner, 100, "BELOW") + restarted = emitter.Runner(definition, conn, sink) + assert restarted.queued_total == runner.queued_total == 2 diff --git a/test_renewal_receiver.py b/test_renewal_receiver.py new file mode 100644 index 0000000..ca430b0 --- /dev/null +++ b/test_renewal_receiver.py @@ -0,0 +1,53 @@ +"""Receiver-side lease receipts preserve the existing exact label set.""" +import copy +import json + +import pytest + +from alertmanager_sink import AlertmanagerSink, _iso + + +def receiver(stale=False): + sink = AlertmanagerSink("https://example.invalid", + {"alertname": "CurrentPolicy", "severity": "critical", + "ttl_s": 60}, "lease", clock=lambda: 100) + event = {"event_id": "identity", "emitter": "lease", "item": "family", + "observed_at": 20, "recipient": "crew"} + prior = {"labels": {"alertname": "OriginalPolicy", "severity": "warning", + "event_id": "identity", "emitter": "lease", "item": "family", + "keeper": "crew"}, + "startsAt": _iso(20), "endsAt": _iso(120), "annotations": {}} + posts = [] + current = copy.deepcopy(prior) + def request(method, path, body=None): + if method == "POST": + posts.append(copy.deepcopy(body[0])) + if not stale: + current.update(body[0]) + return 200, b"" + return 200, json.dumps([current]).encode() + sink._request = request + return sink, event, prior, posts + + +def test_renewal_uses_prior_labels_and_requires_extended_expiry(): + sink, event, prior, posts = receiver() + sink.renew(event) + assert posts[0]["labels"] == prior["labels"] + assert posts[0]["startsAt"] == prior["startsAt"] + assert posts[0]["endsAt"] == _iso(160) + + +def test_accepted_post_with_old_expiry_is_not_a_receipt(): + sink, event, _, _ = receiver(stale=True) + with pytest.raises(ConnectionError, match="lease not received"): + sink.renew(event) + + +def test_absent_or_ambiguous_prior_never_posts_new_policy_labels(): + for rows in ([], [{"labels": {"event_id": "other"}}]): + sink, event, _, posts = receiver() + sink._request = lambda *args: (200, json.dumps(rows).encode()) + with pytest.raises(ConnectionError): + sink.renew(event) + assert posts == [] diff --git a/test_resolution.py b/test_resolution.py index 3b036e2..bdc6078 100644 --- a/test_resolution.py +++ b/test_resolution.py @@ -1,110 +1,22 @@ -"""Recovery retains exact alert identity and cannot overtake an unreceived firing.""" - -import json -import unittest - +"""Complete-inventory recovery must use the existing received-firing protocol.""" import emitter as em -from alertmanager_sink import AlertmanagerSink - - -class RecoveryTests(unittest.TestCase): - def setUp(self): - self.db = em.connect(":memory:") - self.e = em.Emitter( - label="family", - command=[], - interval_s=60, - owner="crew", - from_verdict="BELOW", - to_verdict="ABOVE", - resolve_event="recovered", - ) - - def tearDown(self): - self.db.close() - - def observe(self, verdict, now): - r = {"item": "family-1", "verdict": verdict, "event_id": "crossing-1", "owner": "crew"} - return em.observe(self.db, self.e, [r], now) - - def test_unknown_and_recovery(self): - self.assertEqual(self.observe("BELOW", 100), []) - self.assertEqual(self.observe("ABOVE", 200), ["crossing-1"]) - self.assertEqual(self.observe("ABOVE", 210), []) - self.assertEqual(self.observe("UNKNOWN", 220), []) - resolved = self.observe("BELOW", 300) - self.assertEqual(len(resolved), 1) - self.assertEqual(self.observe("BELOW", 310), []) - firing = json.loads(self.db.execute("select payload from outbox where event_id='crossing-1'").fetchone()[0]) - recovery = json.loads(self.db.execute("select payload from outbox where event_id=?", resolved).fetchone()[0]) - sink = AlertmanagerSink( - "https://example.invalid", {"alertname": "Failures", "severity": "warning"}, "family", clock=lambda: 400 - ) - self.assertEqual(sink.labels(firing), sink.labels(recovery)) - self.assertEqual(sink.payload(firing)["startsAt"], sink.payload(recovery)["startsAt"]) - self.assertEqual(recovery["resolved_at"], 300) - self.assertLess(sink.payload(recovery)["endsAt"], sink.payload(firing)["endsAt"]) - - def test_resolution_waits_for_firing_receipt(self): - self.observe("BELOW", 100) - self.observe("ABOVE", 200) - self.observe("BELOW", 300) - - class Sink: - seen = [] - fail = True - - def deliver(self, event): - if self.fail: - raise ConnectionError("unreceived") - self.seen.append(event) - - sink = Sink() - self.assertEqual(em.deliver_pending(self.db, sink, 400, self.e.label), (0, 1)) - self.assertEqual(sink.seen, []) - sink.fail = False - self.assertEqual(em.deliver_pending(self.db, sink, 500, self.e.label), (2, 0)) - self.assertNotIn("resolution_of", sink.seen[0]) - self.assertEqual(sink.seen[1]["resolution_of"], "crossing-1") - - def test_generation_identity_and_complete_missing(self): - self.e.generation_event_ids = True - self.e.missing_is_from = True - self.observe("BELOW", 100) - first = self.observe("ABOVE", 200) - self.assertEqual(len(first), 1) - self.assertEqual(self.observe("ABOVE", 210), []) - # Partial absence is not a complete zero count. - self.assertEqual(em.observe(self.db, self.e, [], 220, scope=set()), []) - recovered = em.observe(self.db, self.e, [], 300) - self.assertEqual(len(recovered), 1) - second = self.observe("ABOVE", 400) - self.assertEqual(len(second), 1) - self.assertNotEqual(first, second) - - def test_resolution_receipt_requires_valid_readback(self): - self.observe("BELOW", 100) - self.observe("ABOVE", 200) - resolved = self.observe("BELOW", 300) - event = json.loads(self.db.execute("select payload from outbox where event_id=?", resolved).fetchone()[0]) - sink = AlertmanagerSink("https://example.invalid", {"alertname": "Failures", "severity": "warning"}, "family") - requests = [] - - def request(method, path, body=None): - requests.append((method, path, body)) - return (200, b"[]") - - sink._request = request - sink.deliver(event) - self.assertEqual(requests[0][2][0]["labels"]["event_id"], "crossing-1") - self.assertIn("crossing-1", requests[1][1]) - sink._request = lambda method, path, body=None: (200, b'[{"labels":{"event_id":"crossing-1"}}]') - with self.assertRaises(ConnectionError): - sink.deliver(event) - sink._request = lambda method, path, body=None: (200, b"{}") - with self.assertRaises(ConnectionError): - sink.deliver(event) - - -if __name__ == "__main__": - unittest.main() +from test_renewal import Sink + + +def test_missing_complete_inventory_recovers_exact_received_generation(): + db = em.connect(":memory:") + definition = em.Emitter("family", [], 0, "crew", from_verdict="BELOW", + to_verdict="ABOVE", generation_event_ids=True, + resolve_recovery=True, missing_is_from=True) + sink = Sink() + em.observe(db, definition, [{"item": "a", "verdict": "BELOW"}], 10) + em.observe(db, definition, [{"item": "a", "verdict": "ABOVE", "event_id": "a"}], 20) + em.deliver_pending(db, sink, 20, definition.label) + firing = sink.deliveries[0] + em.observe(db, definition, [], 30, scope=set()) + assert db.execute("SELECT COUNT(*) FROM recovery_outbox").fetchone()[0] == 0 + em.observe(db, definition, [], 40) + em.deliver_pending(db, sink, 40, definition.label, resolve_recovery=True) + assert sink.deliveries[1]["prior"] == firing + assert db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0] == 1 + assert db.execute("SELECT COUNT(*) FROM recovery_outbox").fetchone()[0] == 1 From b3ba10aae40fd7386d23924a3ea042c4522a097f Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sun, 11 Oct 2026 18:26:04 -0400 Subject: [PATCH 4/5] docs: use the received-firing recovery option --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 9cebd9d..38b6c56 100644 --- a/README.md +++ b/README.md @@ -178,4 +178,4 @@ this flag, replay production history or claim natural scheduled acceptance. `failure_family_adapter.py` consumes the `dp failure-families` JSON report and a directly typed Quipu `ConfigSetting` containing a single JSON description. Policy fields are `default_threshold`, `window_seconds`, `default_owner`, optional `family_thresholds` (hashed family ID to threshold), and `tool_owners` (tool name to crew owner). Invalid/missing/ambiguous policy or failed aggregation exits nonzero, leaving the emitter UNKNOWN. Ownership comes from this graph policy, with an explicit crew fallback; human recipients are refused. -Configure the emitter with `from: BELOW`, `to: ABOVE`, `resolve_event: recovered`, `generation_event_ids: true`, and `missing_is_from: true`. Keep `baseline_emits: false`: first sighting above threshold is a baseline, not an observed crossing. A complete successful rolling-window inventory can resolve absent families; partial observations and failed runs cannot. Unsupported program shapes are a separate disclosed measurement population, not silently attributed to a known family. Each actual crossing gets a generation-bound identity, stable across retries and restarts. Recovery refers to the original frozen firing labels and cannot overtake its receipt. The existing Alertmanager sink retains its finite TTL; scheduled delivery and producer-liveness monitoring are still deployment requirements. This adapter is not scheduled or enabled by the repository itself. +Configure the emitter with `from: BELOW`, `to: ABOVE`, `resolve_recovery: true`, `generation_event_ids: true`, and `missing_is_from: true`. Keep `baseline_emits: false`: first sighting above threshold is a baseline, not an observed crossing. A complete successful rolling-window inventory can resolve absent families; partial observations and failed runs cannot. Unsupported program shapes are a separate disclosed measurement population, not silently attributed to a known family. Each actual crossing gets a generation-bound identity, stable across retries and restarts. Recovery refers to the original frozen firing labels and cannot overtake its receipt. The existing Alertmanager sink retains its finite TTL; scheduled delivery and producer-liveness monitoring are still deployment requirements. This adapter is not scheduled or enabled by the repository itself. From ad29a1c9f3a1d9b0e4c255138fdbd1d695905b33 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Sun, 11 Oct 2026 18:29:05 -0400 Subject: [PATCH 5/5] fix: reset renewal backoff after received lease refresh --- emitter.py | 13 +++++++------ test_renewal.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/emitter.py b/emitter.py index 3a44fde..a22d240 100644 --- a/emitter.py +++ b/emitter.py @@ -556,7 +556,8 @@ def renew_active(conn: sqlite3.Connection, sink: Sink, now: float, definition: E if not callable(renewal): raise ProtocolError("renewal requires a lease-aware sink") rows = conn.execute( - "SELECT v.item, o.event_id, o.payload, COALESCE(r.attempts,0), o.delivery_scope " + "SELECT v.item, o.event_id, o.payload, " + "CASE WHEN r.last_error IS NOT NULL THEN r.attempts ELSE 0 END, o.delivery_scope " "FROM verdicts v JOIN outbox o ON o.event_id=v.last_event " "LEFT JOIN renewal_receipts r ON r.event_id=o.event_id " "WHERE v.emitter=? AND v.tracked=1 AND v.verdict=? " @@ -585,14 +586,14 @@ def renew_active(conn: sqlite3.Connection, sink: Sink, now: float, definition: E backoff = min(MAX_BACKOFF_S, BASE_BACKOFF_S * 2 ** min(attempts, 20)) conn.execute( "INSERT INTO renewal_receipts(event_id,attempts,next_attempt,last_error) " - "VALUES(?,1,?,?) ON CONFLICT(event_id) DO UPDATE SET " - "attempts=attempts+1,next_attempt=excluded.next_attempt,last_error=excluded.last_error", - (identity, now+backoff, str(exc)[:300])) + "VALUES(?,?,?,?) ON CONFLICT(event_id) DO UPDATE SET " + "attempts=excluded.attempts,next_attempt=excluded.next_attempt,last_error=excluded.last_error", + (identity, attempts+1, now+backoff, str(exc)[:300])) continue conn.execute( "INSERT INTO renewal_receipts(event_id,last_received,attempts,next_attempt) " - "VALUES(?,?,1,0) ON CONFLICT(event_id) DO UPDATE SET " - "last_received=excluded.last_received,attempts=attempts+1,next_attempt=0,last_error=NULL", + "VALUES(?,?,0,0) ON CONFLICT(event_id) DO UPDATE SET " + "last_received=excluded.last_received,attempts=0,next_attempt=0,last_error=NULL", (identity, now)) ok += 1 return ok, failed diff --git a/test_renewal.py b/test_renewal.py index ebe9e42..616fafc 100644 --- a/test_renewal.py +++ b/test_renewal.py @@ -149,3 +149,35 @@ def test_one_shot_queued_counter_does_not_reset_on_restart(): tick(runner, 100, "BELOW") restarted = emitter.Runner(definition, conn, sink) assert restarted.queued_total == runner.queued_total == 2 + + +def test_success_receipts_reset_consecutive_failure_backoff_across_restart(): + definition, conn, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + for now in (100, 200, 300): + tick(runner, now, "ABOVE") + assert len(sink.renewals) == 3 + sink.fail = True + tick(runner, 400, "ABOVE") + assert conn.execute("SELECT attempts,next_attempt FROM renewal_receipts").fetchone() == (1, 430) + restarted = emitter.Runner(definition, conn, sink) + tick(restarted, 430, "ABOVE") + assert conn.execute("SELECT attempts,next_attempt FROM renewal_receipts").fetchone() == (2, 490) + sink.fail = False + tick(restarted, 490, "ABOVE") + assert conn.execute("SELECT attempts,next_attempt FROM renewal_receipts").fetchone() == (0, 0) + sink.fail = True + tick(restarted, 530, "ABOVE") + assert conn.execute("SELECT attempts,next_attempt FROM renewal_receipts").fetchone() == (1, 560) + + +def test_legacy_success_attempt_count_does_not_inflate_failure_backoff(): + _, conn, sink, runner = setup() + tick(runner, 10, "BELOW") + tick(runner, 20, "ABOVE") + tick(runner, 100, "ABOVE") + conn.execute("UPDATE renewal_receipts SET attempts=100,last_error=NULL") + sink.fail = True + tick(runner, 200, "ABOVE") + assert conn.execute("SELECT attempts,next_attempt FROM renewal_receipts").fetchone() == (1, 230)