diff --git a/docs/reviews/2026-10-11-key39-lifecycle-v3-netkingdom-review.json b/docs/reviews/2026-10-11-key39-lifecycle-v3-netkingdom-review.json
new file mode 100644
index 0000000..6937be3
--- /dev/null
+++ b/docs/reviews/2026-10-11-key39-lifecycle-v3-netkingdom-review.json
@@ -0,0 +1,53 @@
+{
+  "reviewer": "net-kingdom-owner (agent net-kingdom, codex)",
+  "decision": "changes_requested",
+  "review_id": "NK-REVIEW-20261011-KEY39-LIFECYCLE-V3",
+  "work_item_id": "NK-WP-ADHOC-2026-10-11-T07",
+  "reviewed_proposal_id": "KEY-WP-0039-T02/attended-lifecycle-v3",
+  "reviewed_manifest_sha256": "fbc31fda2b6194747547b9fbe876095914c3d53231c79abd43a010dbb3273efa",
+  "reviewed_executor_sha256": "1161b3666ebf1f3eb322c7d5112584bce7e074da7e7650c5840e3c5de9311194",
+  "reviewed_tests_sha256": "75e26a9be4422be539f7e0a4d46c3dd6d8e2eab21d3cb6e4ba03cfe8c7c1b9cd",
+  "reviewed_runner_sha256": "bb9610babf44a959342bac700bb5712e97f400b44c4d0659d21df4426e616e19",
+  "previous_review_sha256": "f7293bc3cc896e23840d767bf86bd379f202a665f6a334485081831ad067d262",
+  "comment": "NetKingdom INTENT Taxonomy declaration and Principles 6/7 require bounded exposure with owner-held execution. Same completed work item NK-WP-ADHOC-2026-10-11-T07 independently reviews frozen v3 manifest fbc31fda2b6194747547b9fbe876095914c3d53231c79abd43a010dbb3273efa. V2 findings are resolved, but the new aggregate cleanup bound bypass below needs correction. This is source judgement only, not founder approval.",
+  "v2_findings_status": {
+    "NK-KEY39-V2-F01": "resolved: strict finite nonbool expiry and observation validation, JSON constants refused, expiry limited to maxsession",
+    "NK-KEY39-V2-F02": "resolved: total SIGALRM request guard clips wall and monotonic deadline; postcompletion recheck; known late accepted session revoked privately; unknown partial response retained"
+  },
+  "findings": [
+    {
+      "id": "NK-KEY39-V3-F01",
+      "severity": "medium",
+      "location": "openbao-negative-session.py:delete_temporary",
+      "counterexample": "With CLEANING=True and CLEANUP_MONOTONIC_DEADLINE=time.monotonic()-1, mocked subprocess.run records one Secret DELETE; only subsequent guarded k absence GET raises cleanup_budget_expired.",
+      "impact": "The final native Secret deletion bypasses the promised nonrenewable aggregate cleanup deadline and its remaining-time cap.",
+      "requested_change": "Route the exact UID/resourceVersion-precondition raw DeleteOptions request through guarded k or equivalent precheck/remaining-budget/total-timeout/postcheck. Add expired-budget no-call and near-expiry total-duration regressions."
+    }
+  ],
+  "validation": {
+    "supplied_tests": 39,
+    "independent_tests": 15,
+    "all_passed": true,
+    "counterexamples_reproduced": true
+  },
+  "coverage": [
+    "Prior eleven independent checks adapted to v3; v2 counterexamples now rejected",
+    "Real SIGALRM slow-body interruption and signal/timer restoration",
+    "Partial accepted login timeout records only unknown-outcome boolean",
+    "Aggregate cleanup is nonrenewable and guarded k refuses after expiry",
+    "Direct final delete bypass reproduced independently",
+    "CAS/source/runtime pins, temporary registration/environment isolation, output suppression, normal cleanup and hard-interruption recovery inspected unchanged"
+  ],
+  "authority_boundary": {
+    "source_preparation_only": true,
+    "native_actions": false,
+    "owner_checkout_writes": false,
+    "credential_access": false,
+    "secret_reads": false,
+    "foreign_reviews": false,
+    "founder_approval": false,
+    "native_acceptance": false,
+    "KEY39_task_closure": false
+  },
+  "work_record_anchor_sha256": "aa56b9cd64301f031cedd3cbd703e1698605fe97b0fee7b9ba45343333ff54f5"
+}
diff --git a/docs/reviews/2026-10-11-key39-lifecycle-v4-netkingdom-review.json b/docs/reviews/2026-10-11-key39-lifecycle-v4-netkingdom-review.json
new file mode 100644
index 0000000..6d3ee2d
--- /dev/null
+++ b/docs/reviews/2026-10-11-key39-lifecycle-v4-netkingdom-review.json
@@ -0,0 +1,64 @@
+{
+  "reviewer": "net-kingdom-owner (agent net-kingdom, codex)",
+  "decision": "approved",
+  "review_id": "NK-REVIEW-20261011-KEY39-LIFECYCLE-V4",
+  "work_item_id": "NK-WP-ADHOC-2026-10-11-T07",
+  "reviewed_proposal_id": "KEY-WP-0039-T02/attended-lifecycle-v4",
+  "reviewed_manifest_sha256": "d079f85c47d5c2635e1ee633ea4d1346b1737ce2be9ca273f22a38e620f418d8",
+  "reviewed_source_hashes": {
+    "sso-mfa/k8s/keycape/openbao-negative-session.py": "b4687c2d07819c43b45d0cddcad0488b62eb49175b2a06a43f7806a0838d3a4d",
+    "sso-mfa/k8s/keycape/test_openbao_negative_session.py": "ea2b5f7466c1da016f528093daf168c274238d4803eb5424d7353c66adbecf43"
+  },
+  "reviewed_runner_sha256": "bb9610babf44a959342bac700bb5712e97f400b44c4d0659d21df4426e616e19",
+  "previous_review_sha256": "06ac6616e75eb7690836fbce94b30ebecd3a674a3b256118d180ed13c90db3b1",
+  "original_v2_review_sha256": "f7293bc3cc896e23840d767bf86bd379f202a665f6a334485081831ad067d262",
+  "comment": "NetKingdom INTENT Taxonomy declaration, Principles 6/7 and SECURITY-COMPANION sections 3/7 support this bounded owner-held source preparation. Exact frozen v4 manifest d079f85c47d5c2635e1ee633ea4d1346b1737ce2be9ca273f22a38e620f418d8 reviewed under same completed own work item NK-WP-ADHOC-2026-10-11-T07. Technical source agreement: all independently reproduced v2/v3 gaps resolve. This owner source judgement supplies neither human approval nor native execution authority/acceptance and does not close KEY39 T02.",
+  "findings": [],
+  "prior_findings_status": {
+    "NK-KEY39-V2-F01": "resolved: finite nonbool approval and role observation values; integer bounded session; nonfinite JSON constants refused",
+    "NK-KEY39-V2-F02": "resolved: total Linux main-thread SIGALRM bound across connect/headers/body; finite per-request caps clipped to wall and monotonic action deadlines; postcompletion check and bounded late-session revoke",
+    "NK-KEY39-V3-F01": "resolved: final UID/resourceVersion raw DeleteOptions now uses guarded k; expired aggregate issues no delete and slow delete is clipped to remaining time"
+  },
+  "independent_coverage": [
+    "Exact manifest/executor/test/runner pins independently verified; originals unchanged",
+    "CAS patch tests require UID and resourceVersion; both dry runs and baseline UID/spec/config pins inspected",
+    "Six temporary registrations/env refs and separate Secret with no production credential replacement inspected; single replica Recreate causes two intended restarts",
+    "Captured subprocess stdout/stderr, fixed refusal messages, private stdin, value-free receipts and exclusive receipt creation inspected; synthetic raw error suppression verified",
+    "Normal cleanup refuses concurrent config/spec drift, restores baseline/readback, proves precise removed-client denial then UID/resourceVersion-precondition delete",
+    "Hard-interruption recovery requires separate recovery-approved envelope, reconstructs exact registry/spec baseline, refuses partial registrations/env and baseline drift; uncertain creation UID needs independent owner acknowledgement; issued OpenBao sessions require separate reconciliation",
+    "Fixed token/discovery/login/revoke allowlist, no ambient proxy/redirect, response close and byte bound inspected; unreviewed paths/methods refuse",
+    "Future/stale role observations now refuse; monotonic probe cleanup reserve inspected; two residual deadline counterexamples reproduced",
+    "Real elapsed slow HTTP interruption and SIGALRM timer/handler restoration",
+    "Late full accepted OpenBao session revoked privately then result refused; partial accepted response timeout retains value-free unknown-outcome evidence",
+    "Cleanup aggregate deadline starts once and cannot renew; guarded k refuses after expiry; exact final delete now shares it",
+    "Real elapsed slow deletion clipped to remaining cleanup budget; approved runtime/source pins and precise Secret deletion identity preserved"
+  ],
+  "validation": {
+    "supplied_lifecycle_tests": 41,
+    "unchanged_reviewed_runner_tests": 17,
+    "independent_tests": 17,
+    "all_passed": true,
+    "network_used": false,
+    "kubernetes_used": false
+  },
+  "limitations": [
+    "Linux main-thread setitimer required; preexisting timer or worker thread refuses",
+    "Fixture checks establish source behavior only; no native role/policy, headroom, runtime readiness or successful cleanup observed",
+    "Owner-held token producer remains unresolved; no credential route authorized",
+    "Unknown/failed session revocation requires separate owner reconciliation; cleanup alone is not revocation proof",
+    "Founder exact source/runtime/window approval and each independent owner acknowledgment remain necessary before native execution"
+  ],
+  "authority_boundary": {
+    "source_preparation_only": true,
+    "native_actions": false,
+    "owner_checkout_writes": false,
+    "credential_access": false,
+    "secret_reads": false,
+    "db_access": false,
+    "foreign_reviews": false,
+    "founder_approval": false,
+    "native_acceptance": false,
+    "KEY39_task_closure": false
+  },
+  "work_record_anchor_sha256": "3650c1eeadd74e150e528766523a284b03d02004749c04e9b6518d42f3e32531"
+}
diff --git a/docs/reviews/2026-10-11-key39-lifecycle-v5-netkingdom-review.json b/docs/reviews/2026-10-11-key39-lifecycle-v5-netkingdom-review.json
new file mode 100644
index 0000000..88eac1d
--- /dev/null
+++ b/docs/reviews/2026-10-11-key39-lifecycle-v5-netkingdom-review.json
@@ -0,0 +1,71 @@
+{
+  "reviewer": "net-kingdom-owner (agent net-kingdom, codex)",
+  "decision": "approved",
+  "review_id": "NK-REVIEW-20261011-KEY39-LIFECYCLE-V5",
+  "work_item_id": "NK-WP-ADHOC-2026-10-11-T07",
+  "reviewed_proposal_id": "KEY-WP-0039-T02/attended-lifecycle-v5",
+  "reviewed_manifest_sha256": "d1f997624db46693a7126a6eb6435f6b3a353d5bd67dcc8a539683e4b7147285",
+  "reviewed_source_hashes": {
+    "sso-mfa/k8s/keycape/openbao-negative-session.py": "3bcceebc845b17492040c76223d77088257c547a2b890198e7b5c2b04a0cd360",
+    "sso-mfa/k8s/keycape/test_openbao_negative_session.py": "a428a4a5ea60bd6c4ad10481e133227ad696a4a08653bf39b554b47431fb9f8c"
+  },
+  "reviewed_runner_sha256": "bb9610babf44a959342bac700bb5712e97f400b44c4d0659d21df4426e616e19",
+  "previous_review_sha256": "4f4498203a31bd46dbc3715c0df1cbb9eee63bed5b2c1baf430ff3c4a37c13cb",
+  "original_v2_review_sha256": "f7293bc3cc896e23840d767bf86bd379f202a665f6a334485081831ad067d262",
+  "comment": "NetKingdom INTENT Taxonomy declaration, Principles 6/7 and SECURITY-COMPANION sections 3/7 support bounded owner-held source preparation. Exact frozen v5 manifest d1f997624db46693a7126a6eb6435f6b3a353d5bd67dcc8a539683e4b7147285 reviewed under same completed own work item NK-WP-ADHOC-2026-10-11-T07. Technical source agreement includes bounded exceptional HTTP response cleanup and preserves all earlier failed/revised owner evidence. This owner judgement supplies neither human approval nor native execution authority/acceptance and does not close KEY39 T02.",
+  "findings": [],
+  "prior_findings_status": {
+    "NK-KEY39-V2-F01": "resolved: finite nonbool approval and role observation values; integer bounded session; nonfinite JSON constants refused",
+    "NK-KEY39-V2-F02": "resolved: total Linux main-thread SIGALRM bound across connect/headers/body; finite per-request caps clipped to wall and monotonic action deadlines; postcompletion check and bounded late-session revoke",
+    "NK-KEY39-V3-F01": "resolved: final UID/resourceVersion raw DeleteOptions now uses guarded k; expired aggregate issues no delete and slow delete is clipped to remaining time",
+    "Platform-v4-exceptional-close": "resolved in exact v5: normal close remains within request timer; one exceptional close uses at most 20 ms clipped to positive original remainder and finite nonrenewable cleanup remainder; expired cleanup performs no close; abort failure remains unconfirmed, never successful proof"
+  },
+  "independent_coverage": [
+    "Exact manifest/executor/test/runner pins independently verified; originals unchanged",
+    "CAS patch tests require UID and resourceVersion; both dry runs and baseline UID/spec/config pins inspected",
+    "Six temporary registrations/env refs and separate Secret with no production credential replacement inspected; single replica Recreate causes two intended restarts",
+    "Captured subprocess stdout/stderr, fixed refusal messages, private stdin, value-free receipts and exclusive receipt creation inspected; synthetic raw error suppression verified",
+    "Normal cleanup refuses concurrent config/spec drift, restores baseline/readback, proves precise removed-client denial then UID/resourceVersion-precondition delete",
+    "Hard-interruption recovery requires separate recovery-approved envelope, reconstructs exact registry/spec baseline, refuses partial registrations/env and baseline drift; uncertain creation UID needs independent owner acknowledgement; issued OpenBao sessions require separate reconciliation",
+    "Fixed token/discovery/login/revoke allowlist, no ambient proxy/redirect, response close and byte bound inspected; unreviewed paths/methods refuse",
+    "Future/stale role observations now refuse; monotonic probe cleanup reserve inspected; two residual deadline counterexamples reproduced",
+    "Real elapsed slow HTTP interruption and SIGALRM timer/handler restoration",
+    "Late full accepted OpenBao session revoked privately then result refused; partial accepted response timeout retains value-free unknown-outcome evidence",
+    "Cleanup aggregate deadline starts once and cannot renew; guarded k refuses after expiry; exact final delete now shares it",
+    "Real elapsed slow deletion clipped to remaining cleanup budget; approved runtime/source pins and precise Secret deletion identity preserved",
+    "Independent early RuntimeError read plus slow exceptional close; fixed error and no protected text in receipt",
+    "Independent timed-out body plus slow abort cap; real elapsed timer restores handler and timer",
+    "Independent normal-close error on known accepted session; private bounded self-revoke then abort failure",
+    "Independent expired aggregate exceptional-close zero call/no deadline renewal; prior guard suite adapted only for newly added value-free abort receipt booleans"
+  ],
+  "validation": {
+    "supplied_lifecycle_tests": 46,
+    "unchanged_reviewed_runner_tests": 17,
+    "independent_tests": 21,
+    "all_passed": true,
+    "network_used": false,
+    "kubernetes_used": false,
+    "review_test_adaptation": "Earlier receipt-equality assertion expected only unknown-outcome flag; v5 legitimately adds abort booleans. Updated assertion verifies those booleans and value-free shape; no source defect inferred."
+  },
+  "limitations": [
+    "Linux main-thread setitimer required; preexisting timer or worker thread refuses",
+    "Fixture checks establish source behavior only; no native role/policy, headroom, runtime readiness or successful cleanup observed",
+    "Owner-held token producer remains unresolved; no credential route authorized",
+    "Unknown/failed session revocation requires separate owner reconciliation; cleanup alone is not revocation proof",
+    "Founder exact source/runtime/window approval and each independent owner acknowledgment remain necessary before native execution"
+  ],
+  "authority_boundary": {
+    "source_preparation_only": true,
+    "native_actions": false,
+    "owner_checkout_writes": false,
+    "credential_access": false,
+    "secret_reads": false,
+    "db_access": false,
+    "foreign_reviews": false,
+    "founder_approval": false,
+    "native_acceptance": false,
+    "KEY39_task_closure": false
+  },
+  "work_record_anchor_sha256": "17c1e1a798516509d6ddb4d5c93090910cc6d3dd7b516827abf4fd3031ef80b3",
+  "platform_previous_review_sha256": "c0f9da4850d7b6ce8a41401bfb7cddc39a60cd7bc8767de89d76b5f93198a2d9"
+}
diff --git a/sso-mfa/k8s/keycape/openbao-negative-session.py b/sso-mfa/k8s/keycape/openbao-negative-session.py
new file mode 100644
index 0000000..6159cdb
--- /dev/null
+++ b/sso-mfa/k8s/keycape/openbao-negative-session.py
@@ -0,0 +1,712 @@
+#!/usr/bin/env python3
+"""Founder-attended KEY39 lifecycle. Silent; private values never leave memory/K8s.
+
+inspect is metadata-only (no Secret read). execute requires an exact approved
+envelope, source/runtime pins and founder-held private tokens on stdin. No login,
+custody fetch, production credential, role or policy mutation is implemented.
+"""
+import argparse
+import base64
+import copy
+import hashlib
+import importlib.util
+import json
+import math
+import os
+from pathlib import Path
+import secrets
+import signal
+import subprocess
+import threading
+import time
+from contextlib import contextmanager
+import urllib.parse
+import urllib.request
+import urllib.error
+
+import yaml
+
+CLUSTER = 'a553c742-0115-43d4-99a4-a5ca56fe0786'
+TARGET = 'sso/keycape'
+PROPOSAL = 'KEY-WP-0039-T02/attended-lifecycle-v5'
+IDS = ['openbao-probe-' + c for c in ['aud', 'sub', 'tenant', 'roles', 'scope', 'expiry']]
+ENVS = ['KEYCAPE_' + c.upper().replace('-', '_') + '_CLIENT_SECRET' for c in IDS]
+TEMP = 'keycape-openbao-negative-key39'
+BAO_ORIGINS = {'http://127.0.0.1:18200', 'http://openbao.openbao.svc:8200'}
+REQUEST_DEADLINE = None
+BUSINESS_MONOTONIC_DEADLINE = None
+CLEANUP_MONOTONIC_DEADLINE = None
+CLEANING = False
+MAX_RESPONSE = 256 * 1024
+HTTP_TOTAL_SECONDS = 15
+CLEANUP_TOTAL_SECONDS = 300
+HTTP_ABORT_SECONDS = 0.02
+
+
+class Refused(Exception):
+    pass
+
+
+def require(ok, reason):
+    if not ok:
+        raise Refused(reason)
+
+
+def digest(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def canonical(value):
+    return digest(json.dumps(value, sort_keys=True, separators=(',', ':')).encode())
+
+
+def finite_number(value):
+    try:
+        return type(value) in (int, float) and math.isfinite(value)
+    except OverflowError:
+        return False
+
+
+def read_json(raw):
+    def reject_constant(_):
+        raise Refused('nonfinite_json_constant')
+    return json.loads(raw, parse_constant=reject_constant)
+
+
+def begin_cleanup():
+    global CLEANUP_MONOTONIC_DEADLINE
+    if CLEANUP_MONOTONIC_DEADLINE is None:
+        CLEANUP_MONOTONIC_DEADLINE = time.monotonic() + CLEANUP_TOTAL_SECONDS
+
+
+def check_deadline():
+    if CLEANING:
+        require(CLEANUP_MONOTONIC_DEADLINE is not None
+                and time.monotonic() < CLEANUP_MONOTONIC_DEADLINE, 'cleanup_budget_expired')
+        return
+    if REQUEST_DEADLINE is not None and not CLEANING:
+        require(finite_number(REQUEST_DEADLINE), 'approved_window_invalid')
+        require(time.time() < REQUEST_DEADLINE, 'approved_window_expired')
+    if BUSINESS_MONOTONIC_DEADLINE is not None:
+        require(time.monotonic() < BUSINESS_MONOTONIC_DEADLINE, 'session_business_budget_expired')
+
+
+def request_budget(cap, cleanup=False):
+    require(finite_number(cap) and 0 < cap <= 30, 'request_timeout_invalid')
+    budget = float(cap)
+    if cleanup or CLEANING:
+        if CLEANUP_MONOTONIC_DEADLINE is not None:
+            budget = min(budget, CLEANUP_MONOTONIC_DEADLINE - time.monotonic())
+    else:
+        check_deadline()
+        if REQUEST_DEADLINE is not None:
+            budget = min(budget, REQUEST_DEADLINE - time.time())
+        if BUSINESS_MONOTONIC_DEADLINE is not None:
+            budget = min(budget, BUSINESS_MONOTONIC_DEADLINE - time.monotonic())
+    require(budget > 0 and math.isfinite(budget), 'request_window_exhausted')
+    return budget
+
+
+@contextmanager
+def total_timeout(seconds):
+    """Linux main-thread hard elapsed bound, including blocking headers/body reads."""
+    require(finite_number(seconds) and 0 < seconds <= 30, 'request_timeout_invalid')
+    require(threading.current_thread() is threading.main_thread()
+            and hasattr(signal, 'setitimer'), 'main_thread_linux_timer_required')
+    previous_timer = signal.getitimer(signal.ITIMER_REAL)
+    require(previous_timer == (0.0, 0.0), 'existing_alarm_timer_refused')
+    previous_handler = signal.getsignal(signal.SIGALRM)
+    def expired(*_):
+        raise Refused('request_total_timeout')
+    started = time.monotonic()
+    signal.signal(signal.SIGALRM, expired)
+    signal.setitimer(signal.ITIMER_REAL, seconds)
+    try:
+        yield
+        require(time.monotonic() - started <= seconds, 'request_total_timeout')
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous_handler)
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs):
+        return None
+
+
+class SessionTransport:
+    """Fixed origins, no ambient proxy, bounded reads, same guards in cleanup."""
+    def __init__(self, bao_origin, receipt=None):
+        require(bao_origin in BAO_ORIGINS, 'bao_origin_invalid')
+        self.bao_origin = bao_origin
+        self.receipt = receipt if receipt is not None else {}
+        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
+
+    def request(self, url, *, data=None, headers=None, method='POST', timeout=15):
+        allowed = {'https://kc.coulomb.social/token',
+                   'https://kc.coulomb.social/.well-known/openid-configuration',
+                   self.bao_origin + '/v1/auth/keycape-services/login',
+                   self.bao_origin + '/v1/auth/token/revoke-self'}
+        require(url in allowed, 'http_target_not_reviewed')
+        require(method == ('GET' if url.endswith('/.well-known/openid-configuration') else 'POST'),
+                'http_method_not_reviewed')
+        cleanup = url.endswith('/v1/auth/token/revoke-self')
+        budget = request_budget(min(timeout, HTTP_TOTAL_SECONDS), cleanup=cleanup)
+        total_deadline = time.monotonic() + budget
+        req = urllib.request.Request(url, data=data, headers=headers or {}, method=method)
+        response, status, raw = None, None, None
+        try:
+            with total_timeout(budget):
+                try:
+                    response = self.opener.open(req, timeout=budget)
+                except urllib.error.HTTPError as error:
+                    response = error
+                except (urllib.error.URLError, OSError):
+                    if url == self.bao_origin + '/v1/auth/keycape-services/login':
+                        self.receipt['openbao_login_outcome_unknown'] = True
+                    return None, b''
+                status = response.getcode()
+                raw = response.read(MAX_RESPONSE + 1)
+                require(len(raw) <= MAX_RESPONSE, 'http_response_oversized')
+                response.close()
+                response = None
+            if not cleanup:
+                check_deadline()
+            return status, raw
+        except Refused:
+            if url == self.bao_origin + '/v1/auth/keycape-services/login' and status in (None, 200):
+                # A late response cannot pass; preserve acquired-token cleanup when available.
+                if raw is not None and status == 200:
+                    self.revoke_late_session(raw)
+                else:
+                    self.receipt['openbao_login_outcome_unknown'] = True
+            raise
+        except (urllib.error.URLError, OSError):
+            if url == self.bao_origin + '/v1/auth/keycape-services/login':
+                if raw is not None and status == 200:
+                    self.revoke_late_session(raw)
+                else:
+                    self.receipt['openbao_login_outcome_unknown'] = True
+            return None, b''
+        except Exception:
+            if url == self.bao_origin + '/v1/auth/keycape-services/login':
+                if raw is not None and status == 200:
+                    self.revoke_late_session(raw)
+                else:
+                    self.receipt['openbao_login_outcome_unknown'] = True
+            raise Refused('http_response_failed') from None
+        finally:
+            if response is not None:
+                self.abort_response(response, total_deadline)
+
+    def abort_response(self, response, original_deadline):
+        """One bounded exceptional close; never retry or reset the cleanup clock."""
+        self.receipt['http_abort_close_attempted'] = False
+        self.receipt['http_abort_close_succeeded'] = False
+        begin_cleanup()
+        try:
+            cap = request_budget(HTTP_ABORT_SECONDS, cleanup=True)
+            remaining = original_deadline - time.monotonic()
+            if remaining > 0:
+                cap = min(cap, remaining)
+            self.receipt['http_abort_close_attempted'] = True
+            with total_timeout(cap):
+                response.close()
+            self.receipt['http_abort_close_succeeded'] = True
+        except Exception:
+            self.receipt['http_response_abort_unconfirmed'] = True
+            raise Refused('http_response_abort_unconfirmed') from None
+
+    def revoke_late_session(self, raw):
+        self.receipt['late_acceptance_revoke_attempted'] = False
+        self.receipt['late_acceptance_revoke_succeeded'] = False
+        try:
+            token = read_json(raw)['auth']['client_token']
+            require(isinstance(token, str) and token, 'late_session_response_invalid')
+        except (KeyError, TypeError, ValueError, Refused):
+            self.receipt['openbao_login_outcome_unknown'] = True
+            return
+        begin_cleanup()
+        self.receipt['late_acceptance_revoke_attempted'] = True
+        try:
+            status, _ = self.post(self.bao_origin + '/v1/auth/token/revoke-self',
+                body={}, headers={'X-Vault-Token': token})
+            self.receipt['late_acceptance_revoke_succeeded'] = status == 204
+        except Exception:
+            self.receipt['late_acceptance_revoke_succeeded'] = False
+        finally:
+            del token
+
+    def post(self, url, form=None, body=None, headers=None, timeout=15):
+        headers = dict(headers or {})
+        if form is not None:
+            data = urllib.parse.urlencode(form).encode()
+            headers['Content-Type'] = 'application/x-www-form-urlencoded'
+        else:
+            data = json.dumps(body or {}).encode()
+            headers['Content-Type'] = 'application/json'
+        return self.request(url, data=data, headers=headers, timeout=timeout)
+
+    def discovery(self):
+        status, raw = self.request('https://kc.coulomb.social/.well-known/openid-configuration', method='GET')
+        require(status == 200, 'discovery_failed')
+        return json.loads(raw)
+
+
+def k(arguments, payload=None):
+    # Execute on railiance01 in the founder's session, never workstation defaults.
+    check_deadline()
+    budget = request_budget(30)
+    with total_timeout(budget):
+        result = subprocess.run(['kubectl', '--kubeconfig=/etc/rancher/k3s/k3s.yaml',
+                             '--request-timeout=20s', *arguments],
+                            input=None if payload is None else json.dumps(payload),
+                            text=True, capture_output=True, timeout=budget)
+    check_deadline()
+    require(result.returncode == 0, 'kubernetes_operation_failed')
+    return json.loads(result.stdout) if result.stdout.strip() else None
+
+
+def get(kind, name):
+    return k(['-n', 'sso', 'get', kind, name, '-o', 'json'])
+
+
+def meta(obj):
+    return {key: obj['metadata'][key] for key in ['uid', 'resourceVersion']}
+
+
+def patch(kind, obj, path, value, dry=False):
+    ops = [{'op': 'test', 'path': '/metadata/' + key, 'value': val}
+           for key, val in meta(obj).items()]
+    ops.append({'op': 'replace', 'path': path, 'value': value})
+    args = ['-n', 'sso', 'patch', kind, obj['metadata']['name'], '--type=json',
+            '--patch-file=/dev/stdin', '-o', 'json']
+    if dry:
+        args += ['--dry-run=server']
+    return k(args, ops)
+
+
+def registrations():
+    reference = {'displayName': 'Temporary KEY39 negative probe',
+                 'allowedScopes': ['openbao:login'], 'grantTypes': ['client_credentials'],
+                 'clientType': 'confidential', 'serviceSubject': 'service:secrets-engine',
+                 'tenant': 'tenant:platform', 'roles': ['secrets-engine'], 'tokenLifetime': '1m'}
+    result = []
+    for index, client_id in enumerate(IDS):
+        c = dict(reference, clientId=client_id, secretRef='env:' + ENVS[index])
+        if index:
+            c['audience'] = 'secrets-engine-openbao'
+        if index == 1:
+            c['serviceSubject'] = 'service:openbao-probe'
+        if index == 2:
+            c['tenant'] = 'tenant:openbao-probe'
+        if index == 3:
+            c['roles'] = ['openbao-probe']
+        if index == 4:
+            c['allowedScopes'] = ['openbao:probe']
+        result.append(c)
+    return result
+
+
+class UniqueLoader(yaml.SafeLoader):
+    pass
+
+
+def unique(loader, node):
+    result = {}
+    for key_node, value_node in node.value:
+        key = loader.construct_object(key_node)
+        require(key not in result, 'duplicate_yaml_key')
+        result[key] = loader.construct_object(value_node)
+    return result
+
+
+UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, unique)
+
+
+def candidate_config(raw):
+    config = yaml.load(raw, Loader=UniqueLoader)
+    clients = config['clients']
+    require(isinstance(clients, list) and clients, 'clients_required')
+    ids = [c['clientId'] for c in clients]
+    require(len(ids) == len(set(ids)) and not set(ids).intersection(IDS), 'existing_probe_or_duplicate')
+    require(not {c.get('secretRef') for c in clients}.intersection('env:' + e for e in ENVS),
+            'probe_secretref_collision')
+    production = [c for c in clients if c['clientId'] == 'secrets-engine-openbao']
+    require(len(production) == 1, 'production_client_missing')
+    expected = registrations()[-1]
+    for field in ['allowedScopes', 'grantTypes', 'clientType', 'serviceSubject', 'tenant', 'roles']:
+        require(production[0].get(field) == expected[field], 'production_claim_contract_drift')
+    require(production[0].get('audience', production[0]['clientId']) == 'secrets-engine-openbao',
+            'production_audience_drift')
+    nodes = yaml.compose(raw)
+    node = next(v for key, v in nodes.value if key.value == 'clients')
+    require(isinstance(node, yaml.nodes.SequenceNode) and not node.flow_style,
+            'block_client_sequence_required')
+    index = node.end_mark.index
+    start = raw.rfind('\n', 0, index) + 1
+    if not raw[start:index].strip():
+        index = start
+    fragment = yaml.safe_dump(registrations(), sort_keys=False)
+    fragment = ''.join(' ' * node.start_mark.column + line if line.strip() else line
+                       for line in fragment.splitlines(True))
+    if index and raw[index - 1] != '\n':
+        fragment = '\n' + fragment
+    updated = raw[:index] + fragment + raw[index:]
+    intended = copy.deepcopy(config)
+    intended['clients'].extend(registrations())
+    require(yaml.load(updated, Loader=UniqueLoader) == intended, 'unrelated_config_changed')
+    return updated
+
+
+def candidate_spec(spec):
+    result = copy.deepcopy(spec)
+    require(result.get('replicas', 1) == 1 and result['strategy']['type'] == 'Recreate',
+            'single_replica_recreate_required')
+    container = next(c for c in result['template']['spec']['containers'] if c['name'] == 'keycape')
+    env = container['env']
+    require(not {e['name'] for e in env}.intersection(ENVS), 'probe_env_collision')
+    env.extend({'name': e, 'valueFrom': {'secretKeyRef': {'name': TEMP, 'key': c}}}
+               for c, e in zip(IDS, ENVS))
+    return result
+
+
+def ready(spec, timeout=150):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        dep = get('deployment', 'keycape')
+        require(dep['spec'] == spec, 'deployment_drift_during_readiness')
+        status = dep.get('status', {})
+        if (status.get('observedGeneration') == dep['metadata']['generation']
+                and all(status.get(key) == 1 for key in
+                        ['updatedReplicas', 'readyReplicas', 'availableReplicas', 'replicas'])):
+            return True
+        time.sleep(3)
+    raise Refused('readiness_timeout')
+
+
+def verify_sources(envelope, recovery=False):
+    require(envelope.get('proposal_id') == PROPOSAL and envelope.get('target') == TARGET,
+            'proposal_target_mismatch')
+    require(envelope.get('disposition') == ('recovery-approved' if recovery else 'approved')
+            and envelope.get('founder_decision_id'),
+            'founder_exact_go_required')
+    require(envelope.get('source_hashes') and
+            str(Path(__file__).resolve()) in envelope['source_hashes'], 'executor_pin_required')
+    for filename, expected in envelope['source_hashes'].items():
+        require(digest(Path(filename).read_bytes()) == expected, 'reviewed_source_changed')
+    required = ['net-kingdom', 'key-cape', 'railiance-platform']
+    require(all(envelope.get('owner_acknowledgments', {}).get(owner) for owner in required),
+            'owner_acknowledgments_required')
+    expiry = envelope['valid_until_epoch']
+    maximum = envelope['maximum_session_seconds']
+    require(finite_number(expiry) and type(maximum) is int and
+            420 <= maximum <= 900 and 0 < expiry - time.time() <= maximum,
+            'attended_window_invalid')
+    if recovery:
+        return None
+    role = envelope['role_observation']
+    require(role['mount'] == 'keycape-services' and role['role'] == 'secrets-engine-openbao'
+            and role['issuer'] == 'https://kc.coulomb.social' and role['algorithm'] == 'RS256'
+            and role['bindings_exact'] is True and role['policy_exact'] is True,
+            'role_owner_observation_required')
+    leeway = role['raw_expiration_leeway']
+    require(type(leeway) is int and -1 <= leeway <= 150, 'leeway_outside_session_bound')
+    effective = 0 if leeway == -1 else leeway or 150
+    require(role['effective_expiration_leeway'] == effective, 'leeway_normalization_mismatch')
+    observed = role['observed_at_epoch']
+    require(finite_number(observed) and 0 <= time.time() - observed <= 300,
+            'role_observation_stale_or_future')
+    require(envelope['bao_addr'] in BAO_ORIGINS, 'bao_origin_invalid')
+    return effective
+
+
+def load_probe(path):
+    spec = importlib.util.spec_from_file_location('key39_reviewed_probe', path)
+    result = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(result)
+    return result
+
+
+def removed_client_denied(status, raw):
+    try:
+        body = json.loads(raw)
+        return status == 401 and body.get('error') == 'invalid_profile_usage' and body.get('feature') == 'client_id'
+    except (ValueError, AttributeError, TypeError):
+        return False
+
+
+def recovered_config(raw):
+    config = yaml.load(raw, Loader=UniqueLoader)
+    clients = config['clients']
+    ids = [c['clientId'] for c in clients]
+    present = set(ids).intersection(IDS)
+    require(len(ids) == len(set(ids)) and present in [set(), set(IDS)], 'partial_probe_registry')
+    if not present:
+        return raw
+    require(clients[-6:] == registrations(), 'recovery_registration_drift')
+    node = next(v for key, v in yaml.compose(raw).value if key.value == 'clients')
+    first = node.value[-6].start_mark.index
+    start = raw.rfind('\n', 0, first) + 1
+    end = node.end_mark.index
+    line_start = raw.rfind('\n', 0, end) + 1
+    if not raw[line_start:end].strip():
+        end = line_start
+    restored = raw[:start] + raw[end:]
+    require(candidate_config(restored) == raw, 'recovery_byte_roundtrip_failed')
+    return restored
+
+
+def recovered_spec(spec):
+    restored = copy.deepcopy(spec)
+    container = next(c for c in restored['template']['spec']['containers'] if c['name'] == 'keycape')
+    env = container['env']
+    expected = [{'name': e, 'valueFrom': {'secretKeyRef': {'name': TEMP, 'key': c}}}
+                for c, e in zip(IDS, ENVS)]
+    matches = [e for e in env if e['name'] in ENVS]
+    require(matches in [[], expected], 'partial_or_drifted_probe_env')
+    container['env'] = [e for e in env if e['name'] not in ENVS]
+    return restored
+
+
+def delete_temporary(created):
+    # The exact raw API delete needs the same aggregate/per-request guards as patches.
+    k(['delete', '--raw=/api/v1/namespaces/sso/secrets/' + TEMP, '-f', '/dev/stdin'],
+      {'apiVersion': 'v1', 'kind': 'DeleteOptions', 'preconditions': meta(created)})
+    require(k(['-n', 'sso', 'get', 'secret', TEMP, '--ignore-not-found', '-o', 'json']) is None,
+            'temporary_secret_still_present')
+
+
+def recover(envelope, receipt):
+    global REQUEST_DEADLINE, BUSINESS_MONOTONIC_DEADLINE, CLEANUP_MONOTONIC_DEADLINE, CLEANING
+    verify_sources(envelope, recovery=True)
+    # Separately approved recovery obligations are cleanup, never business writes.
+    REQUEST_DEADLINE = None
+    BUSINESS_MONOTONIC_DEADLINE = None
+    CLEANUP_MONOTONIC_DEADLINE = None
+    begin_cleanup()
+    CLEANING = True
+    transport = SessionTransport(envelope['bao_addr'], receipt=receipt)
+    require(k(['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER,
+            'cluster_identity_changed')
+    config = get('secret', 'keycape-config')
+    dep = get('deployment', 'keycape')
+    raw = base64.b64decode(config['data']['config.yaml'], validate=True).decode()
+    restored = recovered_config(raw)
+    spec = recovered_spec(dep['spec'])
+    require(config['metadata']['uid'] == envelope['config_metadata']['uid']
+            and dep['metadata']['uid'] == envelope['deployment_metadata']['uid']
+            and digest(restored.encode()) == envelope['registry_sha256']
+            and canonical(spec) == envelope['deployment_spec_sha256'], 'recovery_baseline_drift')
+    temp = get('secret', TEMP)
+    require(set(temp.get('data', {})) == set(IDS)
+            and temp['metadata']['uid'] == envelope['temporary_secret_uid'], 'recovery_custody_identity_required')
+    values = {c: base64.b64decode(temp['data'][c], validate=True).decode() for c in IDS}
+    if raw != restored:
+        patch('secret', config, '/data/config.yaml', base64.b64encode(restored.encode()).decode())
+    if dep['spec'] != spec:
+        patch('deployment', dep, '/spec', spec)
+    ready(spec)
+    after = get('secret', 'keycape-config')
+    require(after['data'] == dict(config['data'], **{'config.yaml': base64.b64encode(restored.encode()).decode()}),
+            'recovery_readback_failed')
+    require(envelope['runner_path'] in envelope['source_hashes'], 'runner_pin_required')
+    probe = load_probe(envelope['runner_path'])
+    for client, value in values.items():
+        basic = base64.b64encode((client + ':' + value).encode()).decode()
+        status, body = transport.post('https://kc.coulomb.social/token',
+            form={'grant_type': 'client_credentials', 'scope': 'openbao:login'},
+            headers={'Authorization': 'Basic ' + basic})
+        require(removed_client_denied(status, body), 'removed_client_denial_unconfirmed')
+    observed = get('secret', TEMP)
+    require(observed['data'] == temp['data'] and observed['metadata']['uid'] == temp['metadata']['uid'],
+            'recovery_temporary_secret_drift')
+    delete_temporary(observed)
+    receipt.update(cleanup_verified=True, recovery_only=True)
+
+
+def execute(envelope, private, receipt, checkpoint=lambda: None):
+    global REQUEST_DEADLINE, BUSINESS_MONOTONIC_DEADLINE, CLEANUP_MONOTONIC_DEADLINE, CLEANING
+    leeway = verify_sources(envelope)
+    REQUEST_DEADLINE = envelope['valid_until_epoch']
+    BUSINESS_MONOTONIC_DEADLINE = time.monotonic() + min(
+        REQUEST_DEADLINE - time.time(), envelope['maximum_session_seconds'] - 180)
+    CLEANUP_MONOTONIC_DEADLINE = None
+    CLEANING = False
+    runner_path = envelope['runner_path']
+    require(runner_path in envelope['source_hashes'], 'runner_pin_required')
+    probe = load_probe(runner_path)
+    transport = SessionTransport(envelope['bao_addr'], receipt=receipt)
+    discovery = transport.discovery()
+    require(discovery.get('issuer') == 'https://kc.coulomb.social' and
+            discovery.get('token_endpoint') == 'https://kc.coulomb.social/token',
+            'issuer_discovery_changed')
+    require(isinstance(private.get('human_token'), str) and private['human_token']
+            and isinstance(private.get('coding_agent_token'), str) and private['coding_agent_token'],
+            'founder_private_tokens_required')
+    require(k(['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER,
+            'cluster_identity_changed')
+    before = get('secret', 'keycape-config')
+    deployment = get('deployment', 'keycape')
+    require(not before['metadata'].get('ownerReferences'), 'controller_owned_config')
+    require(meta(before) == envelope['config_metadata']
+            and meta(deployment) == envelope['deployment_metadata']
+            and canonical(deployment['spec']) == envelope['deployment_spec_sha256'],
+            'approved_runtime_revision_changed')
+    raw = base64.b64decode(before['data']['config.yaml'], validate=True).decode()
+    require(digest(raw.encode()) == envelope['registry_sha256'], 'approved_registry_changed')
+    updated = base64.b64encode(candidate_config(raw).encode()).decode()
+    intended_data = dict(before['data'], **{'config.yaml': updated})
+    candidate = candidate_spec(deployment['spec'])
+    ready(deployment['spec'])
+    # Exact GET existence is tested via create (never adopt/delete an existing Secret).
+    values = {c: secrets.token_urlsafe(48) for c in IDS}
+    temporary = {'apiVersion': 'v1', 'kind': 'Secret', 'metadata': {'namespace': 'sso', 'name': TEMP},
+                 'type': 'Opaque', 'data': {c: base64.b64encode(v.encode()).decode() for c, v in values.items()}}
+    patch('secret', before, '/data/config.yaml', updated, dry=True)
+    patch('deployment', deployment, '/spec', candidate, dry=True)
+    k(['create', '-f', '/dev/stdin', '--dry-run=server', '-o', 'json'], temporary)
+    created = None
+    mutations_attempted = False
+    began = time.monotonic()
+    try:
+        # Set before API call: uncertain response must still trigger reconciliation.
+        check_deadline()
+        mutations_attempted = True
+        created = k(['create', '-f', '/dev/stdin', '-o', 'json'], temporary)
+        receipt['temporary_secret_uid'] = created['metadata']['uid']
+        checkpoint()
+        patch('secret', before, '/data/config.yaml', updated)
+        patch('deployment', deployment, '/spec', candidate)
+        ready(candidate)
+        require(get('secret', 'keycape-config')['data'] == intended_data, 'registry_drift')
+        require(time.monotonic() - began < 300, 'probe_start_too_late')
+        # Current admitted agent token stays founder-owned; no existing secret fetched.
+        original_exchange = probe.exchange
+        def exchange(post, endpoint, client, secret, scope):
+            if client == 'codex-railiance-platform':
+                return private['coding_agent_token']
+            return original_exchange(post, endpoint, client, secret, scope)
+        probe.exchange = exchange
+        supplied = dict(values, **{'codex-railiance-platform': 'founder-held-token'})
+        def bounded_post(*args, **kwargs):
+            if not args[0].endswith('/v1/auth/token/revoke-self'):
+                require(time.monotonic() - began < envelope['maximum_session_seconds'] - 180,
+                        'session_cleanup_reserve_reached')
+            return transport.post(*args, **kwargs)
+        def bounded_sleep(seconds):
+            while seconds > 0:
+                check_deadline()
+                require(time.monotonic() - began < envelope['maximum_session_seconds'] - 180,
+                        'session_cleanup_reserve_reached')
+                part = min(seconds, 3)
+                time.sleep(part)
+                seconds -= part
+        result = probe.run(bounded_post, supplied, 'https://kc.coulomb.social/token',
+                           envelope['bao_addr'], 'keycape-services', 'secrets-engine-openbao',
+                           human_token=private['human_token'], leeway=leeway, sleep=bounded_sleep)
+        receipt['negative_probe'] = result
+        checkpoint()
+        check_deadline()
+        require(time.monotonic() - began <= envelope['maximum_session_seconds'], 'session_window_exceeded')
+        require(result['all_refused'], 'negative_proof_incomplete')
+    finally:
+        if mutations_attempted:
+            begin_cleanup()
+            CLEANING = True
+            # Never restore over an unrelated edit; keep custody for owner recovery on drift.
+            now = get('secret', 'keycape-config')
+            require(now['metadata']['uid'] == before['metadata']['uid'] and
+                    now['data'] in [before['data'], intended_data], 'cleanup_registry_drift')
+            if now['data'] == intended_data:
+                patch('secret', now, '/data/config.yaml', before['data']['config.yaml'])
+            now = get('deployment', 'keycape')
+            require(now['metadata']['uid'] == deployment['metadata']['uid'] and
+                    now['spec'] in [deployment['spec'], candidate], 'cleanup_deployment_drift')
+            if now['spec'] == candidate:
+                patch('deployment', now, '/spec', deployment['spec'])
+            ready(deployment['spec'])
+            require(get('secret', 'keycape-config')['data'] == before['data'], 'cleanup_readback_failed')
+            # No fresh JWT is minted: deleted clients must fail at the issuer.
+            for client in IDS:
+                basic = base64.b64encode((client + ':' + values[client]).encode()).decode()
+                status, body = transport.post('https://kc.coulomb.social/token',
+                    form={'grant_type': 'client_credentials', 'scope': 'openbao:login'},
+                    headers={'Authorization': 'Basic ' + basic})
+                require(removed_client_denied(status, body),
+                        'removed_client_denial_unconfirmed')
+            if created is None:
+                # A timed-out create may have succeeded; compare generated bytes before delete.
+                observed = get('secret', TEMP)
+                require(observed.get('data') == temporary['data'], 'temporary_secret_identity_unconfirmed')
+                created = observed
+            observed = get('secret', TEMP)
+            require(observed['metadata']['uid'] == created['metadata']['uid']
+                    and observed.get('data') == temporary['data'], 'temporary_secret_drift')
+            delete_temporary(observed)
+            receipt['cleanup_verified'] = True
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument('mode', choices=['inspect', 'founder-preflight', 'execute', 'recover'])
+    ap.add_argument('--envelope')
+    ap.add_argument('--confirm')
+    ap.add_argument('--receipt', required=True)
+    args = ap.parse_args()
+    receipt = {'proposal_id': PROPOSAL, 'target': TARGET, 'native_accepted': False,
+               'values_emitted': False, 'cleanup_verified': False}
+    # Do not truncate or follow a link; protected receipts cannot overwrite credentials.
+    fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+    def checkpoint():
+        os.lseek(fd, 0, os.SEEK_SET)
+        os.ftruncate(fd, 0)
+        os.write(fd, json.dumps(receipt, sort_keys=True, indent=2).encode())
+        os.fsync(fd)
+    rc = 1
+    try:
+        if args.mode in ['inspect', 'founder-preflight']:
+            require(k(['get', 'namespace', 'kube-system', '-o', 'json'])['metadata']['uid'] == CLUSTER,
+                    'cluster_identity_changed')
+            dep = get('deployment', 'keycape')
+            receipt.update(deployment_metadata=meta(dep), deployment_spec_sha256=canonical(dep['spec']),
+                           image=next(c['image'] for c in dep['spec']['template']['spec']['containers']
+                                      if c['name'] == 'keycape'), mode='metadata_only')
+            if args.mode == 'founder-preflight':
+                require(args.confirm == 'FOUNDER ATTENDED KEY39 PREFLIGHT', 'founder_attendance_required')
+                config = get('secret', 'keycape-config')
+                require(not config['metadata'].get('ownerReferences'), 'controller_owned_config')
+                raw = base64.b64decode(config['data']['config.yaml'], validate=True).decode()
+                candidate_config(raw)
+                candidate_spec(dep['spec'])
+                require(k(['-n', 'sso', 'get', 'secret', TEMP, '--ignore-not-found', '-o', 'json']) is None,
+                        'temporary_secret_already_exists')
+                ready(dep['spec'])
+                receipt.update(config_metadata=meta(config), registry_sha256=digest(raw.encode()),
+                               baseline_ready=True, mode='founder_protected_preflight',
+                               probe_ids_absent=True, probe_env_absent=True, temporary_secret_absent=True)
+        else:
+            require(args.envelope is not None, 'approval_envelope_required')
+            envelope = read_json(Path(args.envelope).read_text())
+            require(Path(args.envelope).stat().st_mode & 0o022 == 0, 'approval_envelope_writable')
+            signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(Refused('session_interrupted')))
+            if args.mode == 'recover':
+                recover(envelope, receipt)
+            else:
+                private = read_json(__import__('sys').stdin.read())
+                execute(envelope, private, receipt, checkpoint=checkpoint)
+                receipt['negative_and_cleanup_verified'] = True
+        rc = 0
+    except Refused as error:
+        receipt.update(status='incomplete_requires_owner_review', reason=str(error))
+    except BaseException:
+        # Fixed outcome only: exception text may contain protected Kubernetes/HTTP data.
+        receipt['status'] = 'incomplete_requires_owner_review'
+    finally:
+        checkpoint()
+        os.close(fd)
+    return rc
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/sso-mfa/k8s/keycape/test_openbao_negative_session.py b/sso-mfa/k8s/keycape/test_openbao_negative_session.py
new file mode 100644
index 0000000..88ecadf
--- /dev/null
+++ b/sso-mfa/k8s/keycape/test_openbao_negative_session.py
@@ -0,0 +1,603 @@
+import base64
+import copy
+import importlib.util
+import json
+from pathlib import Path
+import time
+import io
+import os
+import urllib.error
+import signal
+import threading
+import unittest
+from unittest.mock import patch
+
+spec = importlib.util.spec_from_file_location('session', Path(__file__).with_name('openbao-negative-session.py'))
+s = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(s)
+
+
+def baseline():
+    reference = dict(s.registrations()[-1], clientId='secrets-engine-openbao',
+                     secretRef='env:PRODUCTION', tokenLifetime='15m')
+    return '# preserve this comment\nclients:\n' + s.yaml.safe_dump([reference], sort_keys=False) + 'other: unchanged\n'
+
+
+def envelope():
+    return {'proposal_id': s.PROPOSAL, 'target': s.TARGET, 'disposition': 'approved',
+            'founder_decision_id': 'fixture-decision',
+            'source_hashes': {str(Path(s.__file__).resolve()): s.digest(Path(s.__file__).read_bytes())},
+            'owner_acknowledgments': {o: 'fixture-owner-review' for o in
+                                      ['net-kingdom', 'key-cape', 'railiance-platform']},
+            'valid_until_epoch': time.time() + 900, 'maximum_session_seconds': 900,
+            'bao_addr': 'http://openbao.openbao.svc:8200',
+            'role_observation': {'mount': 'keycape-services', 'role': 'secrets-engine-openbao',
+                 'issuer': 'https://kc.coulomb.social', 'algorithm': 'RS256',
+                 'bindings_exact': True, 'policy_exact': True, 'raw_expiration_leeway': 0,
+                 'effective_expiration_leeway': 150, 'observed_at_epoch': time.time()}}
+
+
+class Tests(unittest.TestCase):
+    def setUp(self):
+        s.REQUEST_DEADLINE = None
+        s.BUSINESS_MONOTONIC_DEADLINE = None
+        s.CLEANUP_MONOTONIC_DEADLINE = None
+        s.CLEANING = False
+    def test_preserve_original_bytes(self):
+        raw = baseline()
+        result = s.candidate_config(raw)
+        start = result.index('clientId: openbao-probe-aud')
+        self.assertIn('# preserve this comment', result[:start])
+        parsed = s.yaml.safe_load(result)
+        self.assertEqual(parsed['clients'][:-6], s.yaml.safe_load(raw)['clients'])
+        self.assertEqual(parsed['other'], 'unchanged')
+
+    def test_isolation_and_sixth_matching_expiry(self):
+        clients = s.registrations()
+        self.assertNotIn('audience', clients[0])
+        for c in clients[1:]:
+            self.assertEqual(c['audience'], 'secrets-engine-openbao')
+        for key in ['roles', 'tenant', 'serviceSubject', 'allowedScopes']:
+            self.assertEqual(clients[-1][key], s.yaml.safe_load(baseline())['clients'][0][key])
+
+    def test_preexisting_probe_refused(self):
+        raw = baseline().replace('other: unchanged', '- clientId: openbao-probe-aud\nother: unchanged')
+        with self.assertRaises(s.Refused):
+            s.candidate_config(raw)
+
+    def test_duplicate_yaml_refused(self):
+        with self.assertRaises(s.Refused):
+            s.candidate_config(baseline() + 'other: shadow\n')
+
+    def test_production_drift_refused(self):
+        with self.assertRaises(s.Refused):
+            s.candidate_config(baseline().replace('tenant:platform', 'tenant:other'))
+
+    def test_env_collision_refused(self):
+        spec = {'replicas': 1, 'strategy': {'type': 'Recreate'}, 'template': {'spec': {'containers': [
+            {'name': 'keycape', 'env': [{'name': s.ENVS[0], 'value': 'fixture'}]}]}}}
+        with self.assertRaises(s.Refused):
+            s.candidate_spec(spec)
+
+    def test_recovery_roundtrip_preserves_every_baseline_byte(self):
+        raw = baseline()
+        self.assertEqual(s.recovered_config(s.candidate_config(raw)), raw)
+        self.assertEqual(s.recovered_config(raw), raw)
+
+    def test_recovery_claim_drift_and_partial_registration_refused(self):
+        raw = s.candidate_config(baseline())
+        with self.assertRaises(s.Refused):
+            s.recovered_config(raw.replace('service:openbao-probe', 'service:changed'))
+        with self.assertRaises(s.Refused):
+            s.recovered_config(raw.replace('clientId: openbao-probe-aud', 'clientId: different-probe'))
+
+    def test_recovery_env_roundtrip(self):
+        spec = {'replicas': 1, 'strategy': {'type': 'Recreate'}, 'template': {'spec': {'containers': [
+            {'name': 'keycape', 'env': [{'name': 'ORIGINAL', 'value': 'fixture'}]}]}}}
+        self.assertEqual(s.recovered_spec(s.candidate_spec(spec)), spec)
+
+    def test_approval_is_not_optional(self):
+        e = envelope()
+        del e['founder_decision_id']
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_source_change_refused(self):
+        e = envelope()
+        e['source_hashes'][str(Path(s.__file__).resolve())] = '0' * 64
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_owner_missing_refused(self):
+        e = envelope()
+        del e['owner_acknowledgments']['net-kingdom']
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_leeway_default_and_disabled(self):
+        e = envelope()
+        self.assertEqual(s.verify_sources(e), 150)
+        e['role_observation'].update(raw_expiration_leeway=-1, effective_expiration_leeway=0)
+        self.assertEqual(s.verify_sources(e), 0)
+
+    def test_stale_role_refused(self):
+        e = envelope()
+        e['role_observation']['observed_at_epoch'] -= 301
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_wrong_effective_leeway_refused(self):
+        e = envelope()
+        e['role_observation']['effective_expiration_leeway'] = 0
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_bao_credentials_in_origin_refused(self):
+        e = envelope()
+        e['bao_addr'] = 'https://user:password@example.invalid'
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_future_role_observation_refused(self):
+        e = envelope()
+        e['role_observation']['observed_at_epoch'] = time.time() + 60
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_arbitrary_https_origin_refused(self):
+        e = envelope()
+        e['bao_addr'] = 'https://attacker.invalid'
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+
+    def test_hostile_proxy_is_disabled(self):
+        build = s.urllib.request.build_opener
+        with patch.dict(os.environ, {'HTTPS_PROXY':'http://hostile-proxy.invalid:8000',
+                                     'HTTP_PROXY':'http://hostile-proxy.invalid:8000'}), \
+             patch.object(s.urllib.request, 'build_opener', wraps=build) as observed:
+            transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        handlers = observed.call_args.args
+        self.assertIsInstance(handlers[0], s.urllib.request.ProxyHandler)
+        self.assertEqual(handlers[0].proxies, {})
+        self.assertIsInstance(handlers[1], s.NoRedirect)
+        self.assertIsNone(handlers[1].redirect_request(None, None, 302, '', {}, 'https://hostile.invalid'))
+
+    def test_redirect_never_follows_or_counts_as_refusal(self):
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        response = urllib.error.HTTPError('https://kc.coulomb.social/token', 302,
+            'redirect', {}, io.BytesIO(b'{}'))
+        with patch.object(transport.opener, 'open', side_effect=response) as observed:
+            status, raw = transport.post('https://kc.coulomb.social/token', form={})
+        self.assertEqual(status, 302)
+        self.assertEqual(observed.call_count, 1)
+
+    def test_oversized_http_body_is_closed_and_refused(self):
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        response = type('Response', (), {'getcode':lambda self: 200,
+            'read':lambda self, size: b'x' * size, 'close':lambda self: setattr(self, 'closed', True)})()
+        with patch.object(transport.opener, 'open', return_value=response):
+            with self.assertRaisesRegex(s.Refused, 'http_response_oversized'):
+                transport.post('https://kc.coulomb.social/token', form={})
+        self.assertTrue(response.closed)
+
+    def test_expired_window_prevents_protected_kubernetes_and_http(self):
+        s.REQUEST_DEADLINE = time.time() - 1
+        with patch.object(s.subprocess, 'run') as backend:
+            with self.assertRaisesRegex(s.Refused, 'approved_window_expired'):
+                s.k(['create', '-f', '/dev/stdin'], {})
+            backend.assert_not_called()
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        with patch.object(transport.opener, 'open') as backend:
+            with self.assertRaisesRegex(s.Refused, 'approved_window_expired'):
+                transport.post('https://kc.coulomb.social/token', form={})
+            backend.assert_not_called()
+
+    def test_cleanup_revoke_allowed_after_window_expiry(self):
+        s.REQUEST_DEADLINE = time.time() - 1
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        response = type('Response', (), {'getcode':lambda self: 204,
+            'read':lambda self, size: b'', 'close':lambda self: None})()
+        with patch.object(transport.opener, 'open', return_value=response):
+            self.assertEqual(transport.post('http://openbao.openbao.svc:8200/v1/auth/token/revoke-self'), (204, b''))
+
+    def test_http_read_has_finite_size(self):
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        sizes = []
+        response = type('Response', (), {'getcode':lambda self: 200,
+            'read':lambda self, size: sizes.append(size) or b'{}', 'close':lambda self: None})()
+        with patch.object(transport.opener, 'open', return_value=response):
+            transport.post('https://kc.coulomb.social/token')
+        self.assertEqual(sizes, [s.MAX_RESPONSE + 1])
+
+    def test_nonfinite_and_boolean_timestamps_refused(self):
+        for value in [float('inf'), float('-inf'), float('nan'), True, False, 'Infinity', 10 ** 1000]:
+            for field in ['valid_until_epoch', 'observed_at_epoch']:
+                with self.subTest(field=field, value_type=type(value).__name__):
+                    e = envelope()
+                    if field == 'valid_until_epoch':
+                        e[field] = value
+                    else:
+                        e['role_observation'][field] = value
+                    with self.assertRaises(s.Refused):
+                        s.verify_sources(e)
+
+    def test_finite_far_future_and_noninteger_window_refused(self):
+        e = envelope()
+        e['valid_until_epoch'] = time.time() + 86400
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e)
+        for value in [True, 900.0, float('inf'), float('nan'), '900']:
+            e = envelope()
+            e['maximum_session_seconds'] = value
+            with self.assertRaises(s.Refused):
+                s.verify_sources(e)
+
+    def test_recovery_also_requires_finite_typed_window(self):
+        e = envelope()
+        e['disposition'] = 'recovery-approved'
+        e['valid_until_epoch'] = float('inf')
+        with self.assertRaises(s.Refused):
+            s.verify_sources(e, recovery=True)
+
+    def test_nonfinite_json_constants_refused(self):
+        for constant in ['Infinity', '-Infinity', 'NaN']:
+            with self.assertRaises(s.Refused):
+                s.read_json('{"valid_until_epoch":' + constant + '}')
+        self.assertFalse(s.finite_number(s.read_json('{"valid_until_epoch":1e999}')['valid_until_epoch']))
+
+    def test_hard_timer_covers_slow_trickle_body_and_closes_response(self):
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        class Trickle:
+            def getcode(self):
+                return 200
+            def read(self, size):
+                for _ in range(100):
+                    time.sleep(0.01)
+                return b'{}'
+            def close(self):
+                self.closed = True
+        response = Trickle()
+        previous = signal.getsignal(signal.SIGALRM)
+        started = time.monotonic()
+        with patch.object(transport.opener, 'open', return_value=response):
+            with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
+                transport.post('https://kc.coulomb.social/token', timeout=0.04)
+        self.assertLess(time.monotonic() - started, 0.25)
+        self.assertTrue(response.closed)
+        self.assertIs(signal.getsignal(signal.SIGALRM), previous)
+        self.assertEqual(signal.getitimer(signal.ITIMER_REAL), (0.0, 0.0))
+
+    def test_hard_timer_spans_headers_and_clips_to_business_window(self):
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        s.REQUEST_DEADLINE = time.time() + 60
+        s.BUSINESS_MONOTONIC_DEADLINE = time.monotonic() + 0.04
+        def slow_headers(*args, **kwargs):
+            time.sleep(1)
+            raise AssertionError('inactivity timeout would have permitted this')
+        started = time.monotonic()
+        with patch.object(transport.opener, 'open', side_effect=slow_headers):
+            with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
+                transport.post('https://kc.coulomb.social/token')
+        self.assertLess(time.monotonic() - started, 0.25)
+
+    def test_late_full_acceptance_is_revoked_then_refused(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        token = 'not-a-real-late-token'
+        class Late:
+            def getcode(self):
+                return 200
+            def read(self, size):
+                s.REQUEST_DEADLINE = time.time() - 1
+                return json.dumps({'auth': {'client_token': token}}).encode()
+            def close(self):
+                pass
+        revoke = type('Response', (), {'getcode':lambda self: 204,
+            'read':lambda self, size: b'', 'close':lambda self: None})()
+        with patch.object(transport.opener, 'open', side_effect=[Late(), revoke]) as calls:
+            with self.assertRaisesRegex(s.Refused, 'approved_window_expired'):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login')
+        self.assertEqual(calls.call_count, 2)
+        self.assertEqual(calls.call_args_list[1].args[0].full_url,
+                         'http://openbao.openbao.svc:8200/v1/auth/token/revoke-self')
+        self.assertTrue(receipt['late_acceptance_revoke_attempted'])
+        self.assertTrue(receipt['late_acceptance_revoke_succeeded'])
+        self.assertNotIn(token, json.dumps(receipt))
+
+    def test_late_acceptance_failed_revoke_remains_unconfirmed(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        class Late:
+            def getcode(self):
+                return 200
+            def read(self, size):
+                s.REQUEST_DEADLINE = time.time() - 1
+                return b'{"auth":{"client_token":"fake-token"}}'
+            def close(self):
+                pass
+        failed = type('Response', (), {'getcode':lambda self: 403,
+            'read':lambda self, size: b'{}', 'close':lambda self: None})()
+        with patch.object(transport.opener, 'open', side_effect=[Late(), failed]):
+            with self.assertRaises(s.Refused):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login')
+        self.assertTrue(receipt['late_acceptance_revoke_attempted'])
+        self.assertFalse(receipt['late_acceptance_revoke_succeeded'])
+
+    def test_partial_accepted_response_timeout_retains_unknown_outcome(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        response = type('Response', (), {'getcode':lambda self: 200,
+            'read':lambda self, size: time.sleep(1), 'close':lambda self: None})()
+        with patch.object(transport.opener, 'open', return_value=response):
+            with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.04)
+        self.assertTrue(receipt['openbao_login_outcome_unknown'])
+
+    def test_cleanup_has_finite_aggregate_deadline_and_restoration(self):
+        s.REQUEST_DEADLINE = time.time() - 1
+        s.begin_cleanup()
+        first = s.CLEANUP_MONOTONIC_DEADLINE
+        s.begin_cleanup()
+        self.assertEqual(s.CLEANUP_MONOTONIC_DEADLINE, first)
+        s.CLEANUP_MONOTONIC_DEADLINE = time.monotonic() - 1
+        s.CLEANING = True
+        with self.assertRaisesRegex(s.Refused, 'cleanup_budget_expired'):
+            s.check_deadline()
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200')
+        with patch.object(transport.opener, 'open') as backend:
+            with self.assertRaises(s.Refused):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/token/revoke-self')
+            backend.assert_not_called()
+
+    def test_expired_cleanup_cannot_issue_native_secret_delete(self):
+        s.CLEANING = True
+        s.CLEANUP_MONOTONIC_DEADLINE = time.monotonic() - 1
+        obj = {'metadata': {'uid':'temporary', 'resourceVersion':'7'}}
+        with patch.object(s.subprocess, 'run') as native:
+            with self.assertRaisesRegex(s.Refused, 'cleanup_budget_expired'):
+                s.delete_temporary(obj)
+            native.assert_not_called()
+
+    def test_secret_delete_total_timeout_clipped_to_cleanup_budget(self):
+        s.CLEANING = True
+        s.CLEANUP_MONOTONIC_DEADLINE = time.monotonic() + 0.04
+        obj = {'metadata': {'uid':'temporary', 'resourceVersion':'7'}}
+        captured = {}
+        def slow_delete(args, **kwargs):
+            captured.update(kwargs)
+            time.sleep(1)
+            return type('Result', (), {'returncode':0, 'stdout':'{}'})()
+        started = time.monotonic()
+        with patch.object(s.subprocess, 'run', side_effect=slow_delete):
+            with self.assertRaisesRegex(s.Refused, 'request_total_timeout'):
+                s.delete_temporary(obj)
+        self.assertLess(time.monotonic() - started, 0.25)
+        self.assertLessEqual(captured['timeout'], 0.04)
+        self.assertEqual(json.loads(captured['input'])['preconditions'], {'uid':'temporary', 'resourceVersion':'7'})
+
+    def test_early_read_failure_slow_close_is_bounded_once(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        class Response:
+            closes = 0
+            def getcode(self):
+                return 200
+            def read(self, size):
+                raise s.Refused('synthetic_read_failure')
+            def close(self):
+                self.closes += 1
+                time.sleep(0.07)
+        response = Response()
+        previous = signal.getsignal(signal.SIGALRM)
+        started = time.monotonic()
+        with patch.object(transport.opener, 'open', return_value=response):
+            with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
+        self.assertLess(time.monotonic() - started, 0.05)
+        self.assertEqual(response.closes, 1)
+        self.assertTrue(receipt['openbao_login_outcome_unknown'])
+        self.assertTrue(receipt['http_abort_close_attempted'])
+        self.assertFalse(receipt['http_abort_close_succeeded'])
+        self.assertTrue(receipt['http_response_abort_unconfirmed'])
+        self.assertIs(signal.getsignal(signal.SIGALRM), previous)
+        self.assertEqual(signal.getitimer(signal.ITIMER_REAL), (0.0, 0.0))
+
+    def test_total_read_timeout_then_slow_close_uses_only_abort_cap(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        class Response:
+            closes = 0
+            def getcode(self):
+                return 200
+            def read(self, size):
+                time.sleep(0.1)
+            def close(self):
+                self.closes += 1
+                time.sleep(0.1)
+        response = Response()
+        started = time.monotonic()
+        with patch.object(transport.opener, 'open', return_value=response):
+            with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
+        self.assertLess(time.monotonic() - started, 0.06)
+        self.assertEqual(response.closes, 1)
+        self.assertTrue(receipt['openbao_login_outcome_unknown'])
+        self.assertFalse(receipt['http_abort_close_succeeded'])
+
+    def test_oversized_body_slow_abort_remains_unproved_and_bounded(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        response = type('Response', (), {'getcode':lambda self: 200,
+            'read':lambda self, size: b'x' * size,
+            'close':lambda self: time.sleep(0.1)})()
+        started = time.monotonic()
+        with patch.object(transport.opener, 'open', return_value=response):
+            with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
+        self.assertLess(time.monotonic() - started, 0.05)
+        self.assertTrue(receipt['openbao_login_outcome_unknown'])
+        self.assertFalse(receipt['http_abort_close_succeeded'])
+
+    def test_raising_normal_close_then_slow_abort_is_private_bounded_unknown(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        class Response:
+            closes = 0
+            def getcode(self):
+                return 200
+            def read(self, size):
+                return b'{}'
+            def close(self):
+                self.closes += 1
+                if self.closes == 1:
+                    raise RuntimeError('private protected error must not escape')
+                time.sleep(0.1)
+        response = Response()
+        previous = signal.getsignal(signal.SIGALRM)
+        started = time.monotonic()
+        with patch.object(transport.opener, 'open', return_value=response):
+            with self.assertRaisesRegex(s.Refused, '^http_response_abort_unconfirmed$'):
+                transport.post('http://openbao.openbao.svc:8200/v1/auth/keycape-services/login', timeout=0.01)
+        self.assertLess(time.monotonic() - started, 0.05)
+        self.assertEqual(response.closes, 2)
+        self.assertTrue(receipt['openbao_login_outcome_unknown'])
+        self.assertFalse(receipt['http_abort_close_succeeded'])
+        self.assertIs(signal.getsignal(signal.SIGALRM), previous)
+        self.assertEqual(signal.getitimer(signal.ITIMER_REAL), (0.0, 0.0))
+
+    def test_expired_cleanup_never_attempts_exceptional_close_or_resets(self):
+        receipt = {}
+        transport = s.SessionTransport('http://openbao.openbao.svc:8200', receipt=receipt)
+        deadline = time.monotonic() - 1
+        s.CLEANUP_MONOTONIC_DEADLINE = deadline
+        response = type('Response', (), {'close':lambda self: self.fail('must not close')})()
+        with patch.object(response, 'close') as closes:
+            with self.assertRaisesRegex(s.Refused, 'http_response_abort_unconfirmed'):
+                transport.abort_response(response, time.monotonic() - 1)
+            closes.assert_not_called()
+        self.assertEqual(s.CLEANUP_MONOTONIC_DEADLINE, deadline)
+        self.assertFalse(receipt['http_abort_close_attempted'])
+        self.assertTrue(receipt['http_response_abort_unconfirmed'])
+
+    def test_thread_or_existing_timer_refused_without_overwrite(self):
+        results = []
+        def in_thread():
+            try:
+                with s.total_timeout(0.1):
+                    results.append('unreachable')
+            except s.Refused:
+                results.append('refused')
+        thread = threading.Thread(target=in_thread)
+        thread.start()
+        thread.join()
+        self.assertEqual(results, ['refused'])
+        signal.setitimer(signal.ITIMER_REAL, 1)
+        try:
+            with self.assertRaisesRegex(s.Refused, 'existing_alarm_timer_refused'):
+                with s.total_timeout(0.1):
+                    pass
+            self.assertGreater(signal.getitimer(signal.ITIMER_REAL)[0], 0.5)
+        finally:
+            signal.setitimer(signal.ITIMER_REAL, 0)
+
+    def test_cleanup_refuses_concurrent_registry_change(self):
+        self.lifecycle(drift=True)
+
+    def test_deleted_client_contract_is_specific(self):
+        self.assertTrue(s.removed_client_denied(401, b'{"error":"invalid_profile_usage","feature":"client_id"}'))
+        for status, body in [(503, b'{}'), (401, b'{"error":"invalid_profile_usage","feature":"Authorization"}'),
+                             (401, b'{"error":"invalid_client"}'), (401, b'not-json')]:
+            self.assertFalse(s.removed_client_denied(status, body))
+
+    def test_cleanup_even_when_probe_fails(self):
+        self.lifecycle(probe_pass=False)
+
+    def test_complete_lifecycle(self):
+        self.lifecycle()
+
+    def lifecycle(self, drift=False, probe_pass=True):
+        raw = baseline()
+        dep_spec = {'replicas': 1, 'strategy': {'type': 'Recreate'}, 'template': {'spec': {'containers': [
+            {'name': 'keycape', 'image': 'fixture@sha256:abc', 'env': []}]}}}
+        objects = {'keycape-config': {'metadata': {'name': 'keycape-config', 'uid': 'cfg', 'resourceVersion': '1'},
+                                     'data': {'config.yaml': base64.b64encode(raw.encode()).decode(),
+                                              'key.pem': 'FAKE_KEY_NOT_READ'}},
+                   'keycape': {'metadata': {'name': 'keycape', 'uid': 'dep', 'resourceVersion': '1'},
+                               'spec': dep_spec}}
+        original = copy.deepcopy(objects)
+        e = envelope()
+        e.update(runner_path='fake-runner', config_metadata=s.meta(objects['keycape-config']),
+                 deployment_metadata=s.meta(objects['keycape']), deployment_spec_sha256=s.canonical(dep_spec),
+                 registry_sha256=s.digest(raw.encode()))
+        e['source_hashes']['fake-runner'] = 'fixture'
+        receipt = {'cleanup_verified': False}
+        class Probe:
+            def exchange(self, *args):
+                return 'fixture-token'
+            def http_post(self, *args, **kwargs):
+                return 401, b'{"error":"invalid_profile_usage","feature":"client_id"}'
+            def run(self, *args, **kwargs):
+                if drift:
+                    objects['keycape-config']['data']['unrelated'] = 'concurrent-owner-change'
+                return {'all_refused': probe_pass}
+        def kube(args, payload=None):
+            if args[0] == 'get':
+                return {'metadata': {'uid': s.CLUSTER}}
+            if args[0] == 'create':
+                if '--dry-run=server' in args:
+                    return copy.deepcopy(payload)
+                created = copy.deepcopy(payload)
+                created['metadata'].update(uid='temp', resourceVersion='1')
+                objects[s.TEMP] = created
+                return created
+            if args[0] == 'delete':
+                self.assertEqual(payload['preconditions'], {'uid':'temp', 'resourceVersion':'1'})
+                del objects[s.TEMP]
+                return {'kind':'Status', 'status':'Success'}
+            if '--ignore-not-found' in args:
+                return objects.get(s.TEMP)
+            raise AssertionError('unexpected kube call')
+        def patched(kind, obj, path, value, dry=False):
+            result = copy.deepcopy(obj)
+            if path == '/spec':
+                result['spec'] = copy.deepcopy(value)
+            else:
+                result['data']['config.yaml'] = value
+            if not dry:
+                objects[obj['metadata']['name']] = result
+            return result
+        def delete(*args, **kwargs):
+            del objects[s.TEMP]
+            return type('Result', (), {'returncode': 0})()
+        class Discovery:
+            def __enter__(self):
+                import io
+                return io.StringIO('{"issuer":"https://kc.coulomb.social","token_endpoint":"https://kc.coulomb.social/token"}')
+            def __exit__(self, *args):
+                pass
+        with patch.object(s, 'verify_sources', return_value=150), patch.object(s, 'load_probe', return_value=Probe()), \
+             patch.object(s, 'SessionTransport') as transport_factory, \
+             patch.object(s, 'k', side_effect=kube), patch.object(s, 'get', side_effect=lambda kind, name: copy.deepcopy(objects[name])), \
+             patch.object(s, 'patch', side_effect=patched), patch.object(s, 'ready', return_value=True), \
+             patch.object(s.subprocess, 'run', side_effect=delete):
+            transport_factory.return_value.discovery.return_value = {
+                'issuer':'https://kc.coulomb.social', 'token_endpoint':'https://kc.coulomb.social/token'}
+            transport_factory.return_value.post.side_effect = Probe().http_post
+            if drift or not probe_pass:
+                with self.assertRaises(s.Refused):
+                    s.execute(e, {'human_token': 'fake', 'coding_agent_token': 'fake'}, receipt)
+            else:
+                s.execute(e, {'human_token': 'fake', 'coding_agent_token': 'fake'}, receipt)
+        if drift:
+            self.assertFalse(receipt['cleanup_verified'])
+            self.assertIn(s.TEMP, objects)
+            self.assertEqual(objects['keycape-config']['data']['unrelated'], 'concurrent-owner-change')
+        else:
+            self.assertTrue(receipt['cleanup_verified'])
+            self.assertNotIn(s.TEMP, objects)
+            self.assertEqual(objects['keycape-config']['data'], original['keycape-config']['data'])
+            self.assertEqual(objects['keycape']['spec'], original['keycape']['spec'])
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/workplans/ADHOC-2026-10-11.md b/workplans/ADHOC-2026-10-11.md
index c7edd75..7c0552e 100644
--- a/workplans/ADHOC-2026-10-11.md
+++ b/workplans/ADHOC-2026-10-11.md
@@ -196,3 +196,40 @@ formal owner review under the temporary source-review stage. This completed
 local review task is distinct from KEY39 T02 and all native owner obligations.
 No founder decision, deployment, credential access, Secret read, grant, live
 acceptance or task closure in other repositories is claimed.
+
+T07 same-trial v3 refinement review: independently reviewed exact manifest
+fbc31fda2b6194747547b9fbe876095914c3d53231c79abd43a010dbb3273efa.
+Both v2 deadline counterexamples now refuse; a real elapsed timer interrupts
+slow HTTP body reads and restores signal state. Thirty-nine supplied and fifteen
+independent checks pass, including a retained new failure reproduction: final
+temporary Secret deletion bypasses the aggregate cleanup deadline through an
+unguarded subprocess call. Changes requested for that exact source path only;
+v2 changes-requested evidence and frozen stages stay unchanged. No native
+authority, owner acceptance or KEY39 task closure is inferred.
+
+
+T07 same-trial final v4 source refinement review: exact frozen manifest
+d079f85c47d5c2635e1ee633ea4d1346b1737ce2be9ca273f22a38e620f418d8
+routes the exact temporary Secret deletion through guarded k. The v3 expired
+cleanup reproduction now makes zero native calls; a real slow-delete timer is
+clipped to remaining cleanup time. Both earlier v2 counterexamples now refuse,
+late full OpenBao acceptance is self-revoked privately before failure, partial
+acceptance stays unknown, and aggregate cleanup does not renew. Forty-one
+supplied lifecycle tests, seventeen unchanged-runner tests and seventeen
+independent checks pass. NetKingdom source agreement only; earlier unsuccessful
+reviews and immutable sources remain history in this same trial. Native token
+production, runtime authority, quiet-window/headroom, execution, revocation/
+cleanup and other owners' task closure remain separate obligations.
+
+T07 same-trial v5 exceptional HTTP-close refinement review: frozen manifest
+d1f997624db46693a7126a6eb6435f6b3a353d5bd67dcc8a539683e4b7147285
+retains NetKingdom's prior v4 agreement and Platform's distinct v4
+changes-requested finding as history. Exceptional response close now has one
+20 ms abort cap clipped to positive original request time and the same
+nonrenewable aggregate cleanup deadline; expired aggregate performs no close.
+Independent early-read and timed-read failure/slow-close checks, known accepted
+session revocation despite normal-close error, signal restoration, value-free
+unknown outcomes and prior deadline/delete/recovery guards pass. Forty-six
+supplied lifecycle tests, seventeen unchanged-runner tests and twenty-one
+independent checks pass. NetKingdom exact source agreement only; same trial,
+no new task, native authority, credential access, runtime acceptance or closure.