#!/usr/bin/env python3
"""Exact sealed-candidate, one-use DualCoach v8 rehearsal launcher."""
from __future__ import annotations
import argparse, asyncio, base64, csv, ctypes, hashlib, importlib, importlib.metadata, io, json, os, select, stat, subprocess, sys, time, uuid, zipfile
from datetime import UTC, date, datetime, time as daytime
from pathlib import Path
from typing import Any, Callable
from zoneinfo import ZoneInfo

PROFILE=Path('/home/cube/.hermes/profiles/dualcoachtest')
os.environ['HERMES_HOME']=str(PROFILE)
PYTHON=Path('/home/cube/.hermes/profiles/dualcoachtest/.strict-runtime/f32a8365-v8/venv/bin/python')
CANDIDATE=Path('/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-clarification-root-final-st_01a013d9')
ROOT=CANDIDATE/'deployment/v8'
FULL='f32a836571b403663da3510d930fd04c1a84c5514cb6345fc575f5a42b580da5'; PRODUCT='5795a78f7830760b10a4b4d909ca44b9142b92e86083d7d47e9c787f5e580d97'
CANDIDATE_SEAL_SHA='8c7c33d6b41d1c2b29f58aa39696df6a8ffe2794b9bc0a4ddf5b16182e1c5695'; RECORD_SEAL_SHA='20d3221230c544eb2210c9393c58afad34876df0f80c50d74c31afac891ec75c'; CONFIG_SHA='f93106b16643227e2ef9dec67a5bbd497e1d353e779da62287898a087071af87'
MANIFEST_SHA='3639c41eb6e76eb7f8b4ae98fe76a644c2c0574ac6d94bf3c395bf16ec0115f9'; HERMES_SHA='c673e7b04c564e6ebbf5010e6e0ad321af3abde7565d6d3ca66d238c9248877b'; PROFILE_SHA='aba7b92da1de6b9f9e071f8734026683c7bcdfd1abaa754ab2c8b6b4d5825fd7'
ACTOR='8527916639'; OWNER='8693203710'; BOT_USERNAME='dual_coach_pilot_test_bot'
UNIT='hermes-gateway-dualcoachtest-strict-f32a8365-v8.service'; OBSERVER_UNIT='dualcoach-lifecycle-observer-f32a8365-v8.service'; BASE_SERVICE='hermes-gateway-dualcoachtest.service'
PROTECTED=('config.yaml','auth.json','.env','customers/registry.json','gateway_state.json','data/owner-actions/draft-deliveries.json','data/scheduled-deliveries.jsonl','data/onboarding/telegram-publication-outbox-v1/ledger.json')
IN_EVENTS=0x8|0x80|0x100
class Blocked(RuntimeError): pass

def sha(path:Path)->str: return hashlib.sha256(path.read_bytes()).hexdigest()
def command(args:list[str],ok:bool=True)->subprocess.CompletedProcess[str]:
 r=subprocess.run(args,text=True,capture_output=True,check=False)
 if ok and r.returncode: raise Blocked(f'command failed ({r.returncode}): {args!r}: {r.stderr.strip()}')
 return r
def append(receipt:Path,event:str,**values:Any)->None:
 raw=(json.dumps({'event':event,'at_utc':datetime.now(UTC).isoformat(),**values},sort_keys=True,separators=(',',':'),ensure_ascii=True)+'\n').encode()
 fd=os.open(receipt,os.O_WRONLY|os.O_APPEND|os.O_CREAT|os.O_CLOEXEC|getattr(os,'O_NOFOLLOW',0),0o600)
 try: os.fchmod(fd,0o600); os.write(fd,raw); os.fsync(fd)
 finally: os.close(fd)
def state(unit:str)->dict[str,str]:
 r=command(['/usr/bin/systemctl','--user','show',unit,'-p','ActiveState','-p','SubState','-p','MainPID'],False)
 return dict(x.split('=',1) for x in r.stdout.splitlines() if '=' in x)
def flags()->dict[str,bool]:
 import yaml
 a=yaml.safe_load((PROFILE/'config.yaml').read_text())['platforms']['telegram']['extra']['adaptive_nutrition']
 return {k:a[k] for k in ('activation','delivery','delivery_enabled')}
