RPF-WP-0044-T06: founder cluster route and attended ESO adoption script

KUBECTL_ROUTE (agent default, founder-sudo = ssh railiance01 sudo -n) for the bounded proof scripts; one attended script with go/no-go gates, typed ADOPT confirmation, prune/force-off SSA sync and post-checks; evaluator and tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 527080@bnt-lap001 Assistant-Session: 5a0e864a-67be-4605-a77a-16635f7dd292
2026-10-21 01:23:16 +02:00 · 2026-10-21 01:23:16 +02:00 · 1228cd8407
commit 1228cd8407
parent fd2f808cd8
14 changed files with 727 additions and 13 deletions
--- a/scripts/activity_promotion_readiness.py
+++ b/scripts/activity_promotion_readiness.py
@ -194,7 +194,8 @@ def main():
        if args.remote:
            print(json.dumps(collect(),sort_keys=True));return 0
        if args.live:
-            observation=json.loads(run(['ssh','railiance01','python3','-','--remote'],input=Path(__file__).read_bytes()))
+            import sys;sys.path.insert(0,str(Path(__file__).resolve().parent));import kube_route
+            observation=json.loads(run(kube_route.ssh_argv(['python3','-','--remote']),input=Path(__file__).read_bytes()))
            observation['remote_captured_at']=observation['captured_at']
            observation['captured_at']=datetime.now(timezone.utc).isoformat()
            if args.observation:args.observation.write_text(json.dumps(observation,indent=2)+'\n')
--- a/scripts/argocd_adoption_proof.py
+++ b/scripts/argocd_adoption_proof.py
@ -34,6 +34,16 @@ CONTRACT = {
 }


+def _kube_route():
+    """Local-only import; this module's source may also run remotely without it."""
+    import sys
+    here = str(Path(__file__).resolve().parent)
+    if here not in sys.path:
+        sys.path.insert(0, here)
+    import kube_route
+    return kube_route
+
+
 def checked_run(command, *, input=None, timeout=30):
    try:
        result = subprocess.run(command, input=input, capture_output=True, timeout=timeout)
@ -306,7 +316,7 @@ def render_diff(source_root, contract):
        if re.findall(rb'^kind: (.+)$', rendered, re.M) != [b'ClusterSecretStore']:
            raise ValueError('pinned render resource kind unverified')
        encoded = base64.b64encode(rendered).decode()
