Prove Target Revenue adoption and prepare bounded issue-core delivery

Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
2026-10-11 22:53:54 +02:00 · 2026-10-11 22:53:54 +02:00 · 1dc1d2f3a1
commit 1dc1d2f3a1
parent 4d2bb19f8a
26 changed files with 1989 additions and 21 deletions
--- a/scripts/activity_promotion_readiness.py
+++ b/scripts/activity_promotion_readiness.py
@ -0,0 +1,210 @@
+#!/usr/bin/env python3
+"""Read-only activity-core promotion gate; signed owner acceptance is mandatory.
+
+D2 grants scoped activation after proof, never substitutes for release-principal
+custody, authenticated issuers or rollback. This checker cannot activate a worker,
+issue credentials, change Argo policy or admit its own caller as a trusted signer.
+"""
+from __future__ import annotations
+import argparse
+import base64
+from datetime import datetime, timezone, timedelta
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+
+PIN = 'a86a14332d76d92a7ed176ea34dca4429902c396'
+OWNER_TASK = 'ACTIVITY-WP-0041-T03'
+AUTHORITY = 'ACTIVITY-WP-0041-image-only-v1'
+APPLICATION_PATH = 'argocd/railiance01/applications/activity-core.application.yaml'
+RESOURCES = {
+    ('', 'ConfigMap', 'activity-core', name) for name in
+    ('actcore-runtime-config', 'actcore-external-activity-definitions', 'actcore-report-schemas', 'actcore-ops-service-inventory')
+} | {('', 'Service', 'activity-core', name) for name in ('actcore-api', 'actcore-worker-metrics')} | {
+    ('apps', 'Deployment', 'activity-core', name) for name in ('actcore-api', 'actcore-worker', 'actcore-event-router')
+}
+CHECKS = {
+    'principal_scope_enforced', 'credential_custody_verified', 'revocation_denial_proven',
+    'independent_receipt_issuers_verified', 'authenticated_ci_digest_binding_proven',
+    'continuous_health_observer_active', 'live_and_rollback_retention_verified',
+    'image_only_admission_enforced', 'source_cas_and_single_flight_proven',
+    'restart_and_lost_response_recovery_proven', 'out_of_scope_denials_proven',
+    'forged_stale_mismatched_receipt_denials_proven', 'failed_health_rollback_proven',
+    'temporal_dispatch_registered', 'durable_audit_sink_verified', 'report_schedule_invariants_verified',
+}
+TARGETS = {
+    'platform_repository': 'coulomb/railiance-platform', 'branch': 'main',
+    'child_path': APPLICATION_PATH, 'child_field': '/spec/source/targetRevision',
+    'root_application': 'railiance-apps-root', 'root_sync_selects_only': 'activity-core',
+    'child_application': 'activity-core', 'prune': False,
+}
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()
+
+
+def stamp(value):
+    result = datetime.fromisoformat(value.replace('Z', '+00:00'))
+    if result.tzinfo is None:
+        raise ValueError('timezone required')
+    return result
+
+
+def run(argv, *, input=None):
+    try:
+        result = subprocess.run(argv, input=input, capture_output=True, timeout=100)
+        if result.returncode:
+            raise ValueError('metadata read unverified; protected output suppressed')
+        return result.stdout
+    except (OSError, subprocess.SubprocessError):
+        raise ValueError('metadata read unverified; protected output suppressed') from None
+
+
+def collect():
+    def get(args):
+        return json.loads(run(['kubectl', *args, '-o', 'json']))
+    nodes = get(['get', 'nodes'])
+    if '92.205.62.239' not in [a['address'] for n in nodes['items'] for a in n['status']['addresses']]:
+        raise ValueError('estate identity mismatch')
+    namespace = get(['get', 'namespace', 'kube-system'])
+    obj = get(['-n', 'argocd', 'get', 'application', 'activity-core'])
+    spec, status = obj['spec'], obj.get('status', {})
+    expected_source = {'repoURL': 'https://forgejo.coulomb.social/coulomb/activity-core.git',
+                       'path': 'k8s/gitops', 'targetRevision': PIN}
+    if spec.get('source') != expected_source:
+        raise ValueError('exact activity source unverified')
+    op = status.get('operationState', {})
+    deployments = []
+    for name in ('actcore-api', 'actcore-worker', 'actcore-event-router'):
+        deployment = get(['-n', 'activity-core', 'get', 'deployment', name])
+        generation = deployment['metadata']['generation']; ds = deployment.get('status', {})
+        desired = deployment['spec'].get('replicas', 1)
+        deployments.append({'name': name, 'generation_observed': ds.get('observedGeneration') == generation,
+                            'replicas': desired, 'ready': ds.get('readyReplicas', 0),
+                            'updated': ds.get('updatedReplicas', 0), 'available': ds.get('availableReplicas', 0)})
+    project = get(['-n', 'argocd', 'get', 'appproject', 'activity-core'])['spec']
+    return {'captured_at': datetime.now(timezone.utc).isoformat(), 'estate_ip': '92.205.62.239',
+            'cluster_uid': namespace['metadata']['uid'], 'application_uid': obj['metadata']['uid'],
+            'source': expected_source, 'project': spec.get('project'),
+            'destination': spec.get('destination'), 'automated_present': 'automated' in spec.get('syncPolicy', {}),
+            'sync_options': spec.get('syncPolicy', {}).get('syncOptions', []),
+            'pending_operation': bool(obj.get('operation')), 'sync': status.get('sync', {}).get('status'),
+            'revision': status.get('sync', {}).get('revision'), 'health': status.get('health', {}).get('status'),
+            'operation': {k: op.get(k) for k in ('phase', 'finishedAt')},
+            'condition_types': [c.get('type') for c in status.get('conditions', [])],
+            'resources': [{k: r.get(k) for k in ('group', 'kind', 'namespace', 'name', 'status')} for r in status.get('resources', [])],
+            'project_scope_matches': (project.get('sourceRepos') == ['https://forgejo.coulomb.social/coulomb/activity-core.git']
+                and project.get('destinations') == [{'server': 'https://kubernetes.default.svc', 'namespace': 'activity-core'}]
+                and project.get('clusterResourceWhitelist', []) == []
+                and {(r.get('group') or '',r['kind']) for r in project.get('namespaceResourceWhitelist', [])}
+                    == {('', 'ConfigMap'),('', 'Service'),('apps','Deployment')}),
+            'deployments': deployments}
+
+
+def adoption_checks(observation, now):
+    if (not 0 <= (now-stamp(observation['captured_at'])).total_seconds() <= 300
+            or observation['estate_ip'] != '92.205.62.239' or not observation['cluster_uid']
+            or not observation['application_uid']):
+        raise ValueError('fresh exact estate observation required')
+    if (observation['source'] != {'repoURL':'https://forgejo.coulomb.social/coulomb/activity-core.git','path':'k8s/gitops','targetRevision':PIN}
+            or observation['project'] != 'activity-core' or observation['revision'] != PIN
+            or observation['destination'] != {'server':'https://kubernetes.default.svc','namespace':'activity-core'}
+            or observation['automated_present'] is not False or observation['pending_operation'] is not False
+            or set(observation['sync_options']) != {'CreateNamespace=false','ApplyOutOfSyncOnly=true','PruneLast=true','FailOnSharedResource=true'}
+            or observation['sync'] != 'Synced' or observation['health'] != 'Healthy'
+            or observation['condition_types'] or observation['project_scope_matches'] is not True):
+        raise ValueError('bounded manual adoption scope unverified')
+    op = observation['operation']
+    if op['phase'] != 'Succeeded' or (stamp(observation['captured_at'])-stamp(op['finishedAt'])).total_seconds() < 86400:
+        raise ValueError('24-hour adoption eligibility unverified')
+    resources = observation['resources']
+    if (len(resources) != 9 or {tuple(r.get(k) or '' for k in ('group','kind','namespace','name')) for r in resources} != RESOURCES
+            or any(r['status'] != 'Synced' for r in resources)):
+        raise ValueError('nine-resource runtime inventory unverified')
+    deployments = observation['deployments']
+    if (len(deployments) != 3 or {d['name'] for d in deployments} != {'actcore-api','actcore-worker','actcore-event-router'}
+            or any(d['generation_observed'] is not True or type(d['replicas']) is not int or d['replicas'] < 1
+                   or any(d[k] != d['replicas'] for k in ('ready','updated','available')) for d in deployments)):
+        raise ValueError('deployment generation/readiness unverified')
+
+
+def owner_acceptance(envelope, trusted_key, trusted_principal, observation, now):
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+    if set(envelope) != {'payload','signature'} or not trusted_principal or len(trusted_key) != 32:
+        raise ValueError('trusted owner acceptance configuration required')
+    payload = envelope['payload']
+    Ed25519PublicKey.from_public_bytes(trusted_key).verify(base64.b64decode(envelope['signature'],validate=True), canonical(payload))
+    if (payload.get('schema_version') != 'activity-core-release-acceptance/v1'
+            or payload.get('owner_task') != OWNER_TASK or payload.get('issuer') != trusted_principal
+            or payload.get('authority') != AUTHORITY or payload.get('application') != 'activity-core'
+            or payload.get('cluster_uid') != observation['cluster_uid'] or payload.get('adoption_revision') != PIN
+            or payload.get('targets') != TARGETS or not payload.get('release_principal')
+            or not re.fullmatch('[a-f0-9]{40}',payload.get('broker_revision',''))
+            or set(payload.get('checks',{})) != CHECKS
+            or any(value is not True for value in payload['checks'].values())):
+        raise ValueError('owner acceptance scope or machine proof incomplete')
+    issued, expires = stamp(payload['issued_at']), stamp(payload['expires_at'])
+    if not 0 <= (now-issued).total_seconds() <= 300 or not now < expires <= issued + timedelta(minutes=5):
+        raise ValueError('owner acceptance stale or expired')
+    health = payload['continuous_health']
+    if (health.get('revision') != PIN or health.get('continuous') is not True
+            or type(health.get('max_gap_seconds')) is not int or not 0 < health['max_gap_seconds'] <= 90
+            or not 0 <= (now-stamp(health['observed_at'])).total_seconds() <= 90
+            or (stamp(health['observed_at'])-stamp(health['healthy_since'])).total_seconds() < 86400):
+        raise ValueError('owner continuous observation proof incomplete')
+    return payload
+
+
+def readiness(observation, *, envelope=None, trusted_key=None, trusted_principal=None, now=None):
+    now = now or datetime.now(timezone.utc)
+    base = {'schema_version':'activity-promotion-readiness/v1','task':'RPF-WP-0048-T02',
+            'owner_dependency':OWNER_TASK,'native_mutation':False,'automatic_sync_candidate':False,
+            'observed_at':now.isoformat(),'checker_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}
+    try:
+        adoption_checks(observation,now)
+    except Exception:
+        return base | {'status':'unverified','reason':'adoption_metadata_unverified'}
+    base |= {'adoption_24h_eligible':True,'adoption_snapshot_is_not_continuous_observation':True,
+             'observation_sha256':hashlib.sha256(canonical(observation)).hexdigest()}
+    if envelope is None or trusted_key is None or trusted_principal is None:
+        return base | {'status':'waiting','reason':'authenticated_owner_activation_acceptance_missing',
+                       'depends_on':[OWNER_TASK],'needs_human':False}
+    try:
+        payload = owner_acceptance(envelope,trusted_key,trusted_principal,observation,now)
+    except Exception:
+        return base | {'status':'unverified','reason':'owner_activation_acceptance_unverified'}
+    return base | {'status':'ready','owner_acceptance_sha256':hashlib.sha256(canonical(envelope)).hexdigest(),
+                   'release_principal':payload['release_principal'],'broker_revision':payload['broker_revision'],
+                   'scope':AUTHORITY,'rollback':'disable dedicated release dispatcher admission; retain manual child policy and prior source pin',
+                   'abort_owner':'railiance-platform'}
+
+
+def main():
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--live',action='store_true');parser.add_argument('--remote',action='store_true',help=argparse.SUPPRESS)
+    parser.add_argument('--observation',type=Path);parser.add_argument('--receipt',type=Path)
+    parser.add_argument('--owner-acceptance',type=Path)
+    parser.add_argument('--trusted-owner-key',type=Path,help='operator-owned raw Ed25519 public key; never receipt-provided')
+    parser.add_argument('--trusted-owner-principal',help='operator-owned issuer binding')
+    args=parser.parse_args()
+    try:
+        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()))
+            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')
+        else:observation=json.loads(args.observation.read_text())
+        envelope=json.loads(args.owner_acceptance.read_text()) if args.owner_acceptance else None
+        key=args.trusted_owner_key.read_bytes() if args.trusted_owner_key else None
+        result=readiness(observation,envelope=envelope,trusted_key=key,trusted_principal=args.trusted_owner_principal)
+    except Exception:result={'schema_version':'activity-promotion-readiness/v1','status':'unverified','reason':'protected_read_or_inputs_unverified','native_mutation':False}
+    if args.receipt:args.receipt.write_text(json.dumps(result,indent=2)+'\n')
+    print(json.dumps(result,sort_keys=True));return 0 if result['status']=='ready' else 1
+
+
+if __name__=='__main__':raise SystemExit(main())
--- a/scripts/argocd_public_transport.py
+++ b/scripts/argocd_public_transport.py
@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Guarded public-only Argo transport prerequisite; no Secret or credential calls."""
+import argparse
+import copy
+from datetime import datetime,timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shlex
+import subprocess
+
+URL='ssh://git@forgejo.coulomb.social:30022/coulomb/rapp-issue-core.git'
+D2='89f1058f-33bc-47a8-8a49-0b44d426c67e'
+TARGETS=[('configmap','argocd-ssh-known-hosts-cm','/data/ssh_known_hosts','railiance-enablement'),('appproject','railiance-tenants','/spec/sourceRepos','railiance-platform')]
+
+class Refused(Exception):pass
+
+def require(ok,reason):
+    if not ok:raise Refused(reason)
+
+def digest(value):return hashlib.sha256(json.dumps(value,sort_keys=True).encode()).hexdigest()
+
+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)
+    except Exception:raise Refused('command_outcome_unverified') from None
+    require(result.returncode==0,'command_outcome_unverified')
+    return result.stdout
+
+def value(obj,path):
+    for key in path.strip('/').split('/'):obj=obj[key]
+    return obj
+
+def replace(obj,path,wanted):
+    keys=path.strip('/').split('/');parent=obj
+    for key in keys[:-1]:parent=parent[key]
+    parent[keys[-1]]=wanted
+
+def normalized(obj):
+    obj=copy.deepcopy(obj)
+    for field in ('managedFields','resourceVersion','generation'):obj.get('metadata',{}).pop(field,None)
+    return obj
+
+def validate(plan):
+    require(plan.get('schema')=='argocd-public-transport/v1' and plan.get('task')=='RPF-WP-0044-T04' and plan.get('approval_ref')==D2,'plan_authority')
+    require(plan.get('target')=='railiance01' and plan.get('namespace')=='argocd' and plan.get('repo_server_tcp_30022_verified') is True,'plan_target')
+    require(all(isinstance(plan.get(k),str) and plan[k] for k in ('diff_ref','rollback','abort_owner','baseline_sha256','public_authority_observation_sha256')),'plan_receipt_contract')
+    require(len(plan.get('resources',[]))==2,'target_count')
+    for item,fixed in zip(plan['resources'],TARGETS):
+        require(tuple(item.get(k) for k in ('kind','name','path','owner'))==fixed,'fixed_target')
+        require(bool(re.fullmatch(r'[a-f0-9-]{36}',item.get('uid',''))),'target_uid')
+    cm,project=plan['resources']
+    require(isinstance(cm['before'],str) and cm['before'].endswith('\n') and isinstance(cm['after'],str),'host_field_type')
+    added=cm['after'][len(cm['before']):]
+    require(cm['after'].startswith(cm['before']) and bool(re.fullmatch(r'\[forgejo\.coulomb\.social\]:30022 (ssh-rsa|ssh-ed25519|ecdsa-sha2-nistp256) [A-Za-z0-9+/=]+\n',added)),'host_addition_scope')
+    require('[forgejo.coulomb.social]:30022 ' not in cm['before'],'host_preimage_already_admitted')
+    require(isinstance(project['before'],list) and all(isinstance(v,str) for v in project['before']) and URL not in project['before'] and project['after']==project['before']+[URL],'repository_addition_scope')
+
+def patch(item,live,rollback=False):
+    before,after=(item['after'],item['before']) if rollback else (item['before'],item['after'])
+    require(live['metadata']['uid']==item['uid'],'target_uid_changed')
+    require(value(live,item['path'])==before,'target_preimage_changed')
+    require(isinstance(live['metadata'].get('resourceVersion'),str),'resource_version_required')
+    return [{'op':'test','path':'/metadata/uid','value':item['uid']},{'op':'test','path':'/metadata/resourceVersion','value':live['metadata']['resourceVersion']},{'op':'test','path':item['path'],'value':before},{'op':'replace','path':item['path'],'value':after}]
+
+def perform(plan,action,receipt,checkpoint=lambda:None):
+    validate(plan)
+    nodes=json.loads(run(['get','nodes','-o','json']))['items']
+    require(len(nodes)==1 and any(a.get('address')=='92.205.62.239' for a in nodes[0]['status']['addresses']),'wrong_estate')
+    read=lambda item:json.loads(run(['-n','argocd','get',item['kind'],item['name'],'-o','json']))
+    rollback=action=='rollback'
+    # Validate both targets before the first write, avoiding an avoidable partial change.
+    live=[read(item) for item in plan['resources']]
+    for item,obj in zip(plan['resources'],live):
+        require(obj['metadata']['uid']==item['uid'],'target_uid_changed')
+        if action=='verify':require(value(obj,item['path'])==item['after'],'verification_field_changed')
+        elif rollback and value(obj,item['path'])==item['before']:continue
+        else:patch(item,obj,rollback)
+    for item,obj in zip(plan['resources'],live):
+        record={'kind':item['kind'],'name':item['name'],'uid':item['uid'],'field':item['path'],'before_sha256':digest(item['before']),'after_sha256':digest(item['after'])};receipt['resources'].append(record)
+        if action=='verify':record['verified']=True;continue
+        if rollback and value(obj,item['path'])==item['before']:record['already_restored']=True;continue
+        body=json.dumps(patch(item,obj,rollback));args=['-n','argocd','patch',item['kind'],item['name'],'--type=json','--patch-file=/dev/stdin']
+        admitted=json.loads(run(args+['--dry-run=server','-o','json'],body))
+        wanted=copy.deepcopy(obj);replace(wanted,item['path'],item['before'] if rollback else item['after'])
+        require(normalized(admitted)==normalized(wanted),'dry_run_modified_other_fields');record['server_dry_run']=True
+        if action=='dry-run':continue
+        receipt['pending_target']=item['kind']+'/'+item['name'];receipt['mutation_attempted']=True;checkpoint()
+        run(args+['-o','name'],body)
+        checked=read(item)
+        require(checked['metadata']['uid']==item['uid'] and normalized(checked)==normalized(wanted),'postchange_projection_unverified')
+        record['changed_verified']=True;receipt['pending_target']=None;checkpoint()
+    receipt['passed']=True;return receipt
+
+def main():
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--plan',type=Path,required=True);parser.add_argument('--plan-sha256',required=True)
+    parser.add_argument('--action',choices=('dry-run','execute','verify','rollback'),default='dry-run')
+    parser.add_argument('--receipt',type=Path,required=True)
+    args=parser.parse_args()
+    receipt={'schema':'argocd-public-transport-receipt/v1','task':'RPF-WP-0044-T04','approval_ref':D2,'action':args.action,'passed':False,'mutation_attempted':False,'resources':[],'secret_or_credential_calls':False,'captured_at':datetime.now(timezone.utc).isoformat()}
+    # Exclusive receipt is a replay guard. Never overwrite another attempt.
+    fd=None
+    try:
+        require(not args.receipt.exists(),'receipt_exists_do_not_replay')
+        fd=os.open(args.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
+        raw=args.plan.read_bytes();require(hashlib.sha256(raw).hexdigest()==args.plan_sha256,'plan_changed')
+        def checkpoint():
+            os.lseek(fd,0,os.SEEK_SET);os.ftruncate(fd,0);os.write(fd,(json.dumps(receipt,sort_keys=True)+'\n').encode());os.fsync(fd)
+        receipt['plan_sha256']=args.plan_sha256;checkpoint();perform(json.loads(raw),args.action,receipt,checkpoint)
+    except Refused as error:receipt['failure']=str(error)
+    except Exception:receipt['failure']='internal_outcome_unverified'
+    finally:
+        if fd is not None:
+            os.lseek(fd,0,os.SEEK_SET);os.ftruncate(fd,0);os.write(fd,(json.dumps(receipt,sort_keys=True)+'\n').encode());os.fsync(fd);os.close(fd)
+    # Only whitelist receipt fields are emitted. Every raw child error stays captured.
+    print(json.dumps(receipt,sort_keys=True));return 0 if receipt['passed'] else 1
+
+if __name__=='__main__':raise SystemExit(main())
--- a/scripts/provision_argocd_issue_core_repository.py
+++ b/scripts/provision_argocd_issue_core_repository.py
@ -0,0 +1,198 @@
+#!/usr/bin/env python3
+"""Silent fixed-scope D2 repository deploy-key issuer/destination actuator.
+
+Run through the existing Platform attended login and Forgejo admin --exec lane.
+No lane grants itself authority. The reviewed contract seals transport/host trust
+and existing egress; the command performs no RBAC, policy or network expansion.
+No implicit recovery: an existing journal, ambiguous result or drift stops replay.
+"""
+import argparse
+import base64
+from datetime import datetime, timezone
+import hashlib
+import fnmatch
+import json
+import os
+import re
+from pathlib import Path
+import stat
+import shlex
+import subprocess
+import sys
+import tempfile
+import urllib.error
+import urllib.request
+
+REPO='coulomb/rapp-issue-core'
+BAO='platform/operators/argocd/repositories/rapp-issue-core'
+URL='ssh://git@forgejo.coulomb.social:30022/coulomb/rapp-issue-core.git'
+SECRET='rapp-issue-core-repository'
+TITLE='argocd-rapp-issue-core-readonly-D2'
+D2='89f1058f-33bc-47a8-8a49-0b44d426c67e'
+FIXED={'work_record':'RPF-WP-0044-T04','scope_approval_ref':D2,'repository':REPO,'read_only':True,'openbao_path':BAO,'destination':'argocd/'+SECRET,'project':'railiance-tenants','url':URL,'wrap_ttl_seconds':60,'new_network_scope':False,'execution_mode':'attended-founder-bounded'}
+
+
+class Refused(Exception):pass
+
+
+def require(ok,reason):
+    if not ok:raise Refused(reason)
+
+
+def captured(argv,body=None,env=None):
+    try:
+        result=subprocess.run(argv,input=body,env=env,capture_output=True,text=True,timeout=60)
+    except Exception:
+        raise Refused('command_unverified') from None
+    return result
+
+
+def run(argv,body=None,env=None):
+    result=captured(argv,body,env)
+    require(result.returncode==0,'command_unverified')
+    return result.stdout
+
+
+def metadata_children():
+    result=captured(['bao','list','-format=json','platform/metadata/operators/argocd/repositories/'])
+    # Installed CLI represents an empty/404 LIST as exit 2, JSON {}, no stderr.
+    if result.returncode==2 and result.stdout.strip()=='{}' and not result.stderr.strip():return []
+    require(result.returncode==0,'custody_metadata_unverified')
+    children=json.loads(result.stdout);require(isinstance(children,list),'custody_metadata_unverified');return children
+
+
+def replay_denied(env):
+    result=captured(['bao','unwrap','-format=json'],env=env)
+    text=result.stderr or ''
+    return result.returncode==2 and not result.stdout.strip() and bool(re.search(r'Code:\s*400\b',text)) and bool(re.search(r'^\* wrapping token is not valid or does not exist\s*$',text,re.M))
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self,*args,**kwargs):raise Refused('redirect_refused')
+
+
+def api(token,method,path,payload=None,expected=(200,)):
+    require(path.startswith('/repos/'+REPO+'/') or path=='/repos/'+REPO,'provider_path_refused')
+    request=urllib.request.Request('https://forgejo.coulomb.social/api/v1'+path,method=method,
+      data=None if payload is None else json.dumps(payload).encode(),
+      headers={'Authorization':'token '+token,'Content-Type':'application/json'})
+    try:
+        with urllib.request.build_opener(NoRedirect()).open(request,timeout=30) as response:
+            require(response.status in expected,'provider_status_refused')
+            raw=response.read(1024*1024+1);require(len(raw)<=1024*1024,'provider_body_bound')
+            return json.loads(raw) if raw else None
+    except Refused:raise
+    except Exception:raise Refused('provider_outcome_unverified') from None
+
+
+def validate_contract(contract):
+    require(isinstance(contract,dict),'contract_type')
+    for key,value in FIXED.items():
+        require(type(contract.get(key)) is type(value) and contract[key]==value,'contract_scope')
+    for key in ('owner_review_ref','ssh_egress_existing_proof_ref','runtime_identity_ref','rollback_ref'):
+        require(isinstance(contract.get(key),str) and contract[key].strip(),'missing_owner_proof')
+    require(isinstance(contract.get('ssh_known_hosts_sha256'),str) and len(contract['ssh_known_hosts_sha256'])==64,'host_pin_required')
+    require(datetime.now(timezone.utc)<datetime.fromisoformat(contract['expires_at']),'expired_contract')
+
+
+def secret_payload(private_key):
+    return {'apiVersion':'v1','kind':'Secret','metadata':{'name':SECRET,'namespace':'argocd','labels':{'argocd.argoproj.io/secret-type':'repository','app.kubernetes.io/part-of':'railiance-gitops'},'annotations':{'railiance.io/credential-task':'RPF-WP-0044-T04'}},'type':'Opaque','stringData':{'type':'git','project':'railiance-tenants','url':URL,'sshPrivateKey':private_key}}
+
+
+def execute(contract,journal):
+    validate_contract(contract)
+    require(os.environ.get('API_TOKEN'),'provider_identity_missing')
+    # Fixed native target over SSH; no workstation kube-context fallback.
+    # Private destination material travels only in encrypted stdin, never argv.
+    def kubectl(*args,body=None):return run(['ssh','-o','BatchMode=yes','-o','ConnectTimeout=10','railiance01',shlex.join(['kubectl','--request-timeout=15s',*args])],body)
+    nodes=json.loads(kubectl('get','nodes','-o','json'))['items']
+    require(len(nodes)==1 and any(a.get('address')=='92.205.62.239' for a in nodes[0]['status']['addresses']),'wrong_estate')
+    known=json.loads(kubectl('-n','argocd','get','configmap','argocd-ssh-known-hosts-cm','-o','json'))['data']['ssh_known_hosts']
+    require(hashlib.sha256(known.encode()).hexdigest()==contract['ssh_known_hosts_sha256'],'host_pin_changed')
+    require('[forgejo.coulomb.social]:30022 ' in known,'host_pin_missing')
+    project=json.loads(kubectl('-n','argocd','get','appproject','railiance-tenants','-o','json'))
+    require(any(fnmatch.fnmatchcase(URL,p) for p in project.get('spec',{}).get('sourceRepos',[])),'project_source_refused')
+    caps=run(['bao','token','capabilities','platform/data/operators/argocd/repositories/rapp-issue-core'])
+    capability_set={x.strip() for x in caps.strip().split(',')}
+    capability_mode='scoped' if {'create','update','read'}<=capability_set else 'attended-root' if capability_set=={'root'} and contract['execution_mode']=='attended-founder-bounded' else None
+    require(capability_mode,'operatorpath_capability_missing')
+    audit_devices=json.loads(run(['bao','audit','list','-format=json']))
+    require(isinstance(audit_devices,dict) and 0<len(audit_devices)<=32,'enabled_audit_metadata_required')
+    metadata=metadata_children()
+    require('rapp-issue-core' not in metadata and 'rapp-issue-core/' not in metadata,'existing_custody_refused')
+    # Metadata-presence read only: no Secret object or value is fetched.
+    existing=kubectl('-n','argocd','get','secret',SECRET,'--ignore-not-found','-o','name')
+    require(not existing.strip(),'existing_destination_refused')
+    repo=api(os.environ['API_TOKEN'],'GET','/repos/'+REPO)
+    require(repo.get('full_name')==REPO and repo.get('private') is True,'repository_identity')
+    keys=api(os.environ['API_TOKEN'],'GET','/repos/'+REPO+'/keys?limit=100')
+    require(isinstance(keys,list) and len(keys)<100,'deploy_key_list_bound')
+    require(not any(k.get('title')==TITLE for k in keys),'existing_provider_key_refused')
+    receipt={'schema':'argocd-issue-core-credential/v1','task':'RPF-WP-0044-T04','approval_ref':D2,'contract_sha256':hashlib.sha256(json.dumps(contract,sort_keys=True).encode()).hexdigest(),'state':'prepared','secret_values_observed_by_agent':False,'provider_key_id':None,'destination': 'argocd/'+SECRET,'wrap_ttl_seconds':60,'private_key_rotation_max_days':180,'execution_mode':'attended-founder-bounded','operatorpath_capability_mode':capability_mode,'audit_device_count':len(audit_devices),'audit_enabled_metadata':True}
+    fd=os.open(journal,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
+    def save(state):
+        receipt['state']=state;receipt['updated_at']=datetime.now(timezone.utc).isoformat()
+        os.lseek(fd,0,os.SEEK_SET);os.ftruncate(fd,0);os.write(fd,(json.dumps(receipt,sort_keys=True)+'\n').encode());os.fsync(fd)
+    try:
+        save('prepared')
+        # Issuer uses private tmpfs; destination is the fixed native railiance01 API.
+        with tempfile.TemporaryDirectory(prefix='argocd-issue-core-',dir='/dev/shm') as folder:
+            private=Path(folder)/'key'
+            run(['ssh-keygen','-q','-t','ed25519','-N','','-C',TITLE,'-f',str(private)])
+            require(stat.S_IMODE(private.stat().st_mode)==0o600,'private_file_mode')
+            public=Path(str(private)+'.pub').read_text().strip();key=private.read_text()
+            receipt['public_key_sha256']=hashlib.sha256(public.encode()).hexdigest()
+            save('provider_request_pending')
+            created=api(os.environ['API_TOKEN'],'POST','/repos/'+REPO+'/keys',{'title':TITLE,'key':public,'read_only':True},expected=(201,))
+            require(type(created.get('id')) is int and created.get('read_only') is True and created.get('key','').split()[:2]==public.split()[:2],'provider_key_unverified')
+            receipt['provider_key_id']=created['id'];save('provider_created')
+            saved=run(['bao','kv','put','-format=json','-cas=0',BAO,'SSH_PRIVATE_KEY=-','SSH_PUBLIC_KEY='+public,'PROVIDER_KEY_ID='+str(created['id']),'REPOSITORY='+REPO],key)
+            version=json.loads(saved).get('data',{}).get('version');require(type(version) is int,'custody_version_unverified')
+            receipt['custody_version']=version;receipt['custody_write_request_id']=json.loads(saved).get('request_id');save('custody_stored')
+            wrap=json.loads(run(['bao','kv','get','-wrap-ttl=60s','-format=json',BAO]))['wrap_info']
+            require(wrap.get('ttl')==60 and isinstance(wrap.get('token'),str) and wrap.get('creation_path')=='platform/data/operators/argocd/repositories/rapp-issue-core','wrap_unverified')
+            receipt['wrap_creation_path']=wrap['creation_path']
+            receipt['wrap_accessor']=wrap.get('accessor');save('unwrap_request_pending')
+            # Token travels only in this fixed child environment, never argv/stdout.
+            wrap_env={**os.environ,'BAO_TOKEN':wrap['token'],'VAULT_TOKEN':wrap['token']}
+            unwrapped=json.loads(run(['bao','unwrap','-format=json'],env=wrap_env))
+            receipt['unwrap_request_id']=unwrapped.get('request_id');delivered=unwrapped['data']['data'];require(delivered['SSH_PRIVATE_KEY']==key and str(delivered['PROVIDER_KEY_ID'])==str(created['id']),'unwrap_content_unverified');save('unwrap_consumed')
+            require(replay_denied(wrap_env),'consumed_wrap_replay_not_proven')
+            receipt['consumed_wrap_replay_denied_400']=True;save('consumed_replay_denied')
+            manifest=json.dumps(secret_payload(delivered['SSH_PRIVATE_KEY']))
+            kubectl('create','--dry-run=server','-f','-','-o','name',body=manifest)
+            save('destination_request_pending')
+            kubectl('create','--field-manager=argocd-issue-core-credential','-f','-','-o','name',body=manifest)
+            # No client-side apply annotation; only one exact Secret is created.
+            require(kubectl('-n','argocd','get','secret',SECRET,'-o','name').strip()=='secret/'+SECRET,'destination_unverified')
+            save('delivered_not_adopted')
+    except BaseException:
+        save('uncertain_do_not_replay');raise
+    finally:os.close(fd)
+
+
+def main():
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--inner',action='store_true');parser.add_argument('--contract',required=True,type=Path);parser.add_argument('--contract-sha256',required=True);parser.add_argument('--journal',required=True,type=Path)
+    args=parser.parse_args()
+    # Suppress the inherited fetch stderr and all child output, including tracebacks.
+    sink=os.open(os.devnull,os.O_WRONLY);os.dup2(sink,1);os.dup2(sink,2);os.close(sink)
+    try:
+        raw=args.contract.read_bytes();require(hashlib.sha256(raw).hexdigest()==args.contract_sha256,'contract_changed')
+        validate_contract(json.loads(raw));require(not args.journal.exists(),'journal_exists_do_not_replay')
+        if not args.inner:
+            source=Path(__file__).resolve()
+            result=subprocess.run(['warden','access','forgejo-admin-api-token','--field','API_TOKEN','--exec','--',sys.executable,str(source),'--inner','--contract',str(args.contract),'--contract-sha256',args.contract_sha256,'--journal',str(args.journal)],capture_output=True,timeout=900)
+            return 0 if result.returncode==0 else 1
+        execute(json.loads(raw),args.journal);return 0
+    except BaseException as error:
+        # Fixed diagnostic only; never overwrite a prior/uncertain execution journal.
+        failure=str(error) if isinstance(error,Refused) else 'internal_outcome_unverified'
+        try:
+            fd=os.open(args.journal,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
+            os.write(fd,(json.dumps({'schema':'argocd-issue-core-credential/v1','state':'preflight_refused','failure':failure,'secret_values_observed_by_agent':False})+'\n').encode());os.fsync(fd);os.close(fd)
+        except Exception:pass
+        return 1
+
+
+if __name__=='__main__':raise SystemExit(main())
--- a/scripts/workload_adoption_proof.py
+++ b/scripts/workload_adoption_proof.py
@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+"""Read-only manual workload adoption proof; never enables automation or runs hooks."""
+from datetime import datetime, timezone
+import argparse
+import copy
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+from urllib.request import urlopen
+
+from argocd_adoption_proof import checked_run, identity, timestamp, tracking_settings
+
+PIN = 'a3a8a27a8cda32ae3c7b55353ee16a131c50a0a0'
+CONTRACT = {
+    'estate_ip': '92.205.62.239', 'application': 'target-revenue',
+    'project': 'railiance-tenants', 'destination': {'server': 'https://kubernetes.default.svc', 'namespace': 'target-revenue'},
+    'repo_url': 'https://forgejo.coulomb.social/coulomb/target-revenue.git', 'source_path': 'k8s/railiance',
+    'revision': PIN, 'tracking_method': 'annotation', 'controller_version': 'v3.5.3',
+    'minimum_proving_seconds': 86400, 'maximum_observation_age_seconds': 300,
+    'health_url': 'https://revenue.coulomb.social/healthz',
+    'resources': [{'group': group, 'kind': kind, 'namespace': 'target-revenue', 'name': name}
+                  for group, kind, name in [('', 'Service', 'target-revenue'), ('apps', 'Deployment', 'target-revenue'),
+                    ('external-secrets.io', 'ExternalSecret', 'target-revenue-runtime'),
+                    ('networking.k8s.io', 'Ingress', 'target-revenue'), ('postgresql.cnpg.io', 'Cluster', 'target-revenue-pg')]],
+    'hooks': ['target-revenue-bootstrap-binky', 'target-revenue-migrate'],
+    'ready_checks': [{'kind': kind, 'namespace': 'target-revenue', 'name': name, 'condition': condition}
+                     for kind, name, condition in [('Deployment', 'target-revenue', 'Available'),
+                     ('ExternalSecret', 'target-revenue-runtime', 'Ready'), ('Cluster', 'target-revenue-pg', 'Ready')]],
+}
+
+
+def validate_contract(contract):
+    if (not re.fullmatch(r'[a-f0-9]{40}', contract['revision']) or contract['minimum_proving_seconds'] < 86400
+            or not 0 < contract['maximum_observation_age_seconds'] <= 300):
+        raise ValueError('invalid proof bounds')
+    if any(r['kind'] not in {'Service', 'Deployment', 'ExternalSecret', 'Ingress', 'Cluster'} for r in contract['resources']):
+        raise ValueError('resource kinds not allowlisted')
+    if len({identity(r) for r in contract['resources']}) != len(contract['resources']):
+        raise ValueError('duplicate resource identities')
+    for r in [*contract['resources'], *contract['ready_checks']]:
+        if r['kind'] not in {'Service', 'Deployment', 'ExternalSecret', 'Ingress', 'Cluster'}:
+            raise ValueError('readiness kind not allowlisted')
+        if any(not re.fullmatch(r'[a-z0-9][a-z0-9.-]*', r[k]) for k in ('name', 'namespace')):
+            raise ValueError('invalid resource identity')
+
+
+def normalize(items, contract, settings):
+    validate_contract(contract)
+    if (settings['method'] != 'annotation' or settings['controller_version'] != contract['controller_version']
+            or settings['installation_id_absent'] is not True):
+        raise ValueError('tracking semantics unverified')
+    resources, hooks = [], []
+    for raw in items:
+        obj = copy.deepcopy(raw)
+        meta = obj['metadata']
+        annotation = meta.get('annotations', {})
+        if 'argocd.argoproj.io/hook' in annotation:
+            if obj['kind'] != 'Job' or annotation['argocd.argoproj.io/hook'] != 'Sync':
+                raise ValueError('unsupported hook')
+            hooks.append(meta['name'])
+            continue
+        group = obj['apiVersion'].split('/')[0] if '/' in obj['apiVersion'] else ''
+        ident = (group, obj['kind'], meta.get('namespace', ''), meta['name'])
+        if ident not in {identity(r) for r in contract['resources']}:
+            raise ValueError('render resource not admitted')
+        tracking = f"{contract['application']}:{group}/{obj['kind']}:{meta['namespace']}/{meta['name']}"
+        annotations = meta.setdefault('annotations', {})
+        if annotations.get('argocd.argoproj.io/tracking-id', tracking) != tracking:
+            raise ValueError('conflicting source tracking annotation')
+        annotations['argocd.argoproj.io/tracking-id'] = tracking
+        resources.append(obj)
+    if (len(resources) != len(contract['resources']) or len({(r['apiVersion'], r['kind'], r['metadata']['name']) for r in resources}) != len(resources)
+            or sorted(hooks) != sorted(contract['hooks'])):
+        raise ValueError('render resource or hook set differs')
+    return resources
+
+
+def decode_objects(raw):
+    text = raw.decode()
+    decoder = json.JSONDecoder()
+    items = []
+    while text.strip():
+        value, end = decoder.raw_decode(text.lstrip())
+        items.extend(value["items"] if value.get("kind") == "List" else [value])
+        text = text.lstrip()[end:]
+    return items
+
+
+def collect(rendered, contract):
+    validate_contract(contract)
+    def get(args):
+        return json.loads(checked_run(['kubectl', *args, '-o', 'json']))
+    nodes = get(['get', 'nodes'])
+    if contract['estate_ip'] not in [a['address'] for n in nodes['items'] for a in n['status']['addresses']]:
+        raise ValueError('estate identity unverified')
+    settings = tracking_settings(contract)
+    parsed = decode_objects(checked_run(['kubectl', 'create', '--dry-run=client', '--validate=false', '-f', '-', '-o', 'json'], input=rendered))
+    desired = normalize(parsed, contract, settings)
+    payload = json.dumps({'apiVersion': 'v1', 'kind': 'List', 'items': desired}, sort_keys=True).encode()
+    diff = subprocess.run(['kubectl', 'diff', '-f', '-'], input=payload, capture_output=True, timeout=120)
+    if diff.returncode not in (0, 1):
+        raise ValueError('native client diff failed; protected output suppressed')
+    obj = get(['-n', 'argocd', 'get', 'application', contract['application']])
+    spec, status = obj['spec'], obj.get('status', {})
+    op = status.get('operationState', {})
+    sync = op.get('operation', {}).get('sync', {})
+    checks = []
+    for check in contract['ready_checks']:
+        r = get(['-n', check['namespace'], 'get', check['kind'], check['name']])
+        ready = [{k: c.get(k) for k in ('status', 'lastTransitionTime')} for c in r.get('status', {}).get('conditions', []) if c.get('type') == check['condition']]
+        group = r['apiVersion'].split('/')[0] if '/' in r['apiVersion'] else ''
+        tracking = f"{contract['application']}:{group}/{r['kind']}:{check['namespace']}/{check['name']}"
+        checks.append({**check, 'uid': r['metadata']['uid'], 'ready': ready,
+                       'tracking_matches': r['metadata'].get('annotations', {}).get('argocd.argoproj.io/tracking-id') == tracking,
+                       'observed_generation': r.get('status', {}).get('observedGeneration'), 'generation': r['metadata'].get('generation'),
+                       'replicas': r.get('status', {}).get('replicas'), 'available_replicas': r.get('status', {}).get('availableReplicas'),
+                       'instances': r.get('status', {}).get('instances'), 'ready_instances': r.get('status', {}).get('readyInstances')})
+    jobs = get(['-n', contract['destination']['namespace'], 'get', 'jobs'])
+    live_hooks = sorted(r['metadata']['name'] for r in jobs['items'] if r['metadata']['name'] in contract['hooks'])
+    return {'captured_at': datetime.now(timezone.utc).isoformat(), 'estate_ip': contract['estate_ip'], 'tracking_settings': settings,
+        'application': {'name': obj['metadata']['name'], 'uid': obj['metadata']['uid'], 'project': spec.get('project'),
+            'source': {k: spec.get('source', {}).get(k) for k in ('repoURL', 'path', 'targetRevision')}, 'destination': spec.get('destination'),
+            'automated_present': 'automated' in spec.get('syncPolicy', {}), 'pending_operation': bool(obj.get('operation')),
+            'sync': status.get('sync', {}).get('status'), 'health': status.get('health', {}).get('status'), 'revision': status.get('sync', {}).get('revision'),
+            'condition_types': [c.get('type') for c in status.get('conditions', [])],
+            'operation': {k: op.get(k) for k in ('phase', 'startedAt', 'finishedAt')},
+            'sync_prune': sync.get('prune', False), 'sync_strategy_apply': 'apply' in sync.get('syncStrategy', {}),
+            'sync_strategy_force': sync.get('syncStrategy', {}).get('apply', {}).get('force', False), 'operation_revision': sync.get('revision'),
+            'resources': [{k: r.get(k) for k in ('group', 'kind', 'namespace', 'name', 'status')} for r in status.get('resources', [])]},
+        'ready_checks': checks, 'live_hooks': live_hooks,
+        'pinned_render_diff': {'revision': contract['revision'], 'path': contract['source_path'], 'client_side_exit_code': diff.returncode,
+            'render_sha256': hashlib.sha256(rendered).hexdigest(), 'normalized_render_sha256': hashlib.sha256(payload).hexdigest(),
+            'render_resource_count': len(desired), 'skipped_hooks': sorted(contract['hooks']), 'output_suppressed': True}}
+
+
+def verify(observation, contract, now=None):
+    validate_contract(contract)
+    now = now or datetime.now(timezone.utc)
+    captured = timestamp(observation['captured_at'])
+    if not 0 <= (now - captured).total_seconds() <= contract['maximum_observation_age_seconds']:
+        raise ValueError('observation stale or future')
+    if not -5 <= (captured - timestamp(observation['metadata_captured_at'])).total_seconds() <= contract['maximum_observation_age_seconds']:
+        raise ValueError('metadata observation stale or clock skew exceeds bound')
+    settings = observation['tracking_settings']
+    if (observation['estate_ip'] != contract['estate_ip'] or settings.get('method') != contract['tracking_method']
+            or settings.get('controller_version') != contract['controller_version'] or settings.get('installation_id_absent') is not True):
+        raise ValueError('estate or tracking semantics unverified')
+    app = observation['application']
+    if (app['name'] != contract['application'] or not app['uid'] or app['project'] != contract['project'] or app['destination'] != contract['destination']
+            or app['source'] != {'repoURL': contract['repo_url'], 'path': contract['source_path'], 'targetRevision': contract['revision']}
+            or app['revision'] != contract['revision']):
+        raise ValueError('source or application identity mismatch')
+    if (app['automated_present'] is not False or app['pending_operation'] is not False or app['sync'] != 'Synced' or app['health'] != 'Healthy'
+            or any(c != 'OrphanedResourceWarning' for c in app['condition_types'])):
+        raise ValueError('manual idle healthy adoption unverified')
+    op = app['operation']
+    if (op['phase'] != 'Succeeded' or app['sync_prune'] is not False or app['sync_strategy_apply'] is not True or app['sync_strategy_force'] is not False or app['operation_revision'] != contract['revision']):
+        raise ValueError('pinned manual apply strategy unverified')
+    start, finish = timestamp(op['startedAt']), timestamp(op['finishedAt'])
+    proof_age = (captured - finish).total_seconds()
+    if finish < start or proof_age < contract['minimum_proving_seconds']:
+        raise ValueError('completed 24h proving period unverified')
+    expected = {identity(r) for r in contract['resources']}
+    if (len(app['resources']) != len(expected) or {identity(r) for r in app['resources']} != expected or any(r['status'] != 'Synced' for r in app['resources'])):
+        raise ValueError('exact synced resource set unverified')
+    checks = observation['ready_checks']
+    expected_checks = {(c['kind'], c['namespace'], c['name'], c['condition']) for c in contract['ready_checks']}
+    if len(checks) != len(expected_checks) or {(c['kind'], c['namespace'], c['name'], c['condition']) for c in checks} != expected_checks:
+        raise ValueError('readiness checks incomplete')
+    for c in checks:
+        if not c['uid'] or c['tracking_matches'] is not True or len(c['ready']) != 1 or c['ready'][0]['status'] != 'True':
+            raise ValueError('tracked resource readiness unverified')
+        if timestamp(c['ready'][0]['lastTransitionTime']) > start:
+            raise ValueError('readiness changed during proving period')
+        if c['kind'] == 'Deployment' and (c['observed_generation'] != c['generation'] or not c['replicas'] or c['available_replicas'] != c['replicas']):
+            raise ValueError('deployment availability incomplete')
+        if c['kind'] == 'Cluster' and (not c['instances'] or c['ready_instances'] != c['instances']):
+            raise ValueError('CNPG readiness incomplete')
+    diff = observation['pinned_render_diff']
+    if (diff.get('revision') != contract['revision'] or diff.get('path') != contract['source_path'] or type(diff.get('client_side_exit_code')) is not int
+            or diff['client_side_exit_code'] != 0 or diff.get('render_resource_count') != len(expected)
+            or diff.get('skipped_hooks') != sorted(contract['hooks']) or diff.get('output_suppressed') is not True
+            or any(not re.fullmatch(r'[a-f0-9]{64}', diff.get(k, '')) for k in ('render_sha256', 'normalized_render_sha256'))):
+        raise ValueError('pinned normalized client zero-diff unverified')
+    if observation.get('live_hooks') != [] or observation.get('public_health') != {'url': contract['health_url'], 'status': 200}:
+        raise ValueError('hook absence or public health unverified')
+    return {'schema_version': 'argocd-workload-adoption-proof/v1', 'status': 'verified', 'application': contract['application'],
+            'revision': contract['revision'], 'captured_at': observation['captured_at'], 'verified_at': now.isoformat(),
+            'proving_seconds': proof_age, 'resource_count': len(expected), 'ready_count': len(checks),
+            'manual_apply_prune_off': True, 'hooks_skipped': sorted(contract['hooks']), 'hooks_absent': True,
+            'child_automation_enabled': False, 'public_health_status': 200, 'native_mutation': False,
+            'verifier_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+            'tracking_helper_sha256': hashlib.sha256(Path(__file__).with_name('argocd_adoption_proof.py').read_bytes()).hexdigest(),
+            'contract_sha256': hashlib.sha256(json.dumps(contract, sort_keys=True).encode()).hexdigest(),
+            'observation_sha256': hashlib.sha256(json.dumps(observation, sort_keys=True).encode()).hexdigest()}
+
+
+def remote_program(common, rendered):
+    import base64
+    return ('import sys,types,base64,json\n' + 'm=types.ModuleType("argocd_adoption_proof");sys.modules[m.__name__]=m\n'
+            + 'exec(compile(base64.b64decode(' + repr(base64.b64encode(common).decode()) + '),"argocd_adoption_proof.py","exec"),m.__dict__)\n'
+            + 'n={"__name__":"workload_proof"};exec(compile(base64.b64decode(' + repr(base64.b64encode(Path(__file__).read_bytes()).decode())
+            + '),"workload_adoption_proof.py","exec"),n)\n'
+            + 'try:\n print(json.dumps(n["collect"](base64.b64decode(' + repr(base64.b64encode(rendered).decode()) + '),n["CONTRACT"]),sort_keys=True))\n'
+            + 'except Exception:\n raise SystemExit("adoption observation incomplete; protected output suppressed")\n').encode()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--live', action='store_true')
+    parser.add_argument('--source-root', type=Path)
+    parser.add_argument('--observation', type=Path, required=True)
+    parser.add_argument('--receipt', type=Path, required=True)
+    args = parser.parse_args()
+    try:
+        if args.live:
+            if not args.source_root:
+                raise ValueError('source root required')
+            import tempfile
+            with tempfile.TemporaryDirectory(prefix='workload-adoption-pin-') as temp:
+                archive = checked_run(['git', '-C', str(args.source_root), 'archive', CONTRACT['revision'], CONTRACT['source_path']])
+                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))
+            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']
+            observation['captured_at'] = datetime.now(timezone.utc).isoformat()
+            args.observation.write_text(json.dumps(observation, indent=2) + '\n')
+        else:
+            observation = json.loads(args.observation.read_text())
+        receipt = verify(observation, CONTRACT)
+    except Exception:
+        receipt = {'schema_version': 'argocd-workload-adoption-proof/v1', 'status': 'unverified',
+                   'error': 'adoption proof incomplete; protected output suppressed', 'native_mutation': False}
+    args.receipt.write_text(json.dumps(receipt, indent=2) + '\n')
+    print(json.dumps(receipt, sort_keys=True))
+    return 0 if receipt['status'] == 'verified' else 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())