diff --git a/README.md b/README.md index 6d3cbdc..38b6c56 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_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. diff --git a/alertmanager_sink.py b/alertmanager_sink.py index 3c971d1..97558d9 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)]) @@ -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) diff --git a/chaski.py b/chaski.py index 550cceb..ec4a22c 100644 --- a/chaski.py +++ b/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: 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 0582c3c..a22d240 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, @@ -133,8 +151,12 @@ class Emitter: from_verdict: str = BLOCKED to_verdict: str = UNBLOCKED baseline_emits: bool = False + generation_event_ids: bool = False + missing_is_from: bool = False change_driven: bool = False externally_triggered: bool = False + renew_interval_s: int = 0 + persist_runs: bool = False resolve_recovery: bool = False @property @@ -204,11 +226,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, @@ -259,7 +292,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=?", @@ -454,6 +487,12 @@ def deliver_pending(conn: sqlite3.Connection, sink: Sink, now: float, emitter: s rows = conn.execute(query + " ORDER BY created LIMIT ?", args).fetchall() ok = failed = 0 for event_id, payload, attempts, _created, table 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: + 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: @@ -502,6 +541,79 @@ 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, " + "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=? " + "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, bound_scope 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: + 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 + 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(?,?,?,?) 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(?,?,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 + + +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. @@ -513,15 +625,22 @@ 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} - self.queued_total = 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 = (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() 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: @@ -529,12 +648,22 @@ 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) self.queued_total += len(queue_recoveries(self.conn, self.emitter)) deliver_pending(self.conn, self.sink, now, self.emitter.label, self.budget, limit=DELIVER_PER_TICK, resolve_recovery=self.emitter.resolve_recovery) + 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: @@ -547,7 +676,14 @@ def _publish(self) -> None: "SELECT (SELECT COUNT(*) FROM outbox WHERE emitter=? AND delivered_at IS NOT NULL)" " + (SELECT COUNT(*) FROM recovery_outbox WHERE emitter=? AND delivered_at IS NOT NULL)", (self.emitter.label, 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: @@ -562,6 +698,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 new file mode 100644 index 0000000..e0812ea --- /dev/null +++ b/failure_family_adapter.py @@ -0,0 +1,165 @@ +"""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 os +from pathlib import Path +import shutil +import tempfile +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}; " + f"window={policy['window_seconds']}s; " + f"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) + # 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") + print(json.dumps(verdicts(json.loads(result.stdout), policy))) + + +if __name__ == "__main__": + main() diff --git a/graph_sink.py b/graph_sink.py index b831e12..899e4c8 100644 --- a/graph_sink.py +++ b/graph_sink.py @@ -158,6 +158,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..d462799 --- /dev/null +++ b/once.py @@ -0,0 +1,161 @@ +"""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 delivery_scope(self): + return self.receiver.delivery_scope() + + def prepare_recovery(self, event): + return self.receiver.prepare_recovery(event) + + def resolve(self, event): + try: + self.receiver.resolve(event) + except Exception: + self.conn.execute("DELETE FROM receiver_setup WHERE receiver=?", (self.key,)) + raise + + 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 new file mode 100644 index 0000000..4acff3b --- /dev/null +++ b/test_failure_family_adapter.py @@ -0,0 +1,85 @@ +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() + + +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..616fafc --- /dev/null +++ b/test_renewal.py @@ -0,0 +1,183 @@ +"""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 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)) + + 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_recovery=True, + 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]["prior"]["event_id"] == 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" + + +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 + + +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) 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 new file mode 100644 index 0000000..bdc6078 --- /dev/null +++ b/test_resolution.py @@ -0,0 +1,22 @@ +"""Complete-inventory recovery must use the existing received-firing protocol.""" +import emitter as em +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