the-custodian/docs/changes/netkingdom-target-2026-10-16/group-postgres-r1/source.patch

diff --git a/docs/service-group-provenance.md b/docs/service-group-provenance.md
--- a/docs/service-group-provenance.md
+++ b/docs/service-group-provenance.md
@@ -80,6 +80,64 @@
 and replay. No arbitrary JSON import, receipt expiry or automatic uncertainty
 clearing is introduced. Owner reconciliation needs independent evidence.
 
+## PostgreSQL source seam (not native admission)
+
+`PostgresGroupProvenance` implements the same reserve/get/claim-revoke/save/check
+journal contract. It is injection-only: the ordinary HTTP server has no
+PostgreSQL environment switch and remains without a group store unless the
+existing local SQLite configuration is explicitly supplied. `GROUP_POSTGRES_DSN`
+or a caller JSON object cannot enable the PostgreSQL path. Production group
+operations remain unadmitted and disabled pending the owner callbacks and gates
+above; a backend constructor is not an access decision.
+
+A trusted owner integration must provide an exact `GroupJournalBinding` with
+opaque authoritative tenant scope, directory instance, accepted case digest,
+source digest, represented principal and journal epoch. There is no guessed
+platform-root scope or request-derived tenant fallback. The owner also injects a
+renewable private connection-info callback using the existing `rapp-postgres`
+consumer/lease contract, and independently durable `JournalContinuity` callbacks.
+These callbacks retain/check/advance exact epoch+directory+head+commit digest outside the
+PostgreSQL state that might be restored. Neither callback authenticates a caller
+or grants a protected side effect. Their native custody, source binding,
+revocation and execution bounds still require independent admission. Only the
+disposable fixture supplies a test checkpoint implementation here.
+
+The v1 migration under `identity-provisioner/migrations/` creates tables within
+the existing consumer schema; it creates no database/schema/roles, inserts no
+epoch and clears no quarantine. Startup does not run migrations or initialize
+missing state. The owner must separately admit an epoch/checkpoint after
+independent continuity/recovery review; absent or quarantined state refuses.
+A restored database with an older head than the independently retained checkpoint
+also refuses rather than treating missing later intents as authority to replay.
+
+PostgreSQL transactions lock the single journal metadata row, commit bounded
+receipts/intents and advance the head and exact transition digest with an exactly-one conditional
+update. The digest binds the prior marker, canonical body, context, request key
+and receipt version; the independent checkpoint must compare/CAS both content
+and sequence, never accept a counter-only callback. Divergent restored clones
+that commit different transitions at the same counter cannot both be accepted.
+Every receipt update also requires its exact context/key/request/version and
+increments its version. The active slot spans all contexts in the journal;
+it serializes cooperating producer operations, not external LDAP writers.
+SQL commit precedes independently durable checkpoint acknowledgment. A lost
+commit/acknowledgment, stale version, missing row or continuity mismatch refuses
+and poisons that backend object; exception handling cannot automatically rewrite
+its journal into success. Known request-binding mismatches remain domain
+refusals. Definite before-mutation refusal records can release the active slot
+while the original key remains terminal.
+
+The container source includes a hash-pinned Linux amd64 CPython3.12 psycopg3
+closure selected from User Engine's existing lockfile. No image build or package
+installation is part of this source proof; build/publication, artifact scanning
+and native runtime/dependency admission remain owner work. SQLite is still
+local/disposable only. The separate PostgreSQL fixture proposal uses a cached
+pinned PostgreSQL image and a cached pinned User Engine Python image, real SQL,
+separate-process reopen, concurrency, stale-CAS and actual pg_dump/pg_restore
+behind an independently retained test checkpoint. That fixture needs exact
+review before Docker execution and does not prove live CNPG HA, TLS/lease
+custody, tenant authority, external writer fencing, production retention or
+protected session/evidence integration.
+
 ## Safety and remaining native acceptance
 
 Before each directory mutation, durable intent is committed. A successful
diff --git a/identity-provisioner/Containerfile b/identity-provisioner/Containerfile
--- a/identity-provisioner/Containerfile
+++ b/identity-provisioner/Containerfile
@@ -3,7 +3,10 @@
 RUN useradd --create-home --uid 10001 provisioner
 WORKDIR /app
 COPY --from=lldap /app/lldap_set_password /app/lldap_set_password