def delivery_count()->int:
 total=0
 for path in (PROFILE/'data/scheduled-deliveries.jsonl',PROFILE/'data/owner-actions/draft-deliveries.json',PROFILE/'data/onboarding/telegram-publication-outbox-v1/ledger.json'):
  if not path.exists(): continue
  try: value=json.loads(path.read_text())
  except json.JSONDecodeError: total+=len(path.read_text().splitlines()); continue
  if isinstance(value,list): total+=len(value)
  elif isinstance(value,dict): total+=sum(len(value.get(k,[])) for k in ('deliveries','records','items','entries'))
 return total
def baseline()->dict[str,Any]:
 if state(BASE_SERVICE)!={'MainPID':'0','ActiveState':'inactive','SubState':'dead'}: raise Blocked('installed service not inactive/dead')
 units=command(['/usr/bin/systemctl','--user','list-units','--all','--no-legend','--plain'],False).stdout
 active=[line for line in units.splitlines() if ('dualcoach-lifecycle-observer-' in line or 'hermes-gateway-dualcoachtest-strict-' in line) and not (' inactive ' in line or ' failed ' in line)]
 if active: raise Blocked('previous strict unit active: '+repr(active))
 if state(UNIT).get('ActiveState') not in (None,'inactive','failed') or state(OBSERVER_UNIT).get('ActiveState') not in (None,'inactive','failed'): raise Blocked('v8 gateway/observer already active')
 processes=command(['/usr/bin/pgrep','-af','hermes-gateway-dualcoachtest-strict-|dualcoach-lifecycle-observer-'],False).stdout.strip()
 if processes: raise Blocked('previous strict process active: '+processes)
 forbidden=[PROFILE/x for x in ('customers/registry.json','gateway.lock','gateway.pid','gateway_state.json','state.db','state.db-shm','state.db-wal','sessions','data/onboarding','data/owner-actions','data/customers','data/scheduled-deliveries.jsonl')]
 live=[str(p) for p in forbidden if p.exists()]
 if live: raise Blocked('live authorities not empty: '+','.join(live))
 safe=flags()
 if safe!={'activation':False,'delivery':False,'delivery_enabled':False}: raise Blocked('config flags not false')
 if delivery_count(): raise Blocked('deliveries not zero')
 private={str(p):format(stat.S_IMODE(p.stat().st_mode),'04o') for p in (PROFILE,PROFILE/'config.yaml',PROFILE/'auth.json',PROFILE/'.env')}
 if private[str(PROFILE)]!='0700' or any(private[str(PROFILE/x)]!='0600' for x in ('config.yaml','auth.json','.env')): raise Blocked('profile configuration is not private')
 import yaml
 gate=importlib.import_module('gateway.platforms.telegram_staff_membership_gate')
 skeleton={'owner':{'user_id':OWNER,'chat_id':OWNER,'topic_id':'0'}}
 inventory=gate.build_staff_chat_inventory(skeleton,yaml.safe_load((PROFILE/'config.yaml').read_text()))
 staff_users={str(row.staff_user_id) for row in inventory.rows if row.staff_user_id is not None}
 if ACTOR==OWNER or ACTOR in staff_users or OWNER not in staff_users: raise Blocked('customer/staff owner separation failed')
 return {'service':'inactive/dead','authorities':'empty','flags':safe,'deliveries':0,'monitors':0,'private_modes':private,'customer_actor_absent_from_staff':True,'owner_separated':True,'staff_chat_inventory_sha256':inventory.sha256}
