NK-WP-0048-T15 source: adapter-owned original-intent reconcile

build() accepts paired holder/observe read-only capabilities and installs
Adapter.reconcile_original; verify accepts the published reconcile record.
Fixture now exercises the adapter's own path. No add replay, no undo after
restart; missing/ambiguous/changed state stays uncertain. Holder choice
remains an owner gate.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 903097@bnt-lap001
Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23

sso-mfa/k8s/keycape/keycape-registration-adapter.py

@@ -85,7 +85,7 @@ def descriptor(check_config_digest):
 def build(*, admitted: Invocation, case_source: bytes, expected_uid: str,
           expected_resource_version: str, case_acceptance: str,
           check_config: Path, check_config_digest: str, publish, authenticate,
-          revoke_session):
+          revoke_session, holder=None, observe=None):
     """Bind one admitted case and observed revision; return a closed kernel Adapter.
 
     publish(bytes)->identifier durably retains only the closed metadata payload.
@@ -94,6 +94,14 @@ def build(*, admitted: Invocation, case_source: bytes, expected_uid: str,
     session; Warden independently authenticates it using its own evidence verifier.
     These callbacks are capabilities, never command strings or case parameters.
 
+    holder(original_attempt_id)->[(identifier, bytes)] and observe()->secret are
+    optional read-only capabilities, injected together or not at all. With them
+    the Adapter gains reconcile_original: it confirms a creation only from one
+    authenticated original created-only record whose post-creation revision and
+    exact client entry still hold. It never adds or removes; anything missing,
+    ambiguous or changed raises so Warden keeps the step uncertain. Choosing the
+    real authenticated holder is an owner gate outside this source.
+
     Only a verified creation in this process may be rolled back. Process loss,
     write/readback ambiguity, publication/revocation failure and source drift
     require owner reconciliation; none are converted to retry-safe failures.
@@ -125,6 +133,9 @@ def build(*, admitted: Invocation, case_source: bytes, expected_uid: str,
         raise Refused('acceptance evidence binding differs')
     if not all(callable(x) for x in (publish, authenticate, revoke_session)):
         raise Refused('owner evidence and revocation capabilities required')
+    if (holder is None) != (observe is None) or not all(
+            x is None or callable(x) for x in (holder, observe)):
+        raise Refused('original-attempt holder and readback must be injected together')
     with contained_output():
         case = yaml.safe_load(case_source)
         client = helper.client_from_case(case)
@@ -213,5 +224,67 @@ def build(*, admitted: Invocation, case_source: bytes, expected_uid: str,
         with contained_output():
             return authenticate(result.evidence, record[1]) is True
 
+    def contained(fn, *args):
+        # Callback output is discarded; refusals below stay this adapter's own.
+        with contained_output():
+            return fn(*args)
+
+    def reconcile_original(ctx):
+        """Read-only confirmation of a restarted/lost-receipt add; never replays or undoes."""
+        o, attempt = ctx.original.invocation, ctx.attempt
+        try:
+            if (o.operation != 'run' or attempt.operation != 'reconcile'
+                    or o.attempt_id == attempt.attempt_id
+                    or not IDENTIFIER.fullmatch(o.attempt_id)
+                    or not IDENTIFIER.fullmatch(attempt.attempt_id)):
+                raise Refused('original intent not separate from reconcile attempt')
+            for inv in (o, attempt):
+                if replace(inv, attempt_id='', operation='admission') != admitted:
+                    raise Refused('invocation differs from admitted case')
+            if ctx.original.source != reviewed_source or source_digest(check_config_digest) != reviewed_source:
+                raise Refused('original source differs')
+            try:
+                found = list(contained(holder, o.attempt_id))
+            except Exception:
+                raise Refused('original-attempt holder unavailable') from None
+            if len(found) != 1:
+                raise Refused('original record missing or ambiguous')
+            ref, raw = found[0]
+            if type(raw) is not bytes or contained(authenticate, ref, raw) is not True:
+                raise Refused('original record unauthenticated')
+            record = json.loads(raw)
+            if (record.get('schema') != 'netkingdom.keycape-registration.attempt.v1'
+                    or record.get('invocation') != json.loads(json.dumps(asdict(o)))
+                    or record.get('source') != reviewed_source):
+                raise Refused('original record belongs to another attempt/case/source/principal')
+            if record.get('outcome') != Outcome.SUCCESS.value or record.get('created') != ['keycape-client']:
+                raise Refused('original record is not created-only provenance')
+            secret = contained(observe)
+            meta = secret['metadata']
+            if {'uid': meta['uid'], 'resource_version': meta['resourceVersion']} != record['after']:
+                raise Refused('native revision changed since creation')
+            if contained(helper.replacement, secret, client)[1] is not False:
+                raise Refused('exact client entry absent')
+        finally:
+            with contained_output():
+                revocation = revoke_session(attempt)
+        if not isinstance(revocation, str) or not IDENTIFIER.fullmatch(revocation):
+            raise Refused('session revocation evidence missing')
+        payload = json.dumps({'schema': 'netkingdom.keycape-registration.reconcile.v1',
+            'invocation': asdict(attempt), 'original': ref, 'source': reviewed_source,
+            'outcome': Outcome.SUCCESS.value, 'created': ('keycape-client',),
+            'after': record['after'], 'session_revocation': revocation},
+            sort_keys=True, separators=(',', ':')).encode()
+        with contained_output():
+            evidence = publish(payload)
+        if not isinstance(evidence, str) or not IDENTIFIER.fullmatch(evidence) or evidence in records:
+            raise Refused('owner evidence identifier invalid or reused')
+        result = OwnerResult(Outcome.SUCCESS, evidence, ('keycape-client',), True, revocation,
+                             attempt_id=attempt.attempt_id, operation='reconcile')
+        # Recorded for verify only; never a creation this process may roll back.
+        records[evidence] = (attempt, payload, result)
+        return result
+
     return Adapter(REFERENCE, reviewed_source, lambda inv: execute(inv, 'add'),
-                   verify, lambda inv, created: execute(inv, 'remove', created))
+                   verify, lambda inv, created: execute(inv, 'remove', created),
+                   reconcile_original=reconcile_original if holder is not None else None)

sso-mfa/k8s/keycape/registration-adapter.md

@@ -59,6 +59,25 @@ published producer source pins apply unchanged through actual BrowserInputs.
 This is a source compatibility fixture, not acknowledgment, owner provenance,
 normal native admission or deployment acceptance.
 
+## Original-intent reconciliation (NK-WP-0048-T15 source)
+
+`build(..., holder=, observe=)` takes two optional read-only capabilities,
+injected together or not at all. With them the Adapter installs
+`reconcile_original` for Warden's `ReconcileContext` (ops-warden `a45cf47`): the
+original `run` intent must be separate from the `reconcile` attempt and both must
+match the admitted case and reviewed source; `holder(original_attempt_id)` must
+return exactly one record that `authenticate` accepts, naming that invocation and
+source with a created-only `success`; and `observe()` must show the recorded
+post-creation uid/resourceVersion with the exact client entry. The parent session
+is revoked either way, and a `netkingdom.keycape-registration.reconcile.v1`
+record is published; `verify` accepts that result like any other owner record.
+It never adds or removes: holder outage, missing/ambiguous/unauthenticated or
+foreign records, a no-change original, and changed revision or entry all raise,
+so Warden keeps the step uncertain. Rollback after restart stays refused, since a
+reconciled creation is not a creation this process verified, until an owner
+rollback contract exists. Which authenticated holder backs `holder` is an owner
+gate; the tests use an in-memory stand-in only.
+
 Remaining work belongs in live work records before T12 closure:
 
 - Warden owns global evidence admission and exact native evidence/session

sso-mfa/k8s/keycape/test_keycape_registration_adapter.py

@@ -80,6 +80,15 @@ class RegistrationTests(unittest.TestCase):
             return copy.deepcopy(self.secret)
         raise AssertionError('unexpected tooling operation')
 
+    def test_original_reconcile_capabilities_are_optional_but_paired(self):
+        self.assertIsNone(a.build(**self.kw).reconcile_original)
+        for extra in ({'holder': lambda attempt: []}, {'observe': lambda: {}},
+                      {'holder': 'cmd', 'observe': lambda: {}}):
+            with self.subTest(extra=sorted(extra)), self.assertRaisesRegex(Refused, 'together'):
+                a.build(**dict(self.kw, **extra))
+        adapter = a.build(**dict(self.kw, holder=lambda attempt: [], observe=lambda: {}))
+        self.assertTrue(callable(adapter.reconcile_original))
+
     def test_actual_add_and_created_only_rollback_preserve_original_bytes(self):
         adapter = a.build(**self.kw)
         original = copy.deepcopy(self.secret['data'])

sso-mfa/k8s/keycape/test_registration_original_intent_reconcile.py

@@ -2,7 +2,8 @@
 adapter against Warden's original-intent interface (ops-warden a45cf47:
 OriginalIntent, Journal.original_intent, Adapter.reconcile_original).
 
-Fixture-only. The original-attempt holder below is the test's own in-memory
+T15 source: reconciliation now runs through the adapter's own
+reconcile_original (build(holder=..., observe=...)). The original-attempt holder below is the test's own in-memory
 evidence store standing in for an owner-authenticated holder; Warden's intent
 row is Warden's dispatch record, not created-only provenance. Selecting the
 actual authenticated holder remains a T15 owner gate. Synthetic Kubernetes
@@ -27,74 +28,18 @@ spec.loader.exec_module(base)
 a = base.a
 
 
-def original_reconciler(adapter, admitted, client, *, holder, authenticate,
+def capture_reasons(adapter, reasons):
-                        observe, publish, revoke_session, reasons):
+    """Fixture-only: Warden contains the owner's refusal, so keep the message."""
-    """Fixture owner reconciler: read-only, never adds, never removes.
+    reconcile = adapter.reconcile_original
 
-    holder(original_attempt_id) -> [(evidence_ref, payload_bytes)] is the
+    def wrapped(ctx):
-    authenticated original-attempt readback; observe() -> current secret.
-    Confirms creation only from one authenticated original created-only record
-    whose post-creation uid/resourceVersion and exact client entry still hold.
-    Anything else raises, so Warden keeps the step uncertain.
-    """
-    confirmed = {}
-
-    def reconcile(ctx):
         try:
-            return confirm(ctx)
+            return reconcile(ctx)
-        except Exception as exc:  # Warden contains the message; keep it for fixtures
+        except Exception as exc:
             reasons.append(f'{type(exc).__name__}: {exc}')
             raise
 
-    def confirm(ctx):
+    return replace(adapter, reconcile_original=wrapped)
-        o, attempt = ctx.original.invocation, ctx.attempt
-        try:
-            if (o.operation != 'run' or attempt.operation != 'reconcile'
-                    or o.attempt_id == attempt.attempt_id
-                    or not IDENTIFIER.fullmatch(o.attempt_id)):
-                raise Refused('original intent not separate from reconcile attempt')
-            for inv in (o, attempt):
-                if replace(inv, attempt_id='', operation='admission') != admitted:
-                    raise Refused('invocation differs from admitted case')
-            if ctx.original.source != adapter.source:
-                raise Refused('original source differs')
-            try:
-                records = holder(o.attempt_id)
-            except Exception:
-                raise Refused('original-attempt holder unavailable') from None
-            if len(records) != 1:
-                raise Refused('original record missing or ambiguous')
-            ref, raw = records[0]
-            if authenticate(ref, raw) is not True:
-                raise Refused('original record unauthenticated')
-            record = json.loads(raw)
-            if (record.get('schema') != 'netkingdom.keycape-registration.attempt.v1'
-                    or record.get('invocation') != json.loads(json.dumps(base.asdict(o)))
-                    or record.get('source') != adapter.source):
-                raise Refused('original record belongs to another attempt/case/source/principal')
-            if record.get('outcome') != Outcome.SUCCESS.value or record.get('created') != ['keycape-client']:
-                raise Refused('original record is not created-only provenance')
-            secret = observe()
-            meta = secret['metadata']
-            if {'uid': meta['uid'], 'resource_version': meta['resourceVersion']} != record['after']:
-                raise Refused('native revision changed since creation')
-            if a.helper.replacement(secret, client)[1] is not False:
-                raise Refused('exact client entry absent')
-        finally:
-            revocation = revoke_session(attempt)
-        evidence = publish(json.dumps({'schema': 'netkingdom.keycape-registration.reconcile.v1',
-            'original': ref, 'attempt': attempt.attempt_id}, sort_keys=True).encode())
-        result = OwnerResult(Outcome.SUCCESS, evidence, ('keycape-client',), True, revocation,
-                             attempt_id=attempt.attempt_id, operation='reconcile')
-        confirmed[evidence] = (attempt, result)
-        return result
-
-    def verify(inv, result):
-        if confirmed.get(getattr(result, 'evidence', None)) == (inv, result):
-            return True
-        return adapter.verify(inv, result)
-
-    return replace(adapter, reconcile_original=reconcile, verify=verify)
 
 
 class OriginalIntentReconcileTests(unittest.TestCase):
@@ -116,14 +61,10 @@ class OriginalIntentReconcileTests(unittest.TestCase):
         """Fresh process: new journal handle, new adapter (empty records), readmission."""
         executor.journal.db.close()
         self.reasons = []
+        # The adapter's own reconcile_original path, given test-only holder/readback.
+        self.kw = dict(self.kw, holder=self.holder, observe=lambda: copy.deepcopy(self.secret))
         fresh, admitted = self.kernel_fixture()
-        adapter = fresh.adapters[a.REFERENCE]
+        fresh.adapters[a.REFERENCE] = capture_reasons(fresh.adapters[a.REFERENCE], self.reasons)
-        client = a.helper.client_from_case(base.yaml.safe_load(self.case_source(admitted)))
-        fresh.adapters[a.REFERENCE] = original_reconciler(
-            adapter, admitted[1], client, holder=self.holder,
-            authenticate=lambda ref, raw: self.evidence.get(ref) == raw,
-            observe=lambda: copy.deepcopy(self.secret), publish=self.publish,
-            revoke_session=self.revoke, reasons=self.reasons)
         return fresh, admitted
 
     def case_source(self, admitted):