-COPY provisioner.py group_provenance.py password_setup.py server.py /app/
+COPY requirements-postgres.txt /app/
+RUN python -m pip install --no-cache-dir --require-hashes -r /app/requirements-postgres.txt
+COPY provisioner.py group_provenance.py group_provenance_postgres.py password_setup.py server.py /app/
+COPY migrations/ /app/migrations/
 USER 10001:10001
 EXPOSE 8080
 CMD ["python", "server.py"]
diff --git a/identity-provisioner/group_provenance.py b/identity-provisioner/group_provenance.py
--- a/identity-provisioner/group_provenance.py
+++ b/identity-provisioner/group_provenance.py
@@ -11,8 +11,17 @@
 from pathlib import Path
 import sqlite3
 import stat
-from typing import Any
+from typing import Any, Protocol
 from urllib.parse import quote
+
+
+class GroupJournalStore(Protocol):
+    """Owner-injected persistence only; grants no caller or native authority."""
+    def check(self) -> None: ...
+    def reserve(self, key: str, binding: str) -> tuple[dict[str, Any], bool]: ...
+    def get(self, key: str, binding: str) -> dict[str, Any]: ...
+    def claim_revoke(self, key: str, binding: str, delete_if_empty: bool) -> dict[str, Any]: ...
+    def save(self, receipt: dict[str, Any]) -> None: ...
 
 
 class GroupProvenanceUnavailable(RuntimeError):