def _record_parity(dist_name:str,wheel_path:Path)->dict[str,Any]:
 dist=importlib.metadata.distribution(dist_name); site=PYTHON.parent.parent/'lib/python3.12/site-packages'; venv=PYTHON.parent.parent
 installed_text=dist.read_text('RECORD')
 if installed_text is None: raise Blocked('installed RECORD missing: '+dist_name)
 installed_rows=list(csv.reader(io.StringIO(installed_text))); installed_by_path={row[0]:row for row in installed_rows}
 verified=0
 for relative,encoded,size_text in installed_rows:
  target=(site/relative).resolve()
  if venv.resolve() not in target.parents or not target.is_file(): raise Blocked(f'installed RECORD path absent/escaped: {dist_name}:{relative}')
  if size_text and target.stat().st_size!=int(size_text): raise Blocked(f'installed RECORD size mismatch: {dist_name}:{relative}')
  if encoded:
   algorithm,value=encoded.split('=',1)
   if algorithm!='sha256': raise Blocked('non-sha256 installed RECORD entry')
   actual=base64.urlsafe_b64encode(hashlib.sha256(target.read_bytes()).digest()).rstrip(b'=').decode()
   if actual!=value: raise Blocked(f'installed RECORD hash mismatch: {dist_name}:{relative}')
  verified+=1
 with zipfile.ZipFile(wheel_path) as archive:
  record_name=next(name for name in archive.namelist() if name.endswith('.dist-info/RECORD')); wheel_rows=list(csv.reader(io.StringIO(archive.read(record_name).decode())))
 data_prefix=record_name.removesuffix('.dist-info/RECORD')+'.data/'
 for relative,encoded,size_text in wheel_rows:
  mapped=relative
  if relative.startswith(data_prefix):
   scheme,separator,suffix=relative[len(data_prefix):].partition('/')
   if not separator: raise Blocked('invalid wheel data path')
   if scheme in {'purelib','platlib'}: mapped=suffix
   elif scheme=='data': mapped=Path(os.path.relpath(venv/suffix,site)).as_posix()
   else: raise Blocked('unsupported wheel data scheme: '+scheme)
  installed=installed_by_path.get(mapped)
  if installed is None or installed[1:]!=[encoded,size_text]: raise Blocked(f'wheel/installed RECORD mismatch: {dist_name}:{relative}')
 return {'distribution':dist_name,'installed_record_entries':len(installed_rows),'wheel_record_entries':len(wheel_rows),'verified_files':verified,'wheel_sha256':sha(wheel_path)}
def installed_proof()->dict[str,Any]:
 if Path(sys.executable).resolve()!=PYTHON.resolve() or sys.prefix!=str(PYTHON.parent.parent): raise Blocked('wrong isolated interpreter')
 if os.environ.get('PYTHONPATH'): raise Blocked('PYTHONPATH must be unset')
 seal=json.loads((CANDIDATE/'SEAL.json').read_text()); manifest=json.loads((CANDIDATE/'candidate-manifest.json').read_text())
 if sha(CANDIDATE/'SEAL.json')!=CANDIDATE_SEAL_SHA or sha(CANDIDATE/'candidate-manifest.json')!=MANIFEST_SHA or sha(CANDIDATE/'seals/installed-runtime-record-parity-final.json')!=RECORD_SEAL_SHA: raise Blocked('sealed evidence drift')
 if seal.get('status')!='INSTALLED_GOLDEN_PATH_PASS' or seal.get('candidate_digest')!=FULL or seal.get('product_digest')!=PRODUCT or manifest.get('candidate_digest')!=FULL or manifest.get('product_digest')!=PRODUCT: raise Blocked('candidate/product binding mismatch')
 origins={}
 for name in ('gateway','gateway.platforms.telegram','gateway.platforms.dualcoach_admin','gateway.platforms.telegram_customer_bootstrap','gateway.platforms.telegram_staff_membership_gate','checkin_cli','checkin_cli.customer_admin','telegram'):
  module_file=importlib.import_module(name).__file__
  if not isinstance(module_file,str): raise Blocked('import has no origin: '+name)
  origin=Path(module_file).resolve()
  if PYTHON.parent.parent not in origin.parents: raise Blocked('non-isolated import: '+name)
  origins[name]=str(origin)
 direct={}; records=[]
 for dist,wheel,want in (('hermes-agent','hermes_agent-0.17.0-py3-none-any.whl',HERMES_SHA),('physique-checkin-cli','physique_checkin_cli-0.1.0-py3-none-any.whl',PROFILE_SHA)):
  wheel_path=CANDIDATE/'artifacts'/wheel
  if sha(wheel_path)!=want: raise Blocked('sealed wheel mismatch: '+dist)
  direct_text=importlib.metadata.distribution(dist).read_text('direct_url.json')
  if direct_text is None: raise Blocked('direct_url missing: '+dist)
  value=json.loads(direct_text); source=Path(value['url'].replace('file://','').replace('%20',' '))
  if source.resolve()!=wheel_path.resolve() or sha(source)!=want: raise Blocked('direct_url wheel mismatch: '+dist)
  direct[dist]={'url':value['url'],'wheel_sha256':want}; records.append(_record_parity(dist,wheel_path))
 return {'interpreter':sys.executable,'site_packages':str(PYTHON.parent.parent/'lib/python3.12/site-packages'),'origins':origins,'direct_url':direct,'record_parity':records,'candidate_seal_sha256':CANDIDATE_SEAL_SHA,'record_seal_sha256':RECORD_SEAL_SHA}