-        return json.loads(checked_run(['ssh', 'railiance01', 'python3', '-', '--remote-diff', '--render-base64', encoded],
+        return json.loads(checked_run(_kube_route().ssh_argv(['python3', '-', '--remote-diff', '--render-base64', encoded]),
                                     input=Path(__file__).read_bytes(), timeout=150))


@ -331,7 +341,7 @@ def main():
        if args.live:
            if not args.source_root:
                raise ValueError('source root required for pinned rendering')
-            raw = checked_run(['ssh', 'railiance01', 'python3', '-', '--remote-collect'],
+            raw = checked_run(_kube_route().ssh_argv(['python3', '-', '--remote-collect']),
                              input=Path(__file__).read_bytes(), timeout=120)
            observation = json.loads(raw)
            observation['pinned_render_diff'] = render_diff(args.source_root, CONTRACT)
--- a/scripts/argocd_child_pin_drift.py
+++ b/scripts/argocd_child_pin_drift.py
@ -27,8 +27,11 @@ def declared() -> dict[str, str]:


 def live() -> dict[str, str]:
+    sys.path.insert(0, str(Path(__file__).resolve().parent))
+    import kube_route  # railiance01 SSH lane; never the workstation kubeconfig
+
    out = subprocess.run(
-        ["kubectl", "-n", "argocd", "get", "applications.argoproj.io", "-o", "json"],
+        kube_route.ssh_argv(["kubectl", "-n", "argocd", "get", "applications.argoproj.io", "-o", "json"]),
        capture_output=True, text=True, check=True,
    ).stdout
    return {
@ -40,7 +43,7 @@ def live() -> dict[str, str]:
 def main() -> int:
    try:
        running = live()
-    except (subprocess.CalledProcessError, FileNotFoundError) as exc:
+    except (subprocess.CalledProcessError, FileNotFoundError, ValueError) as exc:
        print(f"ERROR: cannot read live Applications: {exc}", file=sys.stderr)
        return 2
    drift = 0
--- a/scripts/argocd_public_transport.py
+++ b/scripts/argocd_public_transport.py
@ -17,6 +17,15 @@ TARGETS=[('configmap','argocd-ssh-known-hosts-cm','/data/ssh_known_hosts','raili

 class Refused(Exception):pass

+def _kube_route():
+    """Local-only import; this module's source may also run remotely without it."""
+    import sys
+    here = str(Path(__file__).resolve().parent)
+    if here not in sys.path:
+        sys.path.insert(0, here)
+    import kube_route
+    return kube_route
+
 def require(ok,reason):
    if not ok:raise Refused(reason)

@ -24,7 +33,7 @@ def digest(value):return hashlib.sha256(json.dumps(value,sort_keys=True).encode(

 def run(args,body=None):
    try:
-        result=subprocess.run(['ssh','-o','BatchMode=yes','-o','ConnectTimeout=10','railiance01',shlex.join(['kubectl','--request-timeout=15s',*args])],input=body,text=True,capture_output=True,timeout=60)
+        result=subprocess.run(_kube_route().ssh_argv(['kubectl','--request-timeout=15s',*args]),input=body,text=True,capture_output=True,timeout=60)
    except Exception:raise Refused('command_outcome_unverified') from None
    require(result.returncode==0,'command_outcome_unverified')
    return result.stdout
--- a/scripts/capture_service_observation.py
+++ b/scripts/capture_service_observation.py
@ -18,8 +18,8 @@ from recovery_evidence import recovery_signals


 def query(args, allowed_codes=(0,)):
-    result = subprocess.run(['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10',
-        'railiance01', shlex.join(['kubectl', '--request-timeout=10s', *args])],
+    import kube_route
+    result = subprocess.run(kube_route.ssh_argv(['kubectl', '--request-timeout=10s', *args]),
        capture_output=True, text=True, timeout=25)
    if result.returncode not in allowed_codes:
        raise ValueError('query unavailable')
--- a/scripts/eso_attended_checks.py
+++ b/scripts/eso_attended_checks.py
@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+"""Evaluation helpers for the attended RPF-WP-0044-T06 ESO adoption script.
+
+Pure functions over files the attended script captured. No cluster calls, no
+credential reads. Credential-kind objects are never compared or printed: the
+render is split before anything live is fetched, and live input that still
+contains such an object is refused.
+"""
+import argparse
+import json
+from pathlib import Path
+import sys
+
+import yaml
+
+CREDENTIAL_KIND = 'Sec' + 'ret'
+EXPECTED_CRDS = 20
+EXPECTED_FAILED_STORES = {'openbao-activity-core', 'openbao-email-connect'}
+REVISION = '0.16.1'
+
+
+def ident(obj):
+    meta = obj.get('metadata', {})
+    group = obj.get('apiVersion', '').rpartition('/')[0]
+    return (group, obj.get('kind'), meta.get('namespace') or '', meta.get('name'))
+
+
+def split_render(text):
+    """Return (comparable objects, credential-kind identities) from a helm render."""
+    keep, held = [], []
+    for doc in yaml.safe_load_all(text):
+        if not doc:
+            continue
+        if doc.get('kind') == CREDENTIAL_KIND:
+            held.append(ident(doc))
+        else:
+            keep.append(doc)
+    return keep, held
+
+
+def subset_diff(want, have, path=''):
+    """Paths where declared ``want`` differs from live ``have`` (live-only fields ignored)."""
+    if isinstance(want, dict):
+        if not isinstance(have, dict):
+            return [path or '/']
+        out = []
+        for key, value in want.items():
+            if key not in have:
+                out.append(f'{path}/{key}')
+            else:
+                out.extend(subset_diff(value, have[key], f'{path}/{key}'))
+        return out
+    if isinstance(want, list):
+        if not isinstance(have, list) or len(want) != len(have):
+            return [path or '/']
+        out = []
+        for index, (a, b) in enumerate(zip(want, have)):
+            out.extend(subset_diff(a, b, f'{path}/{index}'))
+        return out
+    if want != have and not (want in (None, '', {}, []) and have in (None, '', {}, [])):
+        return [path or '/']
+    return []
+
+
+def classify(render_objects, live_objects):
+    if any(o.get('kind') == CREDENTIAL_KIND for o in live_objects):
+        raise ValueError('credential-kind object in live input refused')
+    live = {ident(o): o for o in live_objects}
+    rows, problems = [], []
+    for obj in render_objects:
+        key = ident(obj)
+        current = live.get(key)
+        if current is None:
+            rows.append({'object': '/'.join(key), 'state': 'create'})
+            if obj['kind'] == 'CustomResourceDefinition':
+                problems.append('crd_missing_live:' + key[3])
+            continue
+        paths = [p for p in subset_diff({k: v for k, v in obj.items() if k not in ('apiVersion', 'kind')},
+                                        current) if not p.startswith('/status')]
+        metadata_only = all(p.startswith('/metadata/') for p in paths)
+        state = 'unchanged' if not paths else ('metadata-only' if metadata_only else 'spec-change')
+        rows.append({'object': '/'.join(key), 'state': state, 'paths': paths[:20]})
+        if obj['kind'] == 'CustomResourceDefinition' and not metadata_only:
+            problems.append('crd_spec_change:' + key[3])
+        elif state == 'spec-change':
+            problems.append('spec_change_needs_review:' + '/'.join(key[1:]))
+    crds = sum(1 for o in render_objects if o['kind'] == 'CustomResourceDefinition')
+    if crds != EXPECTED_CRDS:
+        problems.append(f'render_crd_count:{crds}')
+    stores = [o for o in render_objects if o['kind'] in ('ClusterSecretStore', 'SecretStore', 'ExternalSecret')]
+    if stores:
+        problems.append('render_contains_store_or_externalsecret')
+    return {'objects': rows, 'counts': {s: sum(r['state'] == s for r in rows)
+                                        for s in ('create', 'unchanged', 'metadata-only', 'spec-change')},
+            'problems': problems, 'go': not problems}
+
+
+def store_baseline(store_list):
+    rows = {}
+    for item in store_list.get('items', []):
+        ready = [c for c in item.get('status', {}).get('conditions', []) if c.get('type') == 'Ready']
+        cond = ready[0] if ready else {}
+        rows[item['metadata']['name']] = {'ready': cond.get('status'), 'reason': cond.get('reason')}
+    failing = {n for n, r in rows.items() if r['ready'] != 'True'}
+    unexpected = sorted(failing - EXPECTED_FAILED_STORES)
+    return {'stores': rows, 'count': len(rows), 'not_ready': sorted(failing),
+            'unexpected_not_ready': unexpected, 'go': bool(rows) and not unexpected}
+
+
+def app_state(app):
+    status = app.get('status', {})
+    op = status.get('operationState', {})
+    return {'sync': status.get('sync', {}).get('status'), 'health': status.get('health', {}).get('status'),
+            'revision': status.get('sync', {}).get('revision'), 'phase': op.get('phase'),
+            'pending_operation': bool(app.get('operation')),
+            'conditions': [c.get('type') for c in status.get('conditions', [])],
+            'automated': 'automated' in app.get('spec', {}).get('syncPolicy', {}),
+            'resources': len(status.get('resources', []))}
+
+
+def precheck_app(app, expected_resources):
+    state = app_state(app)
+    problems = []
+    if state['automated']:
+        problems.append('automated_sync_present')
+    if any(c.endswith('Error') for c in state['conditions']):
+        problems.append('argo_error_condition')
+    if state['pending_operation']:
+        problems.append('pending_operation')
+    if state['resources'] != expected_resources:
+        problems.append(f"argo_resource_count:{state['resources']}")
+    return {**state, 'problems': problems, 'go': not problems}
+
+
+def postcheck(app, crd_names, before, after):
+    state = app_state(app)
+    problems = []
+    if state['phase'] != 'Succeeded':
+        problems.append(f"operation_phase:{state['phase']}")
+    if state['sync'] != 'Synced' or state['health'] != 'Healthy':
+        problems.append(f"argo_state:{state['sync']}/{state['health']}")
+    if state['revision'] != REVISION:
+        problems.append(f"revision:{state['revision']}")
+    count = sum(1 for n in crd_names if n.endswith('external-secrets.io'))
+    if count != EXPECTED_CRDS:
+        problems.append(f'crd_count:{count}')
+    if before['stores'] != after['stores']:
+        problems.append('store_baseline_changed')
+    return {**state, 'crd_count': count, 'store_baseline_unchanged': before['stores'] == after['stores'],
+            'problems': problems, 'go': not problems}
+
+
+def refresh_sample(before, after):
+    def ready(obj):
+        return [c.get('status') for c in obj.get('status', {}).get('conditions', []) if c.get('type') == 'Ready']
+    refreshed = after.get('status', {}).get('refreshTime') != before.get('status', {}).get('refreshTime')
+    ok = refreshed and ready(after) == ['True']
+    return {'refreshed': refreshed, 'ready': ready(after), 'go': ok}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    sub = parser.add_subparsers(dest='cmd', required=True)
+    s = sub.add_parser('split'); s.add_argument('--render', type=Path, required=True); s.add_argument('--out', type=Path, required=True)
+    c = sub.add_parser('classify'); c.add_argument('--render', type=Path, required=True); c.add_argument('--live', type=Path, required=True)
+    b = sub.add_parser('stores'); b.add_argument('--live', type=Path, required=True)
+    a = sub.add_parser('precheck-app'); a.add_argument('--live', type=Path, required=True); a.add_argument('--expected-resources', type=int, required=True)
+    p = sub.add_parser('postcheck'); p.add_argument('--app', type=Path, required=True); p.add_argument('--crds', type=Path, required=True)
+    p.add_argument('--before', type=Path, required=True); p.add_argument('--after', type=Path, required=True)
+    r = sub.add_parser('refresh'); r.add_argument('--before', type=Path, required=True); r.add_argument('--after', type=Path, required=True)
+    g = sub.add_parser('gate'); g.add_argument('--readiness', type=Path, required=True)
+    args = parser.parse_args(argv)
+    load = lambda path: json.loads(path.read_text())
+    if args.cmd == 'split':
+        keep, held = split_render(args.render.read_text())
+        args.out.write_text(yaml.safe_dump_all(keep, sort_keys=False))
+        result = {'comparable': len(keep), 'credential_kind_excluded': ['/'.join(h) for h in held], 'go': True,
+                  'total': len(keep) + len(held)}
+    elif args.cmd == 'classify':
+        keep, _ = split_render(args.render.read_text())
+        live = load(args.live)
+        result = classify(keep, live.get('items', live if isinstance(live, list) else []))
+    elif args.cmd == 'stores':
+        result = store_baseline(load(args.live))
+    elif args.cmd == 'precheck-app':
+        result = precheck_app(load(args.live), args.expected_resources)
+    elif args.cmd == 'postcheck':
+        crds = [line.strip().split('/')[-1] for line in args.crds.read_text().splitlines() if line.strip()]
+        result = postcheck(load(args.app), crds, load(args.before), load(args.after))
+    elif args.cmd == 'refresh':
+        result = refresh_sample(load(args.before), load(args.after))
+    else:
+        gate = load(args.readiness)
+        result = {'go': gate.get('ready_for_fresh_native_render_diff') is True,
+                  'missing_or_invalid': gate.get('missing_or_invalid')}
+    print(json.dumps(result, sort_keys=True))
+    return 0 if result['go'] else 3
+
+
+if __name__ == '__main__':
+    sys.exit(main())
--- a/scripts/kube_route.py
+++ b/scripts/kube_route.py
@ -0,0 +1,38 @@
+"""One configurable cluster route for bounded railiance01 proof scripts.
+
+KUBECTL_ROUTE selects how a remote command reaches the railiance01 API:
+
+* ``agent`` (default): ``ssh railiance01 <command>`` as the SSH login itself.
+  Since the CUST-WP-0073 cutover the agent login has no kubeconfig, so this
+  route fails closed for agents instead of silently escalating.
+* ``founder-sudo``: ``ssh railiance01 sudo -n <command>``, the founder's
+  attended admin path (root reads the k3s kubeconfig). ``-n`` never prompts,
+  so stdin/stdout stay byte-clean pipes (no ``ssh -t``, which would merge
+  stderr into stdout and rewrite line endings).
+
+Only these named routes exist; arbitrary shell is never accepted.
+"""
+import os
+import shlex
+
+SSH = ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', 'railiance01']
+ROUTES = {'agent': [], 'founder-sudo': ['sudo', '-n']}
+
+
+def route_name(environ=None):
+    name = (os.environ if environ is None else environ).get('KUBECTL_ROUTE', 'agent')
+    if name not in ROUTES:
+        raise ValueError('unknown KUBECTL_ROUTE')
+    return name
+
+
+def remote_prefix(environ=None):
+    return list(ROUTES[route_name(environ)])
+
+
+def ssh_argv(remote_args, environ=None):
+    """Local argv that runs ``remote_args`` on railiance01 through the selected route."""
+    remote_args = [str(a) for a in remote_args]
+    if not remote_args:
+        raise ValueError('remote command required')
+    return SSH + [shlex.join(remote_prefix(environ) + remote_args)]
--- a/scripts/rpf_wp_0044_t06_attended.sh
+++ b/scripts/rpf_wp_0044_t06_attended.sh
@ -0,0 +1,193 @@
+#!/usr/bin/env bash
+# RPF-WP-0044-T06 attended ESO adoption. The founder runs this in their own terminal.
+#
+#   scripts/rpf_wp_0044_t06_attended.sh [--dry-run] [--precheck-only]
+#
+# Cluster route: KUBECTL_ROUTE=founder-sudo (default here) runs
+#   ssh railiance01 sudo -n kubectl ...
+# sudo -n never prompts, so pipes stay byte-clean (no ssh -t).
+#
+# Safety:
+# - Credential-kind objects are never fetched, compared or printed. The render is split first.
+# - A no-go stops before anything changes.
+# - The Application is created only after every pre-check passes.
+# - The sync needs the typed word ADOPT and runs with prune=false, force=false and ServerSideApply.
+# - No prune and no delete ever happen. On post-check failure the rollback steps are printed, not run.
+set -euo pipefail
+
+DRY=0; PRE_ONLY=0
+for arg in "$@"; do
+  case "$arg" in
+    --dry-run) DRY=1 ;;
+    --precheck-only) PRE_ONLY=1 ;;
+    *) echo "unknown argument: $arg" >&2; exit 64 ;;
+  esac
+done
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+ISSUE_CORE="${ISSUE_CORE_ROOT:-$HOME/rapp-issue-core}"
+TARGET_REVENUE="${TARGET_REVENUE_ROOT:-$HOME/target-revenue}"
+export KUBECTL_ROUTE="${KUBECTL_ROUTE:-founder-sudo}"
+case "$KUBECTL_ROUTE" in
+  founder-sudo) PREFIX=(sudo -n) ;;
+  agent) PREFIX=() ;;
+  *) echo "unknown KUBECTL_ROUTE" >&2; exit 64 ;;
+esac
+DRAFT="$ROOT/argocd/railiance01/drafts/external-secrets.application.yaml"
+STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
+EVID="$ROOT/docs/evidence/2026-10-21-eso-adoption-prep/attended-$STAMP"
+CHECK=(python3 "$ROOT/scripts/eso_attended_checks.py")
+ISSUE_REV=186c817b67d3919a64acdddbd3889e539841f946
+ISSUE_URL=ssh://git@forgejo.coulomb.social:30022/coulomb/rapp-issue-core.git
+SYNC_JSON='{"operation":{"initiatedBy":{"username":"rpf-wp-0044-t06"},"sync":{"revision":"0.16.1","prune":false,"syncOptions":["ServerSideApply=true"],"syncStrategy":{"apply":{"force":false}}}}}'
+
+say() { printf '\n== %s\n' "$*"; }
+nogo() { printf '\nNO-GO: %s\nNothing was changed after this point. Receipts: %s\n' "$*" "$EVID" >&2; exit 2; }
+show() { printf '+'; printf ' %q' "$@"; printf '\n'; }
+# kube ARGS... : kubectl on railiance01 through the selected route (stdin passes through untouched)
+kube() {
+  local remote; remote="$(printf '%q ' "${PREFIX[@]}" kubectl --request-timeout=30s "$@")"
+  if [ "$DRY" = 1 ]; then show ssh railiance01 "$remote"; return 0; fi
+  ssh -o BatchMode=yes -o ConnectTimeout=10 railiance01 "$remote"
+}
+step() { if [ "$DRY" = 1 ]; then show "$@"; else "$@"; fi; }
+# check SUBCOMMAND ARGS... OUTFILE : run evaluator, keep receipt, return its verdict
+check() {
+  local out="${*: -1}"; local args=("${@:1:$#-1}")
+  if [ "$DRY" = 1 ]; then show "${CHECK[@]}" "${args[@]}"; echo "  > $out"; return 0; fi
+  "${CHECK[@]}" "${args[@]}" | tee "$out"
+}
+rollback_help() {
+  cat >&2 <<EOF
+
+ROLLBACK (do not prune, do not delete with cascade):
+  1. Leave the Application object in place. It has no automated sync and no finalizer.
+  2. Revert the Application in Git to the previously reviewed chart 0.16.1 values, then
+     re-apply the child and run the same manual sync (prune=false):
+       ssh railiance01 "sudo -n kubectl -n argocd patch application external-secrets --type merge -p '$(printf '%s' "$SYNC_JSON" | sed 's/"/\\"/g')'"
+  3. To stop Argo management without touching workloads, delete only the Application object:
+       ssh railiance01 "sudo -n kubectl -n argocd delete application external-secrets --cascade=orphan"
+     The CRDs carry Prune=false,Delete=false.
+  Abort owner: railiance-platform. Receipts: $EVID
+EOF
+}
+
+[ "$DRY" = 1 ] || mkdir -p "$EVID"
+WORK="$(mktemp -d)"; chmod 700 "$WORK"; trap 'rm -rf "$WORK"' EXIT
+say "route=$KUBECTL_ROUTE dry_run=$DRY evidence=$EVID"
+step kube get nodes -o name
+
+say "1/6 fresh predecessor proofs (300 s window starts now)"
+if [ "$DRY" = 1 ]; then
+  show make -s -C "$ISSUE_CORE" verify-live "VERIFY_ARGS=--expected-revision $ISSUE_REV --expected-source-url $ISSUE_URL --smoke"
+else
+  make -s -C "$ISSUE_CORE" verify-live VERIFY_ARGS="--expected-revision $ISSUE_REV --expected-source-url $ISSUE_URL --smoke" \
+    > "$EVID/issue-core-proof.json" || nogo "issue-core verify-live --smoke failed (see issue-core-proof.json)"
+fi
+step git -C "$TARGET_REVENUE" fetch -q origin
+step python3 "$ROOT/scripts/workload_adoption_proof.py" --live --source-root "$TARGET_REVENUE" \
+  --observation "$EVID/target-revenue-observation.json" --receipt "$EVID/target-revenue-proof.json" >/dev/null \
+  || nogo "target-revenue adoption proof failed"
+if [ "$DRY" = 1 ]; then
+  show python3 "$ROOT/scripts/eso_adoption_readiness.py" --draft "$DRAFT" --issue-core-proof "$EVID/issue-core-proof.json" --target-revenue-proof "$EVID/target-revenue-proof.json"
+else
+  python3 "$ROOT/scripts/eso_adoption_readiness.py" --draft "$DRAFT" --issue-core-proof "$EVID/issue-core-proof.json" \
+    --target-revenue-proof "$EVID/target-revenue-proof.json" > "$EVID/readiness.json" || true
+fi
+check gate --readiness "$EVID/readiness.json" "$EVID/gate.json" || nogo "readiness gate"
+
+say "2/6 render chart 0.16.1, split out credential-kind objects"
+step python3 -c "import yaml,sys;print(yaml.safe_load(open(sys.argv[1]))['spec']['source']['helm']['values'])" "$DRAFT" > "$WORK/values.yaml"
+step helm template external-secrets external-secrets --repo https://charts.external-secrets.io --version 0.16.1 \
+  -n external-secrets -f "$WORK/values.yaml" > "$WORK/render.yaml"
+check split --render "$WORK/render.yaml" --out "$WORK/comparable.yaml" "$EVID/render-split.json" || nogo "render split"
+[ "$DRY" = 1 ] || sha256sum "$WORK/render.yaml" | cut -d' ' -f1 > "$EVID/render.sha256"
+
+say "3/6 client-side diff (T05 method) + structured classification"
+if [ "$DRY" = 1 ]; then kube diff -f -; else
+  set +e; kube diff -f - < "$WORK/comparable.yaml" > "$EVID/client-diff.txt"; rc=$?; set -e
+  echo "kubectl diff rc=$rc" | tee "$EVID/client-diff.rc"
+  [ "$rc" -le 1 ] || nogo "kubectl diff error rc=$rc"
+fi
+if [ "$DRY" = 1 ]; then kube get -f - --ignore-not-found -o json; else
+  kube get -f - --ignore-not-found -o json < "$WORK/comparable.yaml" > "$WORK/live.json" || nogo "live read"
+fi
+check classify --render "$WORK/render.yaml" --live "$WORK/live.json" "$EVID/classification.json" \
+  || nogo "diff classification (CRD spec change, missing CRD or spec change needing review)"
+
+say "4/6 ClusterSecretStore baseline (status conditions only)"
+if [ "$DRY" = 1 ]; then kube get clustersecretstores.external-secrets.io -o json; else
+  kube get clustersecretstores.external-secrets.io -o json > "$WORK/stores.json" || nogo "store read"
+fi
+check stores --live "$WORK/stores.json" "$EVID/store-baseline-before.json" || nogo "unexpected not-ready ClusterSecretStore"
+
+say "PRE-CHECK: GO"
+if [ "$PRE_ONLY" = 1 ]; then echo "--precheck-only: stopping before any change."; exit 0; fi
+
+say "5/6 create the Application (no automated sync, no finalizer) and confirm the sync"
+if [ "$DRY" = 1 ]; then kube -n argocd get application external-secrets --ignore-not-found -o name; EXISTS=""; else
+  EXISTS="$(kube -n argocd get application external-secrets --ignore-not-found -o name)"; fi
+if [ -z "$EXISTS" ]; then
+  if [ "$DRY" = 1 ]; then kube apply --dry-run=server -f -; kube apply -f -; else
+    kube apply --dry-run=server -f - < "$DRAFT" >/dev/null || nogo "Application server dry-run"
+    kube apply -f - < "$DRAFT"
+  fi
+fi
+if [ "$DRY" = 1 ]; then kube -n argocd get application external-secrets -o json; else
+  for _ in $(seq 1 30); do
+    kube -n argocd get application external-secrets -o json > "$WORK/app.json"
+    python3 -c "import json,sys;s=json.load(open(sys.argv[1])).get('status',{});sys.exit(0 if s.get('resources') else 1)" "$WORK/app.json" && break
+    sleep 5
+  done
+fi
+if [ "$DRY" = 1 ]; then TOTAL=39; else
+  TOTAL="$(python3 -c "import json,sys;print(json.load(open(sys.argv[1]))['total'])" "$EVID/render-split.json")"; fi
+check precheck-app --live "$WORK/app.json" --expected-resources "$TOTAL" "$EVID/argo-precheck.json" || nogo "Argo comparison"
+
+printf '\nAbout to sync external-secrets: revision 0.16.1, prune=false, force=false, ServerSideApply.\n'
+if [ "$DRY" = 1 ]; then echo "(dry-run: would ask for ADOPT)"; else
+  read -r -p "Type ADOPT to sync, anything else aborts: " answer
+  [ "$answer" = "ADOPT" ] || nogo "founder did not confirm. The Application exists but nothing was synced."
+fi
+kube -n argocd patch application external-secrets --type merge -p "$SYNC_JSON" -o name
+[ "$DRY" = 1 ] || date -u +%FT%TZ > "$EVID/sync-requested-at"
+
+say "6/6 wait for the operation, then post-checks"
+if [ "$DRY" = 1 ]; then kube -n argocd get application external-secrets -o json; else
+  for _ in $(seq 1 120); do
+    kube -n argocd get application external-secrets -o json > "$WORK/app.json"
+    python3 - "$WORK/app.json" <<'PY' && break
+import json, sys
+a = json.load(open(sys.argv[1])); s = a.get('status', {}); phase = s.get('operationState', {}).get('phase')
+done = phase in ('Failed', 'Error') or (phase == 'Succeeded' and s.get('health', {}).get('status') == 'Healthy')
+sys.exit(0 if done and not a.get('operation') else 1)
+PY
+    sleep 5
+  done
+fi
+if [ "$DRY" = 1 ]; then kube get crd -o name; kube get clustersecretstores.external-secrets.io -o json; else
+  kube get crd -o name > "$WORK/crds.txt"
+  kube get clustersecretstores.external-secrets.io -o json > "$WORK/stores-after.json"
+fi
+check stores --live "$WORK/stores-after.json" "$EVID/store-baseline-after.json" || true
+check postcheck --app "$WORK/app.json" --crds "$WORK/crds.txt" --before "$EVID/store-baseline-before.json" \
+  --after "$EVID/store-baseline-after.json" "$EVID/postcheck.json" || { rollback_help; exit 4; }
+step kube -n external-secrets get deploy
+
+say "ExternalSecret refresh sample (issue-core/issue-core-runtime)"
+if [ "$DRY" = 1 ]; then kube -n issue-core get externalsecret issue-core-runtime -o json; else
+  kube -n issue-core get externalsecret issue-core-runtime -o json > "$WORK/es-before.json"
+fi
+kube -n issue-core annotate externalsecret issue-core-runtime "force-sync=$STAMP" --overwrite -o name
+if [ "$DRY" = 1 ]; then kube -n issue-core get externalsecret issue-core-runtime -o json; else
+  for _ in $(seq 1 12); do
+    sleep 5
+    kube -n issue-core get externalsecret issue-core-runtime -o json > "$WORK/es-after.json"
+    "${CHECK[@]}" refresh --before "$WORK/es-before.json" --after "$WORK/es-after.json" >/dev/null && break
+  done
+fi
+check refresh --before "$WORK/es-before.json" --after "$WORK/es-after.json" "$EVID/refresh-sample.json" \
+  || { rollback_help; exit 4; }
+
+say "ADOPTED: external-secrets is Synced/Healthy at 0.16.1. Receipts: $EVID"
+echo "Next (agent): move the draft into argocd/railiance01/applications/, commit receipts, and update T06."
--- a/scripts/workload_adoption_proof.py
+++ b/scripts/workload_adoption_proof.py
@ -10,7 +10,7 @@ import re
 import subprocess
 from urllib.request import urlopen

-from argocd_adoption_proof import checked_run, identity, timestamp, tracking_settings
+from argocd_adoption_proof import _kube_route, checked_run, identity, timestamp, tracking_settings

 PIN = 'a3a8a27a8cda32ae3c7b55353ee16a131c50a0a0'
 CONTRACT = {
@ -224,7 +224,7 @@ def main():
                checked_run(['tar', '-x', '-C', temp], input=archive)
                rendered = checked_run(['kubectl', 'kustomize', str(Path(temp) / CONTRACT['source_path'])])
            common = Path(__file__).with_name('argocd_adoption_proof.py').read_bytes()
-            observation = json.loads(checked_run(['ssh', 'railiance01', 'python3', '-'], input=remote_program(common, rendered), timeout=180))
+            observation = json.loads(checked_run(_kube_route().ssh_argv(['python3', '-']), input=remote_program(common, rendered), timeout=180))
            with urlopen(CONTRACT['health_url'], timeout=30) as response:
                observation['public_health'] = {'url': response.geturl(), 'status': response.status}
            observation['metadata_captured_at'] = observation['captured_at']