diff --git a/scripts/collect-upstream-issuer-proof.py b/scripts/collect-upstream-issuer-proof.py
new file mode 100644
index 0000000..29627c2
--- /dev/null
+++ b/scripts/collect-upstream-issuer-proof.py
@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Read bounded probe logs on stdin and emit only a validated metadata receipt."""
+import json
+import re
+import sys
+
+ISSUERS = {'https://auth.coulomb.social', 'http://auth.coulomb.social',
+           'http://authelia.sso.svc.cluster.local:9091'}
+FAILURES = {'authorization_callback_refused', 'token_exchange_error', 'token_exchange_refused',
+            'token_response_invalid', 'id_token_verification_error', 'issuer_outside_reviewed_set',
+            'id_token_issuer_mismatch', 'id_token_audience_mismatch', 'id_token_expired',
+            'id_token_validity_window', 'id_token_signature', 'provider_metadata_unavailable',
+            'provider_keys_unavailable', 'id_token_nonce_mismatch', 'probe_deadline', 'probe_listener_stopped'}
+
+
+def sanitize(raw):
+    if len(raw) > 8192:
+        raise ValueError()
+    lines = raw.splitlines()
+    if not lines:
+        raise ValueError()
+    data = json.loads(lines[0])
+    if not isinstance(data, dict) or data.get('schema') != 'keycape.upstream-issuer-proof.v1':
+        raise ValueError()
+    if data.get('tokens_retained') is not False or data.get('downstream_credential_issued') is not False:
+        raise ValueError()
+    keys = {'schema', 'status', 'tokens_retained', 'downstream_credential_issued'}
+    if 'observed_at' in data:
+        if not isinstance(data['observed_at'], str) or not re.fullmatch(r'\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ', data['observed_at']):
+            raise ValueError()
+        keys.add('observed_at')
+    if data.get('status') == 'verified':
+        if len(lines) != 1 or data.get('issuer') not in ISSUERS or 'observed_at' not in data:
+            raise ValueError()
+        keys.add('issuer')
+        for field in ['signature_verified', 'audience_verified', 'validity_window_verified', 'nonce_verified']:
+            if data.get(field) is not True:
+                raise ValueError()
+            keys.add(field)
+    elif data.get('status') == 'failed' and data.get('failure') in FAILURES:
+        if len(lines) > 2 or (len(lines) == 2 and lines[1] != 'issuer proof failed'):
+            raise ValueError()
+        keys.add('failure')
+    else:
+        raise ValueError()
+    if set(data) != keys:
+        raise ValueError()
+    return data
+
+
+if __name__ == '__main__':
+    try:
+        result = sanitize(sys.stdin.read(8193))
+    except Exception:
+        print('No valid metadata-only issuer proof receipt.', file=sys.stderr)
+        raise SystemExit(1) from None
+    print(json.dumps(result, indent=2))
+    raise SystemExit(0 if result['status'] == 'verified' else 1)
diff --git a/scripts/render-upstream-issuer-probe.py b/scripts/render-upstream-issuer-probe.py
new file mode 100644
index 0000000..0f2dca8
--- /dev/null
+++ b/scripts/render-upstream-issuer-probe.py
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Render a bounded, metadata-only issuer probe for NetKingdom review.
+
+No cluster mutation or secret lookup. The deployment owner creates the Job
+first, then re-renders children with its real UID before creating them. This
+makes every temporary route/policy/service follow the Job's cleanup lifetime.
+"""
+import argparse
+import base64
+import hashlib
+import json
+import re
+import secrets
+from pathlib import Path
+
+IMAGE_RE = re.compile(r'forgejo\.coulomb\.social/coulomb/key-cape@sha256:[0-9a-f]{64}')
+ISSUERS = ['https://auth.coulomb.social', 'http://auth.coulomb.social',
+           'http://authelia.sso.svc.cluster.local:9091']
+
+
+def render(image, state, job_uid=None):
+    if not IMAGE_RE.fullmatch(image):
+        raise ValueError('an immutable KeyCape image digest is required')
+    if not re.fullmatch(r'[A-Za-z0-9_-]{43}', state):
+        raise ValueError('invalid state')
+    decoded = base64.urlsafe_b64decode(state + '=')
+    if len(decoded) != 32 or base64.urlsafe_b64encode(decoded).decode().rstrip('=') != state:
+        raise ValueError('noncanonical state')
+    if job_uid is not None and not re.fullmatch(r'[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}', job_uid):
+        raise ValueError('invalid Job UID')
+    name = 'keycape-issuer-proof-' + hashlib.sha256(state.encode()).hexdigest()[:12]
+    labels = {'keycape.coulomb.social/issuer-proof': name}
+    peer = {'podSelector': {'matchLabels': labels}}
+    metadata = {'name': name, 'namespace': 'sso', 'labels': labels}
+    args = ['probe-upstream-issuer', '--config=/etc/keycape/config.yaml',
+            '--listen=:8081', '--lifetime=10m', '--allowed-issuers=' + ','.join(ISSUERS)]
+    job = {
+        'apiVersion': 'batch/v1', 'kind': 'Job', 'metadata': dict(metadata),
+        'spec': {'backoffLimit': 0, 'activeDeadlineSeconds': 600, 'ttlSecondsAfterFinished': 300,
+                 'template': {'metadata': {'labels': labels}, 'spec': {
+                     'restartPolicy': 'Never', 'automountServiceAccountToken': False,
+                     'securityContext': {'runAsNonRoot': True, 'runAsUser': 65534, 'runAsGroup': 65534,
+                                         'fsGroup': 65534, 'seccompProfile': {'type': 'RuntimeDefault'}},
+                     'containers': [{'name': 'probe', 'image': image, 'args': args,
+                                     'env': [{'name': 'KEYCAPE_ISSUER_PROBE_STATE', 'value': state}],
+                                     'ports': [{'name': 'probe', 'containerPort': 8081}],
+                                     'securityContext': {'allowPrivilegeEscalation': False,
+                                                         'readOnlyRootFilesystem': True,
+                                                         'capabilities': {'drop': ['ALL']}},
+                                     'resources': {'requests': {'cpu': '25m', 'memory': '32Mi'},
+                                                   'limits': {'cpu': '200m', 'memory': '128Mi'}},
+                                     'readinessProbe': {'httpGet': {'path': '/healthz', 'port': 8081},
+                                                        'periodSeconds': 2},
+                                     'volumeMounts': [{'name': 'config', 'mountPath': '/etc/keycape', 'readOnly': True}]}],
+                     'volumes': [{'name': 'config', 'secret': {'secretName': 'keycape-config',
+                                 'items': [{'key': 'config.yaml', 'path': 'config.yaml', 'mode': 288}]}}]
+                 }}}}
+    children_metadata = dict(metadata)
+    if job_uid:
+        children_metadata['ownerReferences'] = [{'apiVersion': 'batch/v1', 'kind': 'Job',
+             'name': name, 'uid': job_uid, 'blockOwnerDeletion': False}]
+    # Backticks belong to Traefik's rule language, not shell execution.
+    rule = ('Host(`kc.coulomb.social`) && (Path(`/upstream-issuer-proof/' + state +
+            '`) || (Path(`/authorize/callback`) && Query(`state`, `' + state + '`)))')
+    children = [
+        {'apiVersion': 'v1', 'kind': 'Service', 'metadata': dict(children_metadata),
+         'spec': {'selector': labels, 'ports': [{'port': 8081, 'targetPort': 8081}]}},
+        {'apiVersion': 'traefik.io/v1alpha1', 'kind': 'IngressRoute', 'metadata': dict(children_metadata),
+         'spec': {'entryPoints': ['websecure'], 'tls': {'secretName': 'kc-tls'},
+                  'routes': [{'kind': 'Rule', 'match': rule, 'priority': 10000,
+                              'middlewares': [{'name': 'keycape-rate-limit'}, {'name': 'keycape-hsts'}],
+                              'observability': {'accessLogs': False, 'metrics': False, 'tracing': False},
+                              'services': [{'name': name, 'port': 8081}]}]}},
+        {'apiVersion': 'networking.k8s.io/v1', 'kind': 'NetworkPolicy', 'metadata': dict(children_metadata),
+         'spec': {'podSelector': {'matchLabels': labels}, 'policyTypes': ['Ingress', 'Egress'],
+                  'ingress': [{'from': [{'namespaceSelector': {'matchLabels': {'kubernetes.io/metadata.name': 'kube-system'}},
+                                        'podSelector': {'matchLabels': {'app.kubernetes.io/name': 'traefik'}}}],
+                               'ports': [{'port': 8081, 'protocol': 'TCP'}]}],
+                  'egress': [{'to': [{'podSelector': {'matchLabels': {'app.kubernetes.io/name': 'authelia'}}}],
+                              'ports': [{'port': 9091, 'protocol': 'TCP'}]},
+                             {'to': [{'namespaceSelector': {'matchLabels': {'kubernetes.io/metadata.name': 'kube-system'}},
+                                      'podSelector': {'matchLabels': {'k8s-app': 'kube-dns'}}}],
+                              'ports': [{'port': 53, 'protocol': protocol} for protocol in ['TCP', 'UDP']]}]}},
+        {'apiVersion': 'networking.k8s.io/v1', 'kind': 'NetworkPolicy',
+         'metadata': dict(children_metadata, name=name + '-authelia'),
+         'spec': {'podSelector': {'matchLabels': {'app.kubernetes.io/name': 'authelia'}},
+                  'policyTypes': ['Ingress'], 'ingress': [{'from': [peer], 'ports': [{'port': 9091, 'protocol': 'TCP'}]}]}}
+    ]
+    return {'name': name, 'start_url': 'https://kc.coulomb.social/upstream-issuer-proof/' + state,
+            'job': job, 'children': {'apiVersion': 'v1', 'kind': 'List', 'items': children}}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--image', required=True)
+    parser.add_argument('--state', help='reuse the exact random state from the first rendering')
+    parser.add_argument('--job-uid', help='actual UID returned by creating the reviewed Job')
+    parser.add_argument('--output', type=Path, required=True, help='new private packet directory')
+    args = parser.parse_args()
+    state = args.state or secrets.token_urlsafe(32)
+    packet = render(args.image, state, args.job_uid)
+    args.output.mkdir(mode=0o700, parents=False, exist_ok=False)
+    for name, data in [('job.json', packet['job']), ('children.json', packet['children']),
+                       ('packet.json', {'state': state, 'image': args.image, 'job_uid': args.job_uid,
+                                        'name': packet['name'], 'start_url': packet['start_url'],
+                                        'children_bound_to_job': args.job_uid is not None})]:
+        dest = args.output / name
+        with dest.open('x') as stream:
+            dest.chmod(0o600)
+            json.dump(data, stream, indent=2)
+            stream.write('\n')
+    print(json.dumps({'packet_directory': str(args.output), 'job_name': packet['name'],
+                      'children_bound_to_job': args.job_uid is not None, 'cluster_changed': False}))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/scripts/test_render_upstream_issuer_probe.py b/scripts/test_render_upstream_issuer_probe.py
new file mode 100644
index 0000000..ab6f452
--- /dev/null
+++ b/scripts/test_render_upstream_issuer_probe.py
@@ -0,0 +1,79 @@
+import importlib.util
+from pathlib import Path
+import unittest
+
+spec = importlib.util.spec_from_file_location('probe_renderer', Path(__file__).with_name('render-upstream-issuer-probe.py'))
+renderer = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(renderer)
+IMAGE = 'forgejo.coulomb.social/coulomb/key-cape@sha256:' + 'a' * 64
+STATE = 'A' * 43
+UID = '12345678-1234-1234-1234-123456789abc'
+
+
+class RenderProbeTests(unittest.TestCase):
+    def test_dedicated_route_and_pod_cannot_join_production_service(self):
+        packet = renderer.render(IMAGE, STATE, UID)
+        job = packet['job']
+        pod = job['spec']['template']
+        self.assertNotIn('app.kubernetes.io/name', pod['metadata']['labels'])
+        self.assertFalse(pod['spec']['automountServiceAccountToken'])
+        self.assertEqual(job['spec']['activeDeadlineSeconds'], 600)
+        self.assertEqual(job['spec']['backoffLimit'], 0)
+        self.assertEqual(job['spec']['ttlSecondsAfterFinished'], 300)
+        children = packet['children']['items']
+        self.assertEqual(len(children), 4)
+        for child in children:
+            self.assertEqual(child['metadata']['ownerReferences'][0]['uid'], UID)
+        route = children[1]['spec']['routes'][0]
+        self.assertIn('Query(`state`, `' + STATE + '`)', route['match'])
+        self.assertIn('Path(`/authorize/callback`)', route['match'])
+        self.assertNotIn('PathPrefix', route['match'])
+        self.assertFalse(route['observability']['accessLogs'])
+        self.assertFalse(route['observability']['tracing'])
+
+    def test_only_config_yaml_is_projected_without_signing_key_or_api_token(self):
+        pod = renderer.render(IMAGE, STATE)['job']['spec']['template']['spec']
+        self.assertEqual(len(pod['volumes']), 1)
+        secret = pod['volumes'][0]['secret']
+        self.assertEqual(secret['secretName'], 'keycape-config')
+        self.assertEqual(secret['items'], [{'key': 'config.yaml', 'path': 'config.yaml', 'mode': 0o440}])
+        security = pod['containers'][0]['securityContext']
+        self.assertTrue(security['readOnlyRootFilesystem'])
+        self.assertFalse(security['allowPrivilegeEscalation'])
+
+    def test_no_route_or_scope_injection_and_no_mutable_image(self):
+        for state in ['', STATE + '`)', 'B' * 43, 'A' * 22]:
+            with self.assertRaises(ValueError):
+                renderer.render(IMAGE, state)
+        with self.assertRaises(ValueError):
+            renderer.render('forgejo.coulomb.social/coulomb/key-cape:latest', STATE)
+        with self.assertRaises(ValueError):
+            renderer.render(IMAGE, STATE, 'arbitrary-owner')
+
+
+if __name__ == '__main__':
+    unittest.main()
+
+class ReceiptTests(unittest.TestCase):
+    def setUp(self):
+        import json
+        self.json = json
+        spec = importlib.util.spec_from_file_location('collector', Path(__file__).with_name('collect-upstream-issuer-proof.py'))
+        self.collector = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(self.collector)
+        self.receipt = {'schema': 'keycape.upstream-issuer-proof.v1', 'status': 'verified',
+                        'observed_at': '2026-09-08T22:00:00Z', 'issuer': 'https://auth.coulomb.social',
+                        'tokens_retained': False, 'downstream_credential_issued': False,
+                        'signature_verified': True, 'audience_verified': True,
+                        'validity_window_verified': True, 'nonce_verified': True}
+
+    def test_verified_receipt_is_preserved(self):
+        self.assertEqual(self.collector.sanitize(self.json.dumps(self.receipt)), self.receipt)
+
+    def test_extra_claims_unverified_issuers_and_untyped_success_are_refused(self):
+        for patch in [{'sub': 'private-user'}, {'issuer': 'https://unreviewed.example'},
+                      {'signature_verified': 1}, {'tokens_retained': True}]:
+            with self.assertRaises(ValueError):
+                self.collector.sanitize(self.json.dumps(dict(self.receipt, **patch)))
+        with self.assertRaises(ValueError):
+            self.collector.sanitize(self.json.dumps(self.receipt) + '\nprivate-token')
diff --git a/src/cmd/keycape/main.go b/src/cmd/keycape/main.go
index da21c33..e1485fd 100644
--- a/src/cmd/keycape/main.go
+++ b/src/cmd/keycape/main.go
@@ -36,6 +36,14 @@ import (
 const version = "0.1.0"
 
 func main() {
+	if len(os.Args) > 1 && os.Args[1] == "probe-upstream-issuer" {
+		if err := authelia.RunIssuerProbe(context.Background(), os.Args[2:], os.Stdout); err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+		return
+	}
+
 	if len(os.Args) > 1 && (os.Args[1] == "login" || os.Args[1] == "service-token" || os.Args[1] == "verify-client") {
 		if err := authclient.Run(context.Background(), os.Args[1:], os.Stderr); err != nil {
 			fmt.Fprintln(os.Stderr, err)
diff --git a/src/internal/adapters/authelia/issuer_probe.go b/src/internal/adapters/authelia/issuer_probe.go
new file mode 100644
index 0000000..32a5b75
--- /dev/null
+++ b/src/internal/adapters/authelia/issuer_probe.go
@@ -0,0 +1,286 @@
+package authelia
+
+// This one-shot diagnostic uses the existing upstream confidential client and
+// callback URI. It issues no downstream credential and returns no user claims.
+// Deployment requires a route matching ONLY its random state, never all callbacks.
+import (
+	"context"
+	"crypto/rand"
+	"crypto/subtle"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"flag"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"gopkg.in/yaml.v3"
+	"keycape/internal/domain"
+)
+
+// IssuerProbeOptions contains deployment-owned metadata. AllowedIssuers is a
+// finite reviewed set, not a source of URLs to fetch from an untrusted token.
+type IssuerProbeOptions struct {
+	State          string
+	AllowedIssuers []string
+	Lifetime       time.Duration
+}
+
+// IssuerProbe owns one browser-bound flow and emits one metadata-only receipt.
+// Tokens and authorization codes never leave the adapter package.
+type IssuerProbe struct {
+	adapter  *AutheliaAdapter
+	opts     IssuerProbeOptions
+	host     string
+	callback string
+	nonce    string
+	expires  time.Time
+	mu       sync.Mutex
+	started  bool
+	consumed bool
+	done     chan map[string]any
+}
+
+func NewIssuerProbe(cfg Config, client HTTPClient, opts IssuerProbeOptions) (*IssuerProbe, error) {
+	state, err := base64.RawURLEncoding.DecodeString(opts.State)
+	if err != nil || len(state) != 32 || base64.RawURLEncoding.EncodeToString(state) != opts.State {
+		return nil, errors.New("probe state must be a canonical 32-byte base64url value")
+	}
+	redirect, err := url.Parse(cfg.RedirectURI)
+	if err != nil || redirect.Scheme != "https" || redirect.Host == "" || redirect.User != nil || redirect.RawQuery != "" || redirect.Fragment != "" || redirect.Path != "/authorize/callback" {
+		return nil, errors.New("probe requires the existing HTTPS /authorize/callback registration")
+	}
+	if cfg.ClientID == "" || cfg.ClientSecret == "" || len(opts.AllowedIssuers) == 0 || opts.Lifetime <= 0 || opts.Lifetime > 10*time.Minute {
+		return nil, errors.New("probe requires the existing client, reviewed issuers and a lifetime of at most ten minutes")
+	}
+	for _, issuer := range opts.AllowedIssuers {
+		u, err := url.Parse(issuer)
+		if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" {
+			return nil, errors.New("invalid reviewed issuer")
+		}
+	}
+	if client == nil {
+		client = &http.Client{Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
+	}
+	nonce := make([]byte, 32)
+	if _, err := rand.Read(nonce); err != nil {
+		return nil, errors.New("probe entropy unavailable")
+	}
+	return &IssuerProbe{adapter: New(cfg, client), opts: opts, host: redirect.Host, callback: redirect.Path, nonce: base64.RawURLEncoding.EncodeToString(nonce), expires: time.Now().Add(opts.Lifetime), done: make(chan map[string]any, 1)}, nil
+}
+
+// Done receives only allowlisted verification metadata, never a token or user.
+func (p *IssuerProbe) Done() <-chan map[string]any { return p.done }
+
+func (p *IssuerProbe) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Cache-Control", "no-store")
+	w.Header().Set("Referrer-Policy", "no-referrer")
+	w.Header().Set("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'")
+	w.Header().Set("X-Content-Type-Options", "nosniff")
+	if r.Method != http.MethodGet {
+		http.Error(w, "method refused", http.StatusMethodNotAllowed)
+		return
+	}
+	if r.URL.Path == "/healthz" {
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+	if r.Host != p.host {
+		http.NotFound(w, r)
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if time.Now().After(p.expires) || p.consumed {
+		http.Error(w, "probe expired or consumed", http.StatusGone)
+		return
+	}
+	if r.URL.Path == "/upstream-issuer-proof/"+p.opts.State {
+		if p.started {
+			http.Error(w, "probe already started", http.StatusConflict)
+			return
+		}
+		target, err := p.adapter.AuthorizeURL(r.Context(), domain.AuthRequest{State: p.opts.State})
+		if err != nil {
+			http.Error(w, "probe unavailable", http.StatusServiceUnavailable)
+			return
+		}
+		u, err := url.Parse(target)
+		if err != nil || u.Scheme != "https" || u.Host == "" {
+			http.Error(w, "probe unavailable", http.StatusServiceUnavailable)
+			return
+		}
+		q := u.Query()
+		q.Set("nonce", p.nonce)
+		u.RawQuery = q.Encode()
+		p.started = true
+		http.SetCookie(w, &http.Cookie{Name: "__Host-keycape-issuer-probe", Value: p.nonce, Path: "/", Secure: true, HttpOnly: true, SameSite: http.SameSiteLaxMode, MaxAge: int(p.opts.Lifetime.Seconds())})
+		http.Redirect(w, r, u.String(), http.StatusSeeOther)
+		return
+	}
+	if r.URL.Path != p.callback {
+		http.NotFound(w, r)
+		return
+	}
+	q, err := url.ParseQuery(r.URL.RawQuery)
+	cookie, cerr := r.Cookie("__Host-keycape-issuer-probe")
+	if err != nil || len(q["state"]) != 1 || subtle.ConstantTimeCompare([]byte(q.Get("state")), []byte(p.opts.State)) != 1 || !p.started || cerr != nil || subtle.ConstantTimeCompare([]byte(cookie.Value), []byte(p.nonce)) != 1 {
+		http.Error(w, "probe binding refused", http.StatusForbidden)
+		return
+	}
+	p.consumed = true
+	http.SetCookie(w, &http.Cookie{Name: "__Host-keycape-issuer-probe", Value: "", Path: "/", Secure: true, HttpOnly: true, SameSite: http.SameSiteLaxMode, MaxAge: -1})
+	receipt := map[string]any{"schema": "keycape.upstream-issuer-proof.v1", "observed_at": time.Now().UTC().Format(time.RFC3339), "status": "failed", "tokens_retained": false, "downstream_credential_issued": false}
+	if len(q["code"]) != 1 || q.Get("code") == "" || len(q["error"]) > 0 {
+		receipt["failure"] = "authorization_callback_refused"
+	} else {
+		ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
+		defer cancel()
+		if issuer, reason := p.verifyCode(ctx, q.Get("code")); reason != "" {
+			receipt["failure"] = reason
+		} else {
+			receipt["status"] = "verified"
+			receipt["issuer"] = issuer
+			receipt["signature_verified"] = true
+			receipt["audience_verified"] = true
+			receipt["validity_window_verified"] = true
+			receipt["nonce_verified"] = true
+		}
+	}
+	p.done <- receipt
+	// The browser learns only completion. Read the metadata receipt through the
+	// deployment owner's contained log collection, after cleaning up the route.
+	if receipt["status"] != "verified" {
+		http.Error(w, "Issuer proof failed. The operator has the diagnostic result.", http.StatusUnauthorized)
+		return
+	}
+	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	_, _ = io.WriteString(w, "Issuer proof complete. You may close this tab. No application login or downstream credential was issued.\n")
+}
+
+func (p *IssuerProbe) verifyCode(ctx context.Context, code string) (string, string) {
+	// Exchange under the same tokenBaseURL/client/redirect as the production
+	// adapter. Use a bounded reader and cancellation; never relay error bodies.
+	tokenURL := strings.TrimRight(p.adapter.tokenBaseURL(), "/") + "/api/oidc/token"
+	form := url.Values{"grant_type": {"authorization_code"}, "code": {code}, "redirect_uri": {p.adapter.cfg.RedirectURI}, "client_id": {p.adapter.cfg.ClientID}}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, tokenURL, strings.NewReader(form.Encode()))
+	if err != nil {
+		return "", "token_exchange_error"
+	}
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	req.SetBasicAuth(p.adapter.cfg.ClientID, p.adapter.cfg.ClientSecret)
+	response, err := p.adapter.client.Do(req)
+	if err != nil {
+		return "", "token_exchange_error"
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return "", "token_exchange_refused"
+	}
+	raw, err := io.ReadAll(io.LimitReader(response.Body, (1<<20)+1))
+	if err != nil || len(raw) > 1<<20 {
+		return "", "token_response_invalid"
+	}
+	var tokens tokenResponse
+	if json.Unmarshal(raw, &tokens) != nil {
+		return "", "token_response_invalid"
+	}
+	// Parsing chooses among operator-reviewed strings only. Acceptance still
+	// requires independent signature, audience and time verification below.
+	untrusted, err := parseIDTokenClaims(tokens.IDToken)
+	if err != nil {
+		return "", "id_token_verification_error"
+	}
+	issuer := stringClaim(untrusted, "iss")
+	allowed := false
+	for _, candidate := range p.opts.AllowedIssuers {
+		if issuer == candidate {
+			allowed = true
+		}
+	}
+	if !allowed {
+		return "", "issuer_outside_reviewed_set"
+	}
+	cfg := p.adapter.cfg
+	cfg.Issuer = issuer
+	verified, err := newIDTokenVerifier(cfg, p.adapter.client, p.adapter.tokenBaseURL()).Verify(ctx, tokens.IDToken)
+	if err != nil {
+		return "", FailureReason(err)
+	}
+	if subtle.ConstantTimeCompare([]byte(stringClaim(verified, "nonce")), []byte(p.nonce)) != 1 {
+		return "", "id_token_nonce_mismatch"
+	}
+	return issuer, ""
+}
+
+// RunIssuerProbe starts only this diagnostic, not the issuer, user directory or
+// MFA adapters. It decodes only the Authelia section of the existing config.
+func RunIssuerProbe(ctx context.Context, args []string, out io.Writer) error {
+	fs := flag.NewFlagSet("probe-upstream-issuer", flag.ContinueOnError)
+	fs.SetOutput(io.Discard)
+	configPath := fs.String("config", "", "existing config file")
+	listen := fs.String("listen", ":8081", "private probe listener")
+	issuers := fs.String("allowed-issuers", "", "comma-separated reviewed issuer strings")
+	stateEnv := fs.String("state-env", "KEYCAPE_ISSUER_PROBE_STATE", "variable holding random probe state")
+	lifetime := fs.Duration("lifetime", 10*time.Minute, "one-shot lifetime, at most ten minutes")
+	if fs.Parse(args) != nil || fs.NArg() != 0 {
+		return errors.New("invalid issuer probe arguments")
+	}
+	raw, err := os.ReadFile(*configPath)
+	if err != nil {
+		return errors.New("issuer probe config unavailable")
+	}
+	var sections map[string]yaml.Node
+	if yaml.Unmarshal(raw, &sections) != nil {
+		return errors.New("issuer probe config invalid")
+	}
+	node, ok := sections["authelia"]
+	if !ok {
+		return errors.New("issuer probe upstream config absent")
+	}
+	var cfg Config
+	if node.Decode(&cfg) != nil {
+		return errors.New("issuer probe upstream config invalid")
+	}
+	cfg.ClientSecret = os.ExpandEnv(cfg.ClientSecret)
+	probe, err := NewIssuerProbe(cfg, nil, IssuerProbeOptions{State: os.Getenv(*stateEnv), AllowedIssuers: strings.Split(*issuers, ","), Lifetime: *lifetime})
+	if err != nil {
+		return err
+	}
+	listener, err := net.Listen("tcp", *listen)
+	if err != nil {
+		return errors.New("issuer probe listener unavailable")
+	}
+	server := &http.Server{Handler: probe, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 40 * time.Second, IdleTimeout: 5 * time.Second, MaxHeaderBytes: 16384, ErrorLog: log.New(io.Discard, "", 0)}
+	runErr := make(chan error, 1)
+	go func() { runErr <- server.Serve(listener) }()
+	bounded, cancel := context.WithTimeout(ctx, *lifetime)
+	defer cancel()
+	var receipt map[string]any
+	select {
+	case receipt = <-probe.Done():
+	case <-bounded.Done():
+		receipt = map[string]any{"schema": "keycape.upstream-issuer-proof.v1", "status": "failed", "failure": "probe_deadline", "tokens_retained": false, "downstream_credential_issued": false}
+	case <-runErr:
+		receipt = map[string]any{"schema": "keycape.upstream-issuer-proof.v1", "status": "failed", "failure": "probe_listener_stopped", "tokens_retained": false, "downstream_credential_issued": false}
+	}
+	shutdown, stop := context.WithTimeout(context.Background(), 5*time.Second)
+	defer stop()
+	if server.Shutdown(shutdown) != nil {
+		_ = server.Close()
+	}
+	if json.NewEncoder(out).Encode(receipt) != nil {
+		return errors.New("issuer proof receipt could not be written")
+	}
+	if receipt["status"] != "verified" {
+		return errors.New("issuer proof failed")
+	}
+	return nil
+}
diff --git a/src/internal/adapters/authelia/issuer_probe_test.go b/src/internal/adapters/authelia/issuer_probe_test.go
new file mode 100644
index 0000000..d583942
--- /dev/null
+++ b/src/internal/adapters/authelia/issuer_probe_test.go
@@ -0,0 +1,258 @@
+package authelia_test
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"keycape/internal/adapters/authelia"
+)
+
+const probeState = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
+type probeProvider struct {
+	provider
+	calls       int
+	tokenStatus int
+	tokenBody   string
+	t           *testing.T
+}
+
+func (p *probeProvider) Do(r *http.Request) (*http.Response, error) {
+	if r.URL.Path == "/api/oidc/token" {
+		p.calls++
+		user, pass, ok := r.BasicAuth()
+		if !ok || user != testConfig().ClientID || pass != testConfig().ClientSecret {
+			p.t.Fatal("wrong client authentication")
+		}
+		if r.Context().Err() != nil {
+			return nil, r.Context().Err()
+		}
+		if err := r.ParseForm(); err != nil {
+			p.t.Fatal(err)
+		}
+		if r.Form.Get("redirect_uri") != "https://kc.example.com/authorize/callback" || r.Form.Get("code") != "private-auth-code" {
+			p.t.Fatal("wrong exchange binding")
+		}
+		if p.tokenStatus != 0 {
+			return &http.Response{StatusCode: p.tokenStatus, Body: io.NopCloser(strings.NewReader(p.tokenBody))}, nil
+		}
+	}
+	return p.provider.Do(r)
+}
+
+func newProbe(t *testing.T, p *probeProvider, lifetime time.Duration) *authelia.IssuerProbe {
+	t.Helper()
+	p.t = t
+	cfg := testConfig()
+	cfg.RedirectURI = "https://kc.example.com/authorize/callback"
+	cfg.BrowserBaseURL = "https://auth.example.com"
+	probe, err := authelia.NewIssuerProbe(cfg, p, authelia.IssuerProbeOptions{State: probeState, AllowedIssuers: []string{testIssuer, "http://authelia.sso.svc.cluster.local:9091"}, Lifetime: lifetime})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return probe
+}
+func startProbe(t *testing.T, probe *authelia.IssuerProbe) (*http.Cookie, string) {
+	t.Helper()
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest("GET", "https://kc.example.com/upstream-issuer-proof/"+probeState, nil)
+	probe.ServeHTTP(w, r)
+	if w.Code != http.StatusSeeOther {
+		t.Fatalf("start status %d", w.Code)
+	}
+	target, err := url.Parse(w.Header().Get("Location"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if target.Host != "auth.example.com" || target.Query().Get("state") != probeState || target.Query().Get("redirect_uri") != "https://kc.example.com/authorize/callback" {
+		t.Fatal("wrong upstream authorization binding")
+	}
+	cookies := w.Result().Cookies()
+	if len(cookies) != 1 || !cookies[0].Secure || !cookies[0].HttpOnly || cookies[0].SameSite != http.SameSiteLaxMode {
+		t.Fatal("browser binding cookie missing")
+	}
+	return cookies[0], target.Query().Get("nonce")
+}
+func completeProbe(probe *authelia.IssuerProbe, cookie *http.Cookie, query string) *httptest.ResponseRecorder {
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest("GET", "https://kc.example.com/authorize/callback?"+query, nil)
+	if cookie != nil {
+		r.AddCookie(cookie)
+	}
+	probe.ServeHTTP(w, r)
+	return w
+}
+
+func TestIssuerProbeVerifiesSignedIssuerWithoutDisclosingTokens(t *testing.T) {
+	for _, issuer := range []string{testIssuer, "http://authelia.sso.svc.cluster.local:9091"} {
+		t.Run(issuer, func(t *testing.T) {
+			p := &probeProvider{provider: provider{jwks: testJWKS(nil)}}
+			probe := newProbe(t, p, time.Minute)
+			cookie, nonce := startProbe(t, probe)
+			p.idToken = signIDToken(map[string]interface{}{"iss": issuer, "nonce": nonce, "preferred_username": "private-user", "email": "private-email"}, testKeyID, nil)
+			w := completeProbe(probe, cookie, "state="+probeState+"&code=private-auth-code")
+			if w.Code != http.StatusOK {
+				t.Fatalf("status %d", w.Code)
+			}
+			receipt := <-probe.Done()
+			if receipt["status"] != "verified" || receipt["issuer"] != issuer || receipt["nonce_verified"] != true {
+				t.Fatal("missing verified issuer receipt")
+			}
+			raw, _ := json.Marshal(receipt)
+			for _, secret := range []string{p.idToken, "private-user", "private-email", "private-auth-code", testConfig().ClientSecret, nonce} {
+				if strings.Contains(string(raw)+w.Body.String(), secret) {
+					t.Fatal("diagnostic disclosed private material")
+				}
+			}
+			if p.calls != 1 {
+				t.Fatal("wrong exchange count")
+			}
+			if completeProbe(probe, cookie, "state="+probeState+"&code=private-auth-code").Code != http.StatusGone || p.calls != 1 {
+				t.Fatal("callback was replayed")
+			}
+		})
+	}
+}
+
+func TestIssuerProbeRefusesUnboundCallbacksBeforeExchange(t *testing.T) {
+	p := &probeProvider{provider: provider{jwks: testJWKS(nil)}}
+	probe := newProbe(t, p, time.Minute)
+	if completeProbe(probe, nil, "state="+probeState+"&code=private-auth-code").Code != http.StatusForbidden {
+		t.Fatal("callback before start accepted")
+	}
+	cookie, nonce := startProbe(t, probe)
+	p.idToken = signIDToken(map[string]interface{}{"nonce": nonce}, testKeyID, nil)
+	for _, query := range []string{"state=wrong&code=private-auth-code", "state=" + probeState + "&state=other&code=private-auth-code", "state=%xx&code=private-auth-code"} {
+		if completeProbe(probe, cookie, query).Code != http.StatusForbidden {
+			t.Fatal("invalid state accepted")
+		}
+	}
+	if completeProbe(probe, nil, "state="+probeState+"&code=private-auth-code").Code != http.StatusForbidden {
+		t.Fatal("missing browser cookie accepted")
+	}
+	if p.calls != 0 {
+		t.Fatal("unbound callback exchanged a code")
+	}
+	if completeProbe(probe, cookie, "state="+probeState+"&code=private-auth-code").Code != http.StatusOK {
+		t.Fatal("invalid callback consumed valid flow")
+	}
+}
+
+func TestIssuerProbeRejectsTokensAndWritesOnlyFailureCategory(t *testing.T) {
+	other, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cases := []struct {
+		name    string
+		claims  map[string]interface{}
+		key     *rsa.PrivateKey
+		status  int
+		body    string
+		jwksErr error
+	}{
+		{name: "unreviewed issuer", claims: map[string]interface{}{"iss": "https://private-attacker.example"}},
+		{name: "wrong audience", claims: map[string]interface{}{"aud": "private-wrong-client"}},
+		{name: "expired", claims: map[string]interface{}{"iat": time.Now().Add(-2 * time.Hour).Unix(), "exp": time.Now().Add(-time.Hour).Unix()}},
+		{name: "wrong nonce", claims: map[string]interface{}{"nonce": "private-wrong-nonce"}},
+		{name: "forged signature", key: other},
+		{name: "unavailable keys", jwksErr: errors.New("private-provider-body")},
+		{name: "provider refused", status: 401, body: "private-provider-body"},
+		{name: "provider malformed", status: 200, body: "private-provider-body"},
+		{name: "oversize", status: 200, body: strings.Repeat("x", (1<<20)+1)},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			p := &probeProvider{provider: provider{jwks: testJWKS(nil), jwksErr: tc.jwksErr}, tokenStatus: tc.status, tokenBody: tc.body}
+			probe := newProbe(t, p, time.Minute)
+			cookie, nonce := startProbe(t, probe)
+			claims := map[string]interface{}{"nonce": nonce}
+			for k, v := range tc.claims {
+				claims[k] = v
+			}
+			p.idToken = signIDToken(claims, testKeyID, tc.key)
+			w := completeProbe(probe, cookie, "state="+probeState+"&code=private-auth-code")
+			if w.Code != http.StatusUnauthorized {
+				t.Fatalf("bad token accepted: %d", w.Code)
+			}
+			receipt := <-probe.Done()
+			if receipt["status"] != "failed" || receipt["issuer"] != nil {
+				t.Fatal("unverified issuer receipt")
+			}
+			raw, _ := json.Marshal(receipt)
+			if strings.Contains(string(raw)+w.Body.String(), "private-") || strings.Contains(string(raw), p.idToken) {
+				t.Fatal("failure disclosed material")
+			}
+		})
+	}
+}
+
+func TestIssuerProbeBoundsLifetimeAndConfiguration(t *testing.T) {
+	p := &probeProvider{}
+	probe := newProbe(t, p, time.Nanosecond)
+	time.Sleep(time.Millisecond)
+	if completeProbe(probe, nil, "state="+probeState).Code != http.StatusGone {
+		t.Fatal("expired probe accepted")
+	}
+	cfg := testConfig()
+	cfg.RedirectURI = "https://kc.example.com/authorize/callback"
+	for _, state := range []string{"", base64.RawURLEncoding.EncodeToString(make([]byte, 16)), probeState + "="} {
+		if _, err := authelia.NewIssuerProbe(cfg, nil, authelia.IssuerProbeOptions{State: state, AllowedIssuers: []string{testIssuer}, Lifetime: time.Minute}); err == nil {
+			t.Fatal("weak state accepted")
+		}
+	}
+	if _, err := authelia.NewIssuerProbe(cfg, nil, authelia.IssuerProbeOptions{State: probeState, AllowedIssuers: []string{testIssuer}, Lifetime: 11 * time.Minute}); err == nil {
+		t.Fatal("unbounded lifetime accepted")
+	}
+	for _, uri := range []string{"http://kc.example.com/authorize/callback", "https://kc.example.com/new-callback", "https://kc.example.com/authorize/callback?code=secret"} {
+		cfg.RedirectURI = uri
+		if _, err := authelia.NewIssuerProbe(cfg, nil, authelia.IssuerProbeOptions{State: probeState, AllowedIssuers: []string{testIssuer}, Lifetime: time.Minute}); err == nil {
+			t.Fatal("changed callback accepted")
+		}
+	}
+}
+
+func TestIssuerProbeErrorsDoNotPrintArguments(t *testing.T) {
+	var out strings.Builder
+	err := authelia.RunIssuerProbe(context.Background(), []string{"--private-secret=never-print"}, &out)
+	if err == nil || strings.Contains(fmt.Sprint(err)+out.String(), "never-print") {
+		t.Fatal("argument disclosure")
+	}
+}
+
+func TestIssuerProbeCLIReadsOnlyUpstreamAndBoundsIdleRun(t *testing.T) {
+	cfg := filepath.Join(t.TempDir(), "config.yaml")
+	content := "authelia:\n  baseURL: https://auth.example.com\n  clientId: keycape\n  clientSecret: private-config-secret\n  redirectURI: https://kc.example.com/authorize/callback\nprivateKeyPEM: /must-not-be-read\nlldap:\n  password: private-unrelated-password\n"
+	if err := os.WriteFile(cfg, []byte(content), 0600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PROBE_TEST_STATE", probeState)
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	var out strings.Builder
+	err := authelia.RunIssuerProbe(ctx, []string{"--config=" + cfg, "--state-env=PROBE_TEST_STATE", "--allowed-issuers=https://auth.example.com", "--listen=127.0.0.1:0", "--lifetime=1s"}, &out)
+	if err == nil {
+		t.Fatal("cancelled probe succeeded")
+	}
+	var receipt map[string]interface{}
+	if json.Unmarshal([]byte(out.String()), &receipt) != nil || receipt["failure"] != "probe_deadline" {
+		t.Fatal("expected bounded diagnostic receipt")
+	}
+	if strings.Contains(out.String()+err.Error(), "private-") {
+		t.Fatal("config material disclosed")
+	}
+}
diff --git a/workplans/KEY-WP-0013-approval-engine-resource-audience.md b/workplans/KEY-WP-0013-approval-engine-resource-audience.md
index e924a54..bcc271d 100644
--- a/workplans/KEY-WP-0013-approval-engine-resource-audience.md
+++ b/workplans/KEY-WP-0013-approval-engine-resource-audience.md
@@ -213,3 +213,33 @@ proof remain distinct gates. Public discovery currently advertises
 `https://auth.coulomb.social`; that alone is not the signed-token observation.
 
 T06 completion: full Go suite and vet passed; published code `dcebd46` and pulled the image by digest `sha256:7ff54c54e63ee172ae9e6e7fd2da96e427352f712343d74626ee6fe0f6f82611`. Container `keycape verify-client --help` confirms the command is present. `docs/approval-clients-rollout.md` and its proposed deployment patch record configuration/issuer, first-provision, single-instance rollout, acceptance and rollback. This closes verifier preparation only; T02 and T05 retain the explicit live dependencies.
+
+
+## Prepare a bounded upstream issuer observation before client rollout
+
+```task
+id: KEY-WP-0013-T07
+status: progress
+priority: high
+assignee: the-custodian
+```
+
+2026-09-08: T02's actual upstream signed-token issuer gate had no executable
+observation path. A downstream OpenBao/KeyCape login cannot return the upstream
+Authelia token; it remains inside the adapter. Prepare `probe-upstream-issuer`
+and an exact-state temporary route so one attended flow can be verified before
+changing the normal KeyCape Deployment or its configuration.
+
+Acceptance: signed issuer/audience/time/nonce proof without token or user-claim
+output; browser/state binding and one-shot refusal; finite issuer allowlist;
+ten-minute bound; no downstream credential; current Traefik route isolation;
+immutable image; reviewed temporary Job/config projection, network policies and
+owner-reference cleanup. Normal callbacks and the production issuer stay on
+the existing Deployment. The probe mounts only `config.yaml`, not `key.pem`,
+and has no Kubernetes API token. That file contains existing credential data;
+its use by the temporary diagnostic needs deployment-owner admission.
+
+Source tests, image and deployment packet close this preparation task. The
+attended live receipt and issuer pin remain in T02, alongside the named CCR
+reviews and custody/rollout acceptance. Preparing the diagnostic does not
+approve CCR-2026-0017/0018 or close any live factory gate.