def wait_path(path:Path,predicate:Callable[[Path],bool],timeout:float)->None:
 if path.exists() and predicate(path): return
 path.parent.mkdir(parents=True,exist_ok=True)
 libc=ctypes.CDLL('libc.so.6',use_errno=True); fd=libc.inotify_init1(os.O_CLOEXEC)
 if fd<0 or libc.inotify_add_watch(fd,os.fsencode(path.parent),IN_EVENTS)<0: raise Blocked('inotify subscription failed')
 poll=select.poll(); poll.register(fd,select.POLLIN); deadline=time.monotonic()+timeout
 try:
  while True:
   left=deadline-time.monotonic()
   if left<=0 or not poll.poll(max(1,int(left*1000))): raise Blocked('event timeout: '+str(path))
   os.read(fd,65536)
   if path.exists() and predicate(path): return
 finally: os.close(fd)
def prepare()->dict[str,Any]:
 admin=importlib.import_module('checkin_cli.customer_admin'); boot=importlib.import_module('gateway.platforms.telegram_customer_bootstrap')
 customer='task26_claim_'+datetime.now(UTC).strftime('%Y%m%d%H%M%S')+'_'+uuid.uuid4().hex[:8]; registry=PROFILE/'customers/registry.json'; registry.parent.mkdir(mode=0o700)
 registry.write_text(json.dumps({'version':1,'owner':{'user_id':OWNER,'chat_id':OWNER,'topic_id':'0'},'customers':[],'registry_mode':'ordinary_v1','diagnostic_session_digest':None},separators=(',',':'))+'\n'); registry.chmod(0o600)
 today=date.today(); admin.register_customer(registry,admin.CustomerDraft(customer,'DualCoach customer claim',ACTOR,ACTOR,'0',today,daytime(8),0,1,2000,150,('meal_1','meal_2','meal_3')))
 draft=boot.CustomerDraft(customer,'DualCoach customer claim',today.isoformat(),'08:00',0,1,2000,150,('meal_1','meal_2','meal_3'))
 made=boot.RoomBootstrapStore(boot.room_bootstrap_state_dir(PROFILE)).prepare_rehearsal_customer_invite(draft,bot_username=BOT_USERNAME,owner_id=OWNER)
 if made.session.expires_at-made.session.created_at != __import__('datetime').timedelta(minutes=30): raise Blocked('invite TTL is not exactly 30 minutes')
 kst=ZoneInfo('Asia/Seoul')
 return {'customer_id':customer,'customer_key':made.session.customer_key,'session_id':made.session.session_id,'generation':1,'sid_hash':made.session.sid_hash,'invite_ttl_seconds':1800,'invite_created_at_utc':made.session.created_at.isoformat(),'invite_expires_at_utc':made.session.expires_at.isoformat(),'invite_created_at_kst':made.session.created_at.astimezone(kst).isoformat(),'invite_expires_at_kst':made.session.expires_at.astimezone(kst).isoformat(),'customer_link':made.customer_link}
def start_runtime(temp:Path)->dict[str,Any]:
 ready=ROOT/'monitor/observer-ready.json'; events=ROOT/'monitor/observer-events.jsonl'
 command(['/usr/bin/systemd-run','--user',f'--unit={OBSERVER_UNIT}','--collect','--property=Type=exec',str(PYTHON),str(ROOT/'lifecycle_observer_v8.py'),'observe','--profile',str(PROFILE),'--events',str(events),'--ready',str(ready)])
 wait_path(ready,lambda p:json.loads(p.read_text()).get('status')=='ARMED',10)
 gateway_state=PROFILE/'gateway_state.json'; journal=PROFILE/'data/onboarding/telegram-staff-membership-v1/events.jsonl'
 command(['/usr/bin/systemd-run','--user',f'--unit={UNIT}','--collect','--property=Type=simple',f'--property=WorkingDirectory={PROFILE}',f'--setenv=HERMES_HOME={PROFILE}',f'--setenv=VIRTUAL_ENV={PYTHON.parent.parent}',f'--setenv=PATH={PYTHON.parent}:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin','--setenv=PYTHONDONTWRITEBYTECODE=1','--setenv=DUALCOACH_PROFILE_PACKAGE=/home/cube/.hermes/profiles/dualcoachtest/.strict-runtime/f32a8365-v8/venv/lib/python3.12/site-packages',str(PYTHON),'-m','hermes_cli.main','--profile','dualcoachtest','gateway','run'])
 wait_path(gateway_state,lambda p:json.loads(p.read_text()).get('platforms',{}).get('telegram',{}).get('state')=='connected',120)
 wait_path(journal,lambda p:any(json.loads(x).get('event')=='subscription_armed' for x in p.read_text().splitlines()),30)
 armed=next(json.loads(x) for x in reversed(journal.read_text().splitlines()) if json.loads(x).get('event')=='subscription_armed')
 return {'gateway':state(UNIT),'observer':state(OBSERVER_UNIT),'subscription_epoch_id':armed['subscription_epoch_id'],'subscription_armed_at_utc':armed['observed_at_utc'],'staff_chat_inventory_sha256':armed['staff_chat_inventory_sha256']}
