feat: add scheduled failure-family alerts and retained leases

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_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.

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)])
@@ -128,6 +135,39 @@ 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:
+        """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")
+        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}
 
@@ -227,3 +267,10 @@ def resolve(self, event: dict) -> 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)

chaski.py

@@ -342,6 +342,23 @@ 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:
+            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_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_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)
+            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")
         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")
@@ -352,9 +369,12 @@ 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)),
+                            generation_event_ids=bool(d.get("generation_event_ids", False)),
+                            missing_is_from=bool(d.get("missing_is_from", False)),
                             resolve_recovery=recovery,
                             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:

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.