diff --git a/identity-provisioner/group_provenance_postgres.py b/identity-provisioner/group_provenance_postgres.py
new file mode 100644
--- /dev/null
+++ b/identity-provisioner/group_provenance_postgres.py
@@ -0,0 +1,253 @@
+"""Injection-only PostgreSQL implementation of the existing owner journal seam.
+
+Binding and continuity callbacks are supplied by a trusted owner integration,
+never constructed from an HTTP payload. Neither is an authorization decision.
+No server environment switch, automatic schema creation or recovery admission.
+"""
+from __future__ import annotations
+
+from contextlib import contextmanager
+from dataclasses import asdict, dataclass
+import hashlib
+import json
+import re
+from typing import Any, Callable, Protocol
+from uuid import UUID
+
+from group_provenance import GroupProvenance, GroupProvenanceUnavailable
+
+
+@dataclass(frozen=True)
+class GroupJournalBinding:
+    tenant_scope: str
+    directory: str
+    case_digest: str
+    source_digest: str
+    principal: str
+    epoch: str
+
+    def __post_init__(self):
+        for value in (self.tenant_scope, self.directory, self.principal):
+            if not isinstance(value, str) or not 1 <= len(value) <= 256 or any(ord(c) < 32 for c in value):
+                raise ValueError("explicit owner journal binding required")
+        for value in (self.case_digest, self.source_digest):
+            if not isinstance(value, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", value):
+                raise ValueError("exact owner source and case digests required")
+        if not isinstance(self.epoch, str) or str(UUID(self.epoch)) != self.epoch:
+            raise ValueError("explicit canonical journal epoch required")
+
+    @property
+    def digest(self):
+        return hashlib.sha256(json.dumps(asdict(self), sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def journal_genesis(epoch: str, directory: str) -> str:
+    """Deterministic initial marker for explicit owner migration/admission only."""
+    return hashlib.sha256(json.dumps({"schema": "nk-group-journal-genesis.v1", "epoch": epoch,
+        "directory": directory}, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+class JournalContinuity(Protocol):
+    """Existing independently retained owner checkpoint, outside restored SQL state.
+
+    Implementations compare/advance exact epoch+directory+head+commit digest atomically and
+    durably. An in-memory implementation is a fixture only. Missing/stale or
+    ambiguous checkpoints must refuse; no callback grants mutation authority.
+    """
+    def check(self, epoch: str, directory: str, head: int, digest: str) -> bool: ...
+    def advance(self, epoch: str, directory: str, previous: int, previous_digest: str, current: int, current_digest: str) -> bool: ...
+
+
+class PostgresGroupProvenance:
+    def __init__(self, *, connect: Callable, binding: GroupJournalBinding,
+                 continuity: JournalContinuity, capacity: int = 256):
+        if not callable(connect) or not isinstance(binding, GroupJournalBinding) or not 1 <= capacity <= 4096:
+            raise ValueError("explicit owner connector, binding and bounded capacity required")
+        if not callable(getattr(continuity, "check", None)) or not callable(getattr(continuity, "advance", None)):
+            raise ValueError("independent owner continuity callbacks required")
+        self.connect, self.binding, self.continuity, self.capacity = connect, binding, continuity, capacity
+        self._poisoned = False
+
+    def _continuity(self, method, *args):
+        try:
+            if getattr(self.continuity, method)(self.binding.epoch, self.binding.directory, *args) is not True:
+                raise GroupProvenanceUnavailable()
+        except Exception:
+            raise GroupProvenanceUnavailable() from None
+
+    @contextmanager
+    def _transaction(self):
+        if self._poisoned:
+            raise GroupProvenanceUnavailable()
+        connection = None
+        tx = None
+        try:
+            connection = self.connect()
+            with connection:
+                with connection.cursor() as cursor:
+                    cursor.execute("SET LOCAL statement_timeout = '5s'")
+                    cursor.execute("SET LOCAL idle_in_transaction_session_timeout = '15s'")
+                    cursor.execute("SELECT epoch::text, directory, head, head_digest, quarantined FROM identity_provisioner.group_journal_meta WHERE singleton = true FOR UPDATE")
+                    meta = cursor.fetchone()
+                    if not meta or meta[0] != self.binding.epoch or meta[1] != self.binding.directory or meta[4] is not False or type(meta[2]) is not int or meta[2] < 0 or not isinstance(meta[3], str) or not re.fullmatch(r"[0-9a-f]{64}", meta[3]):
+                        raise GroupProvenanceUnavailable()
+                    self._continuity("check", meta[2], meta[3])
+                    tx = {"cursor": cursor, "head": meta[2], "digest": meta[3], "dirty": False}
+                    yield tx
+                    if tx["dirty"]:
+                        cursor.execute("UPDATE identity_provisioner.group_journal_meta SET head = head + 1, head_digest = %s WHERE singleton = true AND epoch = %s AND directory = %s AND head = %s AND head_digest = %s AND quarantined = false",
+                                       (tx["next_digest"], self.binding.epoch, self.binding.directory, tx["head"], tx["digest"]))
+                        if cursor.rowcount != 1:
+                            raise GroupProvenanceUnavailable()
+            # SQL commit precedes the independent checkpoint acknowledgment.
+            # A lost acknowledgment refuses before the next directory effect.
+            if tx and tx["dirty"]:
+                self._continuity("advance", tx["head"], tx["digest"], tx["head"] + 1, tx["next_digest"])
+        except ValueError:
+            raise
+        except GroupProvenanceUnavailable:
+            # Caller mismatches are domain refusals. Storage/continuity failures
+            # poison this object; durable pending records are never rewritten
+            # automatically by an exception handler after ambiguous commit.
+            self._poisoned = True
+            raise
+        except Exception:
+            self._poisoned = True
+            raise GroupProvenanceUnavailable() from None
+        finally:
+            if connection is not None:
+                try:
+                    connection.close()
+                except Exception:
+                    self._poisoned = True
+                    raise GroupProvenanceUnavailable() from None
+
+    def check(self):
+        with self._transaction():
+            pass
+
+    def _read(self, cursor, key, request_binding):
+        cursor.execute("SELECT binding, version, body, context_json FROM identity_provisioner.group_receipts WHERE context_digest = %s AND key = %s",
+                       (self.binding.digest, key))
+        row = cursor.fetchone()
+        if row is None:
+            return None
+        if row[0] != request_binding:
+            raise ValueError("ensure receipt request binding mismatch")
+        if row[3] != json.dumps(asdict(self.binding), sort_keys=True):
+            raise GroupProvenanceUnavailable()
+        receipt = GroupProvenance._load(row[2], key, request_binding)
+        if type(row[1]) is not int or row[1] < 1 or type(receipt.get("_pg_version")) is not int or receipt.get("_pg_version") != row[1] or receipt.get("_pg_context") != self.binding.digest:
+            raise GroupProvenanceUnavailable()
+        return receipt
+
+    @staticmethod
+    def _inputs(key, binding):
+        if not isinstance(key, str) or not 1 <= len(key) <= 256 or not isinstance(binding, str) or not re.fullmatch(r"[0-9a-f]{64}", binding):
+            raise ValueError("bounded exact journal request binding required")
+
+    def _active(self, cursor):
+        cursor.execute("SELECT 1 FROM identity_provisioner.group_receipts WHERE active = true LIMIT 1")
+        if cursor.fetchone():
+            raise GroupProvenanceUnavailable()
+
+    def _transition(self, tx, receipt, operation):
+        # Bind the exact transition, not merely a counter. Two restored/forked
+        # copies that commit different bodies at the same next head cannot both
+        # match an independently retained checkpoint after reopening.
+        transition = {"schema": "nk-group-journal-transition.v1", "epoch": self.binding.epoch,
+            "directory": self.binding.directory, "prior_head": tx["head"], "prior_digest": tx["digest"],
+            "head": tx["head"] + 1, "operation": operation, "context": asdict(self.binding),
+            "key": receipt["key"], "request_binding": receipt["binding"],
+            "version": receipt["_pg_version"], "body": receipt}
+        tx["next_digest"] = hashlib.sha256(json.dumps(transition, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+        tx["dirty"] = True
+
+    def reserve(self, key, binding):
+        self._inputs(key, binding)
+        with self._transaction() as tx:
+            cursor = tx["cursor"]
+            receipt = self._read(cursor, key, binding)
+            if receipt is not None:
+                return receipt, False
+            self._active(cursor)
+            cursor.execute("SELECT COUNT(*) FROM identity_provisioner.group_receipts")
+            if cursor.fetchone()[0] >= self.capacity:
+                raise GroupProvenanceUnavailable()
+            receipt = {"key": key, "binding": binding, "state": "pending", "intent": None, "groups": {},
+                       "created_groups": [], "created_memberships": [], "_pg_version": 1, "_pg_context": self.binding.digest}
+            cursor.execute("INSERT INTO identity_provisioner.group_receipts (context_digest, context_json, key, binding, version, state, active, body) VALUES (%s, %s, %s, %s, 1, 'pending', true, %s)",
+                           (self.binding.digest, json.dumps(asdict(self.binding), sort_keys=True), key, binding, json.dumps(receipt, sort_keys=True)))
+            if cursor.rowcount != 1:
+                raise GroupProvenanceUnavailable()
+            self._transition(tx, receipt, "reserve")
+        return receipt, True
+
+    def get(self, key, binding):
+        self._inputs(key, binding)
+        with self._transaction() as tx:
+            receipt = self._read(tx["cursor"], key, binding)
+            if receipt is None:
+                raise ValueError("unknown ensure receipt or request binding mismatch")
+            return receipt
+
+    def _write(self, tx, receipt, operation="save"):
+        self._inputs(receipt.get("key"), receipt.get("binding"))
+        if receipt.get("_pg_context") != self.binding.digest or type(receipt.get("_pg_version")) is not int or receipt["_pg_version"] < 1:
+            raise GroupProvenanceUnavailable()
+        encoded = json.dumps(receipt, sort_keys=True)
+        GroupProvenance._load(encoded, receipt["key"], receipt["binding"])
+        updated = json.loads(encoded)
+        updated["_pg_version"] += 1
+        active = True if receipt["state"] in {"pending", "revoking", "uncertain"} else None
+        tx["cursor"].execute("UPDATE identity_provisioner.group_receipts SET state = %s, active = %s, body = %s, version = version + 1 WHERE context_digest = %s AND key = %s AND binding = %s AND version = %s",
+                             (receipt["state"], active, json.dumps(updated, sort_keys=True), self.binding.digest, receipt["key"], receipt["binding"], receipt["_pg_version"]))
+        if tx["cursor"].rowcount != 1:
+            raise GroupProvenanceUnavailable()
+        self._transition(tx, updated, operation)
+        return updated
+
+    def save(self, receipt):
+        with self._transaction() as tx:
+            updated = self._write(tx, receipt)
+        # The provisioner keeps references to nested group pre/poststate across
+        # intent persistence. Replace only this backend's acknowledged version;
+        # _write serializes without changing the caller's mutation body.
+        receipt["_pg_version"] = updated["_pg_version"]
+
+    def claim_revoke(self, key, binding, delete_if_empty):
+        self._inputs(key, binding)
+        if not isinstance(delete_if_empty, bool):
+            raise ValueError("delete_if_empty must be boolean")
+        with self._transaction() as tx:
+            receipt = self._read(tx["cursor"], key, binding)
+            if receipt is None:
+                raise ValueError("unknown ensure receipt or request binding mismatch")
+            if receipt["state"] == "revoked":
+                if receipt["delete_if_empty"] != delete_if_empty:
+                    raise ValueError("completed rollback options cannot change")
+                return receipt
+            if receipt["state"] != "ensured":
+                raise GroupProvenanceUnavailable()
+            self._active(tx["cursor"])
+            receipt.update(state="revoking", delete_if_empty=delete_if_empty)
+            updated = self._write(tx, receipt, "claim_revoke")
+        return updated
+
+
+def psycopg_connector(*, connection_info: Callable[[], dict[str, Any]]):
+    """Owner-provided renewable lease values stay in memory; never env/argv/logs.
+
+    No connection happens here. Lazy import keeps disabled group endpoints free
+    of a runtime dependency. The admitted owner supplies TLS/host/db/role facts.
+    """
+    if not callable(connection_info):
+        raise ValueError("owner connection-info callback required")
+    def connect():
+        try:
+            import psycopg
+            values = dict(connection_info())
+            return psycopg.connect(**values, connect_timeout=5)
+        except Exception:
+            raise GroupProvenanceUnavailable() from None
+    return connect
diff --git a/identity-provisioner/migrations/group_journal_postgres_v1.sql b/identity-provisioner/migrations/group_journal_postgres_v1.sql
new file mode 100644
--- /dev/null
+++ b/identity-provisioner/migrations/group_journal_postgres_v1.sql
@@ -0,0 +1,23 @@
+-- Run only by the admitted migration role in its existing owner schema.
+-- Does not create a database/schema/credential, admit an epoch or clear quarantine.
+CREATE TABLE identity_provisioner.group_journal_meta (
+    singleton boolean PRIMARY KEY CHECK (singleton = true),
+    epoch uuid NOT NULL,
+    directory text NOT NULL CHECK (octet_length(directory) BETWEEN 1 AND 1024),
+    head bigint NOT NULL CHECK (head >= 0),
+    head_digest text NOT NULL CHECK (head_digest ~ '^[0-9a-f]{64}$'),
+    quarantined boolean NOT NULL DEFAULT true
+);
+CREATE TABLE identity_provisioner.group_receipts (
+    context_digest text NOT NULL CHECK (context_digest ~ '^[0-9a-f]{64}$'),
+    context_json text NOT NULL CHECK (octet_length(context_json) <= 16384),
+    key text NOT NULL CHECK (char_length(key) BETWEEN 1 AND 256),
+    binding text NOT NULL CHECK (binding ~ '^[0-9a-f]{64}$'),
+    version bigint NOT NULL CHECK (version >= 1),
+    state text NOT NULL CHECK (state IN ('pending','ensured','revoking','revoked','uncertain','refused')),
+    active boolean UNIQUE CHECK (active IS NULL OR active = true),
+    body text NOT NULL CHECK (octet_length(body) <= 1048576),
+    PRIMARY KEY (context_digest, key)
+);
+-- Epoch insertion, initial continuity checkpoint and admission are separate
+-- owner actions; this migration leaves all operations refused without them.
diff --git a/identity-provisioner/provisioner.py b/identity-provisioner/provisioner.py
--- a/identity-provisioner/provisioner.py
+++ b/identity-provisioner/provisioner.py
@@ -11,7 +11,7 @@
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 
-from group_provenance import GroupProvenance, GroupProvenanceUnavailable
+from group_provenance import GroupJournalStore, GroupProvenanceUnavailable
 
 
 @dataclass(frozen=True)
@@ -64,7 +64,7 @@
         opener: Callable = urlopen,
         auth_rejected_cooldown: float = 30.0,
         clock: Callable[[], float] = time.monotonic,
-        group_provenance: GroupProvenance | None = None,
+        group_provenance: GroupJournalStore | None = None,
     ) -> None:
         self.group_provenance = group_provenance
         self.base_url = base_url.rstrip("/")
@@ -243,7 +243,7 @@
             timeout=timeout,
         )
 
-    def _group_store(self) -> GroupProvenance:
+    def _group_store(self) -> GroupJournalStore:
         store = self.group_provenance
         if store is None:
             raise GroupProvenanceUnavailable()
diff --git a/identity-provisioner/requirements-postgres.txt b/identity-provisioner/requirements-postgres.txt
new file mode 100644
--- /dev/null
+++ b/identity-provisioner/requirements-postgres.txt
@@ -0,0 +1,5 @@
+# Linux amd64 CPython3.12 runtime closure from user-engine/uv.lock.
+# Build-only dependency resolution; no packages are installed at service startup.
+psycopg==3.3.5 --hash=sha256:ce5aa5cdb4f9379f00f487590e5890bfa7df9a164648c969ffa628505e21af4e
+psycopg-binary==3.3.5 --hash=sha256:682a17a57415c3ca1731eec018ed031f012ffcb81ba74806eb219cb396065672
+typing-extensions==4.16.0 --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8
diff --git a/identity-provisioner/tests/test_group_postgres.py b/identity-provisioner/tests/test_group_postgres.py
new file mode 100644
--- /dev/null
+++ b/identity-provisioner/tests/test_group_postgres.py
@@ -0,0 +1,231 @@
+"""Binding, CAS and ambiguous-commit refusal tests; native SQL proof is separate."""
+from dataclasses import replace
+import json
+from unittest.mock import Mock
+import unittest
+
+from group_provenance import GroupProvenanceUnavailable
+from group_provenance_postgres import GroupJournalBinding, PostgresGroupProvenance
+from server import configured_group_provenance
+from unittest.mock import patch
+import os
+
+EPOCH = '41fbb0ed-ea3a-4682-996b-e9c94a6aa9c6'
+BINDING = GroupJournalBinding('tenant:fixture:one', 'synthetic-directory', 'sha256:' + '1' * 64,
+                             'sha256:' + '2' * 64, 'synthetic-owner', EPOCH)
+REQUEST = '3' * 64
+HEAD_DIGEST = 'a' * 64
+META = (EPOCH, 'synthetic-directory', 7, HEAD_DIGEST, False)
+
+
+def fixture(rows=(), rowcount=1, commit_error=None):
+    cursor = Mock()
+    cursor.__enter__ = Mock(return_value=cursor)
+    cursor.__exit__ = Mock(return_value=False)
+    cursor.fetchone.side_effect = rows
+    cursor.rowcount = rowcount
+    connection = Mock()
+    connection.__enter__ = Mock(return_value=connection)
+    connection.__exit__ = Mock(return_value=False, side_effect=commit_error)
+    connection.cursor.return_value = cursor
+    connector = Mock(return_value=connection)
+    continuity = Mock()
+    continuity.check.return_value = True
+    continuity.advance.return_value = True
+    store = PostgresGroupProvenance(connect=connector, binding=BINDING, continuity=continuity)
+    return store, connector, connection, cursor, continuity
+
+
+def receipt(version=4):
+    return {'key': 'synthetic-operation-key', 'binding': REQUEST, 'state': 'pending', 'intent': None,
+            'groups': {}, 'created_groups': [], 'created_memberships': [],
+            '_pg_version': version, '_pg_context': BINDING.digest}
+
+
+class PostgresSourceTests(unittest.TestCase):
+    def test_explicit_binding_and_continuity_required_without_connecting(self):
+        connector = Mock()
+        for binding, continuity in ((None, Mock()), (BINDING, None), ({'tenant_scope': 'caller-json'}, Mock())):
+            with self.assertRaises(ValueError):
+                PostgresGroupProvenance(connect=connector, binding=binding, continuity=continuity)
+        connector.assert_not_called()
+        for change in ({'tenant_scope': ''}, {'principal': ''}, {'case_digest': 'unsigned'}, {'epoch': 'default'}):
+            with self.assertRaises(ValueError):
+                replace(BINDING, **change)
+        self.assertNotEqual(BINDING.digest, replace(BINDING, principal='another-owner').digest)
+        self.assertNotEqual(BINDING.digest, replace(BINDING, tenant_scope='another-explicit-scope').digest)
+
+    def test_missing_environment_callback_path_keeps_production_disabled(self):
+        with patch.dict(os.environ, {'GROUP_POSTGRES_DSN': 'synthetic-only', 'GROUP_PROVENANCE_POSTGRES': 'true'}, clear=True):
+            self.assertIsNone(configured_group_provenance())
+
+    def test_checkpoint_mismatch_poisoned_without_receipt_query_or_effect(self):
+        store, connect, _, cursor, checkpoint = fixture([META])
+        checkpoint.check.return_value = False
+        with self.assertRaises(GroupProvenanceUnavailable):
+            store.reserve('synthetic-operation-key', REQUEST)
+        self.assertFalse(any('group_receipts' in c.args[0] for c in cursor.execute.call_args_list))
+        with self.assertRaises(GroupProvenanceUnavailable):
+            store.check()
+        self.assertEqual(1, connect.call_count)
+        checkpoint.advance.assert_not_called()
+
+    def test_restored_epoch_or_quarantine_refused(self):
+        for meta in ((EPOCH, 'other-directory', 7, HEAD_DIGEST, False), (EPOCH, 'synthetic-directory', 7, HEAD_DIGEST, True),
+                     ('98aa2b6d-a32a-44a5-a89f-b2480befc00e', 'synthetic-directory', 7, HEAD_DIGEST, False), None):
+            store, _, _, _, checkpoint = fixture([meta])
+            with self.assertRaises(GroupProvenanceUnavailable):
+                store.check()
+            checkpoint.check.assert_not_called()
+
+    def test_exactly_one_version_update_required_before_checkpoint_advance(self):
+        store, _, _, cursor, checkpoint = fixture([META], rowcount=0)
+        original = receipt()
+        with self.assertRaises(GroupProvenanceUnavailable):
+            store.save(original)
+        query, params = next(c.args for c in cursor.execute.call_args_list if c.args[0].startswith('UPDATE identity_provisioner.group_receipts'))
+        self.assertIn('AND version = %s', query)
+        self.assertEqual(4, params[-1])
+        self.assertEqual(4, original['_pg_version'])
+        checkpoint.advance.assert_not_called()
+
+    def test_success_acknowledges_commit_and_updates_receipt_version(self):
+        store, _, connection, _, checkpoint = fixture([META])
+        original = receipt()
+        store.save(original)
+        self.assertEqual(5, original['_pg_version'])
+        connection.__exit__.assert_called_once()
+        args = checkpoint.advance.call_args.args
+        self.assertEqual((EPOCH, 'synthetic-directory', 7, HEAD_DIGEST, 8), args[:5])
+        self.assertRegex(args[5], r'^[0-9a-f]{64}$')
+        self.assertNotEqual(HEAD_DIGEST, args[5])
+        connection.close.assert_called_once()
+
+    def test_save_preserves_nested_receipt_references_held_by_provisioner(self):
+        store, *_ = fixture([META])
+        original = receipt()
+        original['groups']['forge-coulomb-owners'] = {'id': None, 'before': [], 'after': ['tegwick']}
+        state = original['groups']['forge-coulomb-owners']
+        groups, created, memberships = original['groups'], original['created_groups'], original['created_memberships']
+        original['intent'] = {'operation': 'create_group', 'name': 'forge-coulomb-owners'}
+        store.save(original)
+        self.assertIs(groups, original['groups'])
+        self.assertIs(state, original['groups']['forge-coulomb-owners'])
+        self.assertIs(created, original['created_groups'])
+        self.assertIs(memberships, original['created_memberships'])
+        state['id'] = 4
+        self.assertEqual(4, original['groups']['forge-coulomb-owners']['id'])
+        self.assertEqual(5, original['_pg_version'])
+
+    def test_actual_provisioner_ensure_with_postgres_save_retains_created_ids(self):
+        from test_service_groups import Directory, req
+        directory = Directory()
+        directory.members[3].add('tegwick')
+        store, _, _, cursor, _ = fixture([META] * 32)
+        original = receipt(version=1)
+        payload = req()
+        original['key'] = payload['idempotency_key']
+        original['binding'] = directory._group_binding(payload['groups'], payload['members'], payload['correlation_id'])
+        # Reservation input is already covered by SQL unit/native oracles. Keep
+        # real save/_write/transaction and the real provisioner creation path.
+        store.reserve = Mock(return_value=(original, True))
+        directory.group_provenance = store
+        result = directory.ensure_service_groups(payload)
+        self.assertEqual('groups_ensured', result.status)
+        self.assertEqual({'forge-coulomb-owners', 'forge-coulomb-readers'}, {g['name'] for g in result.created_groups})
+        self.assertEqual(1, len(result.created_memberships))
+        self.assertEqual({'tegwick'}, directory.members[3])
+        bodies = [json.loads(c.args[1][2]) for c in cursor.execute.call_args_list
+                  if c.args[0].startswith('UPDATE identity_provisioner.group_receipts')]
+        self.assertTrue(all(g['id'] is not None for g in bodies[-1]['groups'].values()))
+        self.assertEqual('ensured', bodies[-1]['state'])
+
+    def test_connection_close_failure_is_sanitized_and_stops_continuation(self):
+        store, _, connection, _, _ = fixture([META])
+        connection.close.side_effect = OSError('synthetic-private-diagnostic')
+        with self.assertRaisesRegex(GroupProvenanceUnavailable, '^group_provenance_unavailable$'):
+            store.check()
+        self.assertTrue(store._poisoned)
+
+    def test_commit_failure_never_advances_owner_checkpoint_or_receipt(self):
+        store, _, _, _, checkpoint = fixture([META], commit_error=OSError('synthetic-secret-diagnostic'))
+        original = receipt()
+        with self.assertRaisesRegex(GroupProvenanceUnavailable, '^group_provenance_unavailable$'):
+            store.save(original)
+        self.assertEqual(4, original['_pg_version'])
+        checkpoint.advance.assert_not_called()
+
+    def test_lost_checkpoint_acknowledgment_refuses_then_poisoned(self):
+        store, connect, _, _, checkpoint = fixture([META])
+        checkpoint.advance.side_effect = OSError('synthetic-secret-diagnostic')
+        original = receipt()
+        with self.assertRaisesRegex(GroupProvenanceUnavailable, '^group_provenance_unavailable$'):
+            store.save(original)
+        self.assertEqual(4, original['_pg_version'])
+        with self.assertRaises(GroupProvenanceUnavailable):
+            store.save(original)
+        self.assertEqual(1, connect.call_count)
+
+    def test_cross_case_body_or_version_cannot_be_imported(self):
+        for body in (dict(receipt(), _pg_context='0' * 64), dict(receipt(), _pg_version=6)):
+            store, _, _, _, _ = fixture([META, (REQUEST, 4, json.dumps(body), json.dumps(BINDING.__dict__, sort_keys=True))])
+            with self.assertRaises(GroupProvenanceUnavailable):
+                store.get('synthetic-operation-key', REQUEST)
+
+    def test_same_counter_divergent_body_or_context_yields_distinct_marker(self):
+        store, *_ = fixture()
+        markers = []
+        for change in ({}, {'intent': {'operation': 'different'}}, {'key': 'another-exact-key'}, {'_pg_version': 5}):
+            tx = {'head': 7, 'digest': HEAD_DIGEST}
+            store._transition(tx, dict(receipt(), **change), 'save')
+            markers.append(tx['next_digest'])
+        self.assertEqual(4, len(set(markers)))
+        tx = {'head': 7, 'digest': 'b' * 64}
+        store._transition(tx, receipt(), 'save')
+        self.assertNotIn(tx['next_digest'], markers)
+        foreign, *_ = fixture()
+        foreign.binding = replace(BINDING, principal='another-owner')
+        tx = {'head': 7, 'digest': HEAD_DIGEST}
+        foreign._transition(tx, receipt(), 'save')
+        self.assertNotIn(tx['next_digest'], markers)
+
+    def test_reopened_loser_clone_at_same_head_refuses_different_marker(self):
+        # Both DBs report head8. The independent callback accepts only winner
+        # digest; loser must be refused before loading any receipt or effect.
+        loser_digest = 'b' * 64
+        store, _, _, cursor, checkpoint = fixture([(EPOCH, 'synthetic-directory', 8, loser_digest, False)])
+        checkpoint.check.side_effect = lambda epoch, directory, head, digest: (head, digest) == (8, 'c' * 64)
+        with self.assertRaises(GroupProvenanceUnavailable):
+            store.check()
+        checkpoint.check.assert_called_once_with(EPOCH, 'synthetic-directory', 8, loser_digest)
+        self.assertFalse(any('group_receipts' in c.args[0] for c in cursor.execute.call_args_list))
+
+    def test_existing_request_mismatch_is_refused_not_rebound(self):
+        store, _, _, _, checkpoint = fixture([META, ('0' * 64, 4, '{}', '{}')])
+        with self.assertRaisesRegex(ValueError, 'binding mismatch'):
+            store.get('synthetic-operation-key', REQUEST)
+        checkpoint.advance.assert_not_called()
+        self.assertFalse(store._poisoned)
+
+    def test_active_fence_capacity_refusal_and_no_insert(self):
+        for rows in ([META, None, (1,)], [META, None, None, (256,)]):
+            store, _, _, cursor, checkpoint = fixture(rows)
+            with self.assertRaises(GroupProvenanceUnavailable):
+                store.reserve('synthetic-operation-key', REQUEST)
+            self.assertFalse(any('INSERT' in c.args[0] for c in cursor.execute.call_args_list))
+            checkpoint.advance.assert_not_called()
+
+    def test_new_reservation_retains_bound_context_and_write_intent(self):
+        store, _, _, cursor, checkpoint = fixture([META, None, None, (0,)])
+        created, fresh = store.reserve('synthetic-operation-key', REQUEST)
+        self.assertTrue(fresh)
+        self.assertEqual('pending', created['state'])
+        self.assertEqual(BINDING.digest, created['_pg_context'])
+        insert = next(c for c in cursor.execute.call_args_list if c.args[0].startswith('INSERT'))
+        self.assertIn('context_json', insert.args[0])
+        self.assertEqual(BINDING.tenant_scope, json.loads(insert.args[1][1])['tenant_scope'])
+        checkpoint.advance.assert_called_once()
+
+
+if __name__ == '__main__':
+    unittest.main()