async def membership(temp:Path,prepared:dict[str,Any],runtime:dict[str,Any],token:str)->dict[str,Any]:
 import yaml
 Bot=importlib.import_module('telegram').Bot; gate=importlib.import_module('gateway.platforms.telegram_staff_membership_gate')
 registry=PROFILE/'customers/registry.json'; inventory=gate.build_staff_chat_inventory(json.loads(registry.read_text()),yaml.safe_load((PROFILE/'config.yaml').read_text()))
 if inventory.sha256!=runtime['staff_chat_inventory_sha256']: raise Blocked('membership inventory drift')
 deployment=temp/'deployment.json'; deployment.write_text(json.dumps({'candidate_digest':FULL,'candidate_core_digest':PRODUCT,'candidate_inventory_digest':PRODUCT,'hermes_wheel_sha256':HERMES_SHA,'profile_wheel_sha256':PROFILE_SHA})); deployment.chmod(0o600)
 async with Bot(token) as bot:
  result=await gate.create_pre_activation_evidence(bot,inventory,output_path=temp/'membership.json',deployment_receipt_path=deployment,registry_path=registry,config_path=PROFILE/'config.yaml',customer_id=prepared['customer_id'],customer_user_id=int(ACTOR),bootstrap_session_id=prepared['session_id'],bootstrap_generation=1,subscription_epoch_id=runtime['subscription_epoch_id'],subscription_armed_at_utc=runtime['subscription_armed_at_utc'])
 groups=result.get('membership_results'); private=result.get('private_dm_results')
 if not isinstance(groups,list) or not groups or any(x.get('status') not in {'left','kicked'} for x in groups): raise Blocked('left/kicked proof missing')
 if not isinstance(private,list) or any(x.get('identity_separated') is not True for x in private): raise Blocked('private identity separation missing')
 return result
async def prestart_membership(temp:Path,token:str)->dict[str,Any]:
 import yaml
 Bot=importlib.import_module('telegram').Bot; gate=importlib.import_module('gateway.platforms.telegram_staff_membership_gate')
 skeleton={'owner':{'user_id':OWNER,'chat_id':OWNER,'topic_id':'0'}}
 inventory=gate.build_staff_chat_inventory(skeleton,yaml.safe_load((PROFILE/'config.yaml').read_text()))
 async with Bot(token) as bot: result=await gate.observe_customer_absence(bot,inventory,customer_user_id=int(ACTOR))
 groups=result.get('membership_results'); private=result.get('private_dm_results')
 if not isinstance(groups,list) or not groups or any(x.get('status') not in {'left','kicked'} for x in groups): raise Blocked('prestart left/kicked proof missing')
 if not isinstance(private,list) or any(x.get('identity_separated') is not True for x in private): raise Blocked('prestart owner separation missing')
 path=temp/'prestart-membership.json'; path.write_text(json.dumps({'customer_user_id':ACTOR,'staff_chat_inventory_sha256':inventory.sha256,**result},sort_keys=True)+'\n'); path.chmod(0o600)
 return {'evidence_sha256':sha(path),'staff_chat_inventory_sha256':inventory.sha256,**result}
def protected_snapshot()->dict[str,Any]:
 files=[]
 for relative in PROTECTED:
  path=PROFILE/relative
  if not path.exists(): files.append({'path':relative,'absent':True}); continue
  info=path.lstat()
  if path.is_symlink() or not path.is_file(): raise Blocked('untrusted protected profile path: '+relative)
  files.append({'path':relative,'bytes':info.st_size,'mode':format(stat.S_IMODE(info.st_mode),'04o'),'sha256':sha(path)})
 encoded=json.dumps(files,sort_keys=True,separators=(',',':')).encode()
 return {'digest':hashlib.sha256(encoded).hexdigest(),'files':files}
def provider(temp:Path)->dict[str,Any]:
 provider_auth_check=importlib.import_module('gateway.platforms.dualcoach_admin').provider_auth_check
 result=provider_auth_check(receipt_directory=temp/'provider',allow_billable_active_probe=True)
 if int(result.exit)!=0 or result.receipt_path is None: raise Blocked('provider readiness failed: '+result.result.value)
 return {'result':result.result.value,'receipt_sha256':sha(result.receipt_path),'probe':'nonpersistent readiness'}
def provider_before_mutation(action:Callable[[],dict[str,Any]])->dict[str,Any]:
 before=protected_snapshot(); result=action(); after=protected_snapshot()
 if before!=after: raise Blocked('provider readiness changed protected profile bytes')
 return {**result,'candidate':FULL,'product':PRODUCT,'candidate_seal_sha256':CANDIDATE_SEAL_SHA,'record_seal_sha256':RECORD_SEAL_SHA,'config_sha256':CONFIG_SHA,'protected_profile_digest':before['digest']}
def ordered_pre_runtime(provider_action:Callable[[],dict[str,Any]],provider_done:Callable[[dict[str,Any]],None],prepare_action:Callable[[],dict[str,Any]],runtime_action:Callable[[],dict[str,Any]])->tuple[dict[str,Any],dict[str,Any],dict[str,Any]]:
 provider_result=provider_before_mutation(provider_action)
 provider_done(provider_result)
 prepared=prepare_action()
 runtime=runtime_action()
 return provider_result,prepared,runtime
async def send(link:str,token:str)->dict[str,str]:
 Bot=importlib.import_module('telegram').Bot
 async with Bot(token) as bot: sent=await bot.send_message(chat_id=int(ACTOR),text='DualCoach 시작 초대입니다. 아래 링크를 눌러 본인이 직접 시작해 주세요.\n'+link,disable_web_page_preview=True)
 return {'message_id':str(sent.message_id),'chat_id':str(sent.chat_id)}
def rollback(root:Path,reason:str)->None:
 for unit in (UNIT,OBSERVER_UNIT): command(['/usr/bin/systemctl','--user','stop',unit],False)
 archive=root/'rollback-archive'; archive.mkdir(mode=0o700,exist_ok=True)
 for rel in ('customers','data/onboarding','data/owner-actions','data/customers','gateway.lock','gateway.pid','gateway_state.json','state.db','state.db-shm','state.db-wal','sessions'):
  source=PROFILE/rel
  if source.exists() and not source.is_symlink():
   target=archive/rel; target.parent.mkdir(parents=True,mode=0o700,exist_ok=True); os.rename(source,target)
 append(root/'receipt.jsonl','CLEAN_ROLLBACK',status='BLOCKED_CLEAN_ROLLBACK',reason=reason,baseline=baseline())
def main()->int:
 parser=argparse.ArgumentParser(); parser.add_argument('mode',choices=('dry-run','execute')); args=parser.parse_args(); root=Path(__file__).resolve().parent; receipt=root/'receipt.jsonl'
 try:
  proof=installed_proof(); base=baseline()
  if args.mode=='dry-run': append(receipt,'DRY_RUN',status='PASS_ZERO_MUTATIONS',installed=proof,baseline=base); return 0
  if receipt.exists() and any(json.loads(x).get('event')=='ONE_USE' for x in receipt.read_text().splitlines()): raise Blocked('one-use marker exists; retry forbidden')
  append(receipt,'ONE_USE',status='CLAIMED_V8',marker=uuid.uuid4().hex)
  temp=ROOT/'private-runtime-evidence'; temp.mkdir(mode=0o700)
  token=importlib.import_module('gateway.config').load_gateway_preflight_inputs(PROFILE).telegram.secret
  def provider_action()->dict[str,Any]: return provider(temp)
  def provider_done(result:dict[str,Any])->None: append(receipt,'PROVIDER_READY',**result)
  def prepare_action()->dict[str,Any]:
   result=prepare(); private=result.pop('customer_link'); result['_private_link']=private; append(receipt,'PREPARED',**{k:v for k,v in result.items() if k!='_private_link'}); return result
  def runtime_action()->dict[str,Any]:
   result=start_runtime(temp); append(receipt,'RUNTIME_READY',**result); return result
  provider_result=provider_before_mutation(provider_action); provider_done(provider_result)
  pre_member=asyncio.run(prestart_membership(temp,token)); append(receipt,'PRESTART_MEMBERSHIP_READY',evidence_sha256=pre_member['evidence_sha256'],membership_results=pre_member['membership_results'],private_dm_results=pre_member['private_dm_results'])
  prepared=prepare_action(); runtime=runtime_action()
  link=prepared.pop('_private_link')
  member=asyncio.run(membership(temp,prepared,runtime,token)); append(receipt,'MEMBERSHIP_READY',evidence_sha256=member['evidence_sha256'],membership_results=member['membership_results'],private_dm_results=member['private_dm_results'])
  if sha(CANDIDATE/'candidate-manifest.json')!=MANIFEST_SHA or sha(CANDIDATE/'SEAL.json')!=CANDIDATE_SEAL_SHA or sha(PROFILE/'config.yaml')!=CONFIG_SHA or flags()!=base['flags'] or delivery_count()!=0: raise Blocked('READY invariant changed')
  append(receipt,'READY',candidate=FULL,config_sha256=CONFIG_SHA,deliveries=0)
  message_hash=hashlib.sha256(('DualCoach 시작 초대입니다. 아래 링크를 눌러 본인이 직접 시작해 주세요.\n'+link).encode()).hexdigest()
  append(receipt,'INVITE_INTENT',status='ATTEMPTING_UNKNOWN_IS_TERMINAL',destination=ACTOR,customer_id=prepared['customer_id'],session_id=prepared['session_id'],message_sha256=message_hash)
  try: sent=asyncio.run(send(link,token))
  except BaseException as exc: append(receipt,'INVITE_UNKNOWN',status='TERMINAL_NO_RETRY',error_type=type(exc).__name__); raise Blocked('invite outcome unknown; terminal no retry') from exc
  append(receipt,'INVITE_RESULT',status='SENT_EXACTLY_ONCE',invite_count=1,**sent)
  handoff=f'Open latest Telegram message ID {sent["message_id"]} and tap Start only.'
  append(receipt,'READY_CUSTOMER_CLAIM',status='READY_CUSTOMER_CLAIM',candidate=FULL,product=PRODUCT,hermes_wheel_sha256=HERMES_SHA,profile_wheel_sha256=PROFILE_SHA,runtime=str(PYTHON.parent.parent),profile_package=str(PYTHON.parent.parent/'lib/python3.12/site-packages'),customer_id=prepared['customer_id'],customer_key=prepared['customer_key'],session_id=prepared['session_id'],sid_hash=prepared['sid_hash'],membership_evidence_sha256=member['evidence_sha256'],provider=provider_result,gateway_unit=UNIT,observer_unit=OBSERVER_UNIT,gateway=state(UNIT),observer=state(OBSERVER_UNIT),subscription_epoch_id=runtime['subscription_epoch_id'],invite_message_id=sent['message_id'],invite_count=1,invite_ttl_seconds=prepared['invite_ttl_seconds'],invite_created_at_utc=prepared['invite_created_at_utc'],invite_expires_at_utc=prepared['invite_expires_at_utc'],invite_created_at_kst=prepared['invite_created_at_kst'],invite_expires_at_kst=prepared['invite_expires_at_kst'],activation=False,delivery=False,delivery_enabled=False,authority_state='READY_CUSTOMER_CLAIM',exact_user_action=handoff)
  print('READY_CUSTOMER_CLAIM'); print(handoff); return 0
 except Exception as exc:
  if args.mode=='execute' and receipt.exists() and any(json.loads(x).get('event')=='ONE_USE' for x in receipt.read_text().splitlines()):
   try: rollback(root,str(exc))
   except Exception as cleanup: print(f'ROLLBACK FAILURE: {cleanup}',file=sys.stderr)
  print(f'BLOCKED: {exc}',file=sys.stderr); return 2
if __name__=='__main__': raise SystemExit(main())