From 73535de57361f9d5057cfd255dd6efc036c1d4a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Wed, 21 Oct 2026 18:00:21 +0200 Subject: [PATCH 1/3] test(v0.49): prove missed deadline is delivered once after cold outage --- tests/integrations/pg-mdm/time/audit.py | 66 +++++++++++++++++++++++-- 1 file changed, 63 insertions(+), 3 deletions(-) diff --git a/tests/integrations/pg-mdm/time/audit.py b/tests/integrations/pg-mdm/time/audit.py index fd0356f..707e7b6 100644 --- a/tests/integrations/pg-mdm/time/audit.py +++ b/tests/integrations/pg-mdm/time/audit.py @@ -15,8 +15,9 @@ def main(): image, output = sys.argv[1:3] maintenance = sys.argv[3:] == ["--maintenance"] compatibility_only = sys.argv[3:] == ["--compatibility"] - if sys.argv[3:] and not (maintenance or compatibility_only): - raise ValueError("only --maintenance or --compatibility is supported after IMAGE OUTPUT") + outage_mode = sys.argv[3:] == ["--outage"] + if sys.argv[3:] and not (maintenance or compatibility_only or outage_mode): + raise ValueError("only --maintenance, --compatibility or --outage is supported after IMAGE OUTPUT") output = Path(output).resolve() output.mkdir(parents=True, exist_ok=False) root = Path(__file__).resolve().parents[4] @@ -311,7 +312,49 @@ def cases(): temporal_source=temporal_source, cases_expected=expected) sql("ALTER SYSTEM SET pg_react.databases = 's1_idle'") sql("ALTER SYSTEM SET pg_react.worker_role = 'mdm_s1_runtime'") - command("docker", "restart", container) + outage_before = None + if outage_mode: + # A genuine PostgreSQL outage starts before the *future* deadline, + # remains down while that deadline passes, and ends only after it. + # This is not a manual clock input, source update or worker tick. + outage_postmaster_before = timestamp(sql("SELECT pg_postmaster_start_time()")) + outage_before = json.loads(sql( + "SELECT jsonb_build_object(" + "'cases',(SELECT COALESCE(jsonb_agg(to_jsonb(c) ORDER BY case_key),'[]') FROM mdm_steward.policy_cases_v1 c)," + "'receipts',(SELECT COALESCE(jsonb_agg(to_jsonb(r) ORDER BY created_at,receipt_id),'[]') FROM mdm_steward.policy_receipts_v1 r)," + "'requests',(SELECT COALESCE(jsonb_agg(to_jsonb(r) ORDER BY binding_id,request_key),'[]') FROM pgreact_mdm.intent_requests r)," + "'attempts',(SELECT COALESCE(jsonb_agg(to_jsonb(a) ORDER BY episode_id,attempt_no),'[]') FROM pgreact_mdm.intent_attempts a)," + "'work',(SELECT COALESCE(jsonb_agg(to_jsonb(w) ORDER BY work_id),'[]') FROM pgreact.work w)," + "'business',(SELECT COALESCE(jsonb_agg(to_jsonb(s) ORDER BY id),'[]') FROM public.policy_qualification_source s)," + "'technical',(SELECT COALESCE(jsonb_agg(to_jsonb(s) ORDER BY case_key),'[]') FROM s1_audit.source s))")) + assert outage_before["cases"] == before + assert outage_before["business"] == source + assert outage_before["technical"] == temporal_source + assert not [r for r in outage_before["receipts"] + if r["request_body"].get("policy_revision") == "s1-idle-feasibility"] + command("docker", "stop", "-t", "10", container) + stopped_at = datetime.datetime.now(datetime.timezone.utc) + assert stopped_at < deadline, { + "error": "PostgreSQL did not stop before the case was due", + "stopped_at": stopped_at.isoformat(), "deadline": deadline.isoformat()} + assert command("docker", "inspect", "-f", "{{.State.Running}}", container).strip() == "false" + # Wait for the due boundary while PostgreSQL is completely stopped. + # This sleep is a controlled outage stimulus, not the evidence of + # delivery; the later real worker receipt and complete vectors are. + while datetime.datetime.now(datetime.timezone.utc) <= deadline + datetime.timedelta(seconds=2): + time.sleep(0.2) + resumed_at = datetime.datetime.now(datetime.timezone.utc) + assert resumed_at > deadline + command("docker", "start", container) + observations["outage_across_deadline"] = { + "postmaster_before": outage_postmaster_before.isoformat(), + "stopped_at": stopped_at.isoformat(), + "deadline_at": deadline.isoformat(), + "resume_requested_at": resumed_at.isoformat(), + "complete_state_before": outage_before, + } + else: + command("docker", "restart", container) end = time.monotonic() + 120 receipts = [] while time.monotonic() < end: @@ -392,6 +435,17 @@ def cases(): assert receipt["request_body"]["action"] == "ESCALATE" and receipt["request_body"]["arguments"] == {"level": 2}, receipts delivered_at = timestamp(receipt["created_at"]) assert delivered_at >= deadline, receipts + if outage_mode: + postmaster_after = timestamp(sql("SELECT pg_postmaster_start_time()")) + assert postmaster_after > outage_postmaster_before and postmaster_after > deadline + assert delivered_at >= postmaster_after >= resumed_at + observations["outage_across_deadline"].update({ + "postmaster_after": postmaster_after.isoformat(), + "delivered_at": delivered_at.isoformat(), + "one_case_caught_up": True, + }) + observations["checks"].append({ + "name": "outage-deadline-catchup-single-case", "result": "passed"}) actual = cases() assert actual == expected, {"expected": expected, "actual": actual} assert json.loads(sql("SELECT jsonb_agg(to_jsonb(s) ORDER BY s.id) " @@ -644,6 +698,12 @@ def cases(): "unchanged": True, } observations["checks"].append({"name": "post-delivery-restart-no-replay", "result": "passed"}) + if outage_mode: + # Focused T49.06 proof: this run also checked post-delivery cycles + # and a second real restart without replay. The ordinary full + # audit runs separately to retain all later security/MDM vectors. + observations["entry_decision"] = "FOCUSED_OUTAGE_QUALIFIED" + return assert json.loads(sql("SELECT jsonb_agg(to_jsonb(r) ORDER BY created_at,receipt_id) " "FROM mdm_steward.policy_receipts_v1 r")) == retained_receipts sql("CREATE ROLE mdm_s1_runtime_no_set LOGIN") From c9c54038732c8dc1c44dcfbea2422bbf48e45314 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Wed, 21 Oct 2026 18:00:42 +0200 Subject: [PATCH 2/3] ci(v0.49): run actual missed-deadline outage and no-replay qualification --- .github/workflows/producer-compatibility.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/producer-compatibility.yml b/.github/workflows/producer-compatibility.yml index de03f0b..ff48694 100644 --- a/.github/workflows/producer-compatibility.yml +++ b/.github/workflows/producer-compatibility.yml @@ -46,6 +46,12 @@ jobs: run: | python3 tests/integrations/pg-mdm/time/audit.py pgreact:producer-compatibility producer-evidence/runtime python3 -c 'import json; o=json.load(open("producer-evidence/runtime/observations.json")); passed={c["name"] for c in o["checks"] if c["result"]=="passed"}; assert {"future-idle-mdm-receipt","stale-operator-runtime-version"} <= passed, {"passed":passed,"failure":o.get("failure")}; assert o["stale_operator_runtime_version"]["exact"] and o["stale_operator_runtime_version"]["complete_state_unchanged"]' + - name: Qualify real downtime crossing an undelivered deadline + run: | + set -euo pipefail + python3 tests/integrations/pg-mdm/time/audit.py \ + pgreact:producer-compatibility producer-evidence/outage-runtime --outage + python3 -c 'import json; o=json.load(open("producer-evidence/outage-runtime/observations.json")); names={c["name"] for c in o["checks"] if c["result"]=="passed"}; assert o["entry_decision"]=="FOCUSED_OUTAGE_QUALIFIED"; assert {"outage-deadline-catchup-single-case","future-idle-mdm-receipt","post-delivery-idle-quiescence","post-delivery-restart-no-replay"} <= names, names; assert o["outage_across_deadline"]["one_case_caught_up"]' - name: Record and compare inherited packaged shared libraries run: | set -euo pipefail From d37a2dca02d4924d82465940c8d27388d38d1b5b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geir=20O=2E=20Gr=C3=B8nmo?= Date: Wed, 21 Oct 2026 18:43:43 +0200 Subject: [PATCH 3/3] test(v0.49): record qualified outage receipt and carry forward restart regression --- tests/integrations/pg-mdm/time/audit.py | 96 +++++++++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/tests/integrations/pg-mdm/time/audit.py b/tests/integrations/pg-mdm/time/audit.py index 707e7b6..a3323cd 100644 --- a/tests/integrations/pg-mdm/time/audit.py +++ b/tests/integrations/pg-mdm/time/audit.py @@ -699,6 +699,11 @@ def cases(): } observations["checks"].append({"name": "post-delivery-restart-no-replay", "result": "passed"}) if outage_mode: + # This branch already proved the complete canonical MDM receipt, + # actual action/case/source vector and worker role above. Record + # the normal proof marker before ending this focused outage run. + observations["checks"].append({ + "name": "future-idle-mdm-receipt", "result": "passed"}) # Focused T49.06 proof: this run also checked post-delivery cycles # and a second real restart without replay. The ordinary full # audit runs separately to retain all later security/MDM vectors. @@ -1463,6 +1468,97 @@ def sql_literal(value): observations["checks"].append({ "name": "stale-operator-runtime-version", "result": "passed" }) + # The active binding and the exact rejected stale operator attempts + # must survive another true PostgreSQL postmaster epoch. Observe + # multiple resumed managed-worker cycles, then compare complete + # persisted cases, receipts, runtime binding rows, and React/source + # work to rule out a delayed replay or binding-state regression. + final_before = json.loads(sql(quiescence_sql)) + final_binding_before = json.loads(sql( + "SELECT jsonb_build_object(" + "'bindings',(SELECT COALESCE(jsonb_agg(to_jsonb(b) ORDER BY binding_id),'[]')" + " FROM pgreact_mdm.policy_intent_bindings b)," + "'runtime',(SELECT COALESCE(jsonb_agg(to_jsonb(r) ORDER BY binding_id),'[]')" + " FROM pgreact_mdm.policy_intent_runtime r)," + "'mdm_runtime',(SELECT COALESCE(jsonb_agg(to_jsonb(r) ORDER BY binding_id),'[]')" + " FROM mdm_internal.policy_binding_runtime r))")) + assert any(item["binding_id"] == resumed["binding_id"] + and item["runtime_version"] == resumed["reactivated_runtime_version"] + for item in final_binding_before["runtime"]), final_binding_before + final_start = timestamp(sql("SELECT pg_postmaster_start_time()")) + final_status = json.loads(sql("SELECT pgreact_api.managed_status()")) + assert final_status["process"]["state"] == "ready", final_status + final_heartbeat = timestamp(final_status["process"]["heartbeat_at"]) + command("docker", "restart", container) + end = time.monotonic() + 60 + restarted = None + new_postmaster = None + while time.monotonic() < end: + probe = subprocess.run( + ["docker", "exec", container, "pg_isready", "-U", "postgres"], + capture_output=True) + if probe.returncode: + time.sleep(0.2) + continue + new_postmaster = timestamp(sql("SELECT pg_postmaster_start_time()")) + restarted = json.loads(sql("SELECT pgreact_api.managed_status()")) + worker = restarted.get("process") or {} + heartbeat = worker.get("heartbeat_at") + if (new_postmaster > final_start and worker.get("state") == "ready" + and heartbeat and timestamp(heartbeat) > final_heartbeat): + break + time.sleep(0.2) + else: + raise AssertionError({ + "error": "resumed binding did not recover after restart", + "postmaster_before": final_start.isoformat(), + "postmaster_after": str(new_postmaster), + "status": restarted, + }) + restart_heartbeat_first = timestamp(restarted["process"]["heartbeat_at"]) + end = time.monotonic() + 35 + while time.monotonic() < end: + restarted = json.loads(sql("SELECT pgreact_api.managed_status()")) + worker = restarted.get("process") or {} + heartbeat = worker.get("heartbeat_at") + if (worker.get("state") == "ready" and heartbeat + and timestamp(heartbeat) > restart_heartbeat_first + + datetime.timedelta(seconds=3)): + break + time.sleep(0.2) + else: + raise AssertionError("resumed binding worker did not complete later restart cycles") + final_after = json.loads(sql(quiescence_sql)) + final_binding_after = json.loads(sql( + "SELECT jsonb_build_object(" + "'bindings',(SELECT COALESCE(jsonb_agg(to_jsonb(b) ORDER BY binding_id),'[]')" + " FROM pgreact_mdm.policy_intent_bindings b)," + "'runtime',(SELECT COALESCE(jsonb_agg(to_jsonb(r) ORDER BY binding_id),'[]')" + " FROM pgreact_mdm.policy_intent_runtime r)," + "'mdm_runtime',(SELECT COALESCE(jsonb_agg(to_jsonb(r) ORDER BY binding_id),'[]')" + " FROM mdm_internal.policy_binding_runtime r))")) + assert final_after == final_before, { + "error": "post-resume restart replayed or changed durable state", + "before": final_before, "after": final_after, + } + assert final_binding_after == final_binding_before, { + "error": "post-resume restart changed binding version or state", + "before": final_binding_before, "after": final_binding_after, + } + observations["post_resume_restart"] = { + "postmaster_before": final_start.isoformat(), + "postmaster_after": new_postmaster.isoformat(), + "first_heartbeat": restart_heartbeat_first.isoformat(), + "later_heartbeat": heartbeat, + "full_before": final_before, + "full_after": final_after, + "bindings_before": final_binding_before, + "bindings_after": final_binding_after, + "unchanged": True, + } + observations["checks"].append({ + "name": "post-resume-restart-no-replay", "result": "passed" + }) observations["entry_decision"] = "INPUT_BLOCKED" observations["remaining_inputs"] = ["1000-case/churn workload measurements", "signed numeric operating envelope"] print("PASS: future idle deadline -> actual MDM receipt; registered callback failure and role reset; complete case/source vectors", flush=True)