#!/usr/bin/env python3
"""Minimal one-use isolated DualCoach launcher with one append-only receipt."""
from __future__ import annotations
import argparse, asyncio, ctypes, hashlib, importlib, importlib.metadata, json, os, select, shutil, stat, subprocess, sys, time, uuid
from datetime import UTC, date, datetime, time as daytime
from pathlib import Path
from typing import Any, Callable

PROFILE=Path('/home/cube/.hermes/profiles/dualcoachtest')
os.environ['HERMES_HOME']=str(PROFILE)
PYTHON=Path('/home/cube/.hermes/profiles/dualcoachtest/.strict-runtime/dac4e812/venv/bin/python')
CANDIDATE=Path('/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-strict-final-candidate-successor-v3-st_01a00f35')
RUNTIME=Path('/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-isolated-runtime-dac4e812-st_01a0122b')
FULL='b6d78bc1e68ead7340d92b534fe3a6d0257c8aa2b02cd3d58e43215fbd8a3443'; PRODUCT='dac4e81281e8ab7f5c46461e79d405998e0273fe9f98dc09ff67b73528083092'
RUNTIME_SEAL='6283a4cf71ba9982c8e75bce88f09997f6eb786a88b788f7858199aeaadb27a9'; CONFIG_SHA='f93106b16643227e2ef9dec67a5bbd497e1d353e779da62287898a087071af87'
HERMES_SHA='ee787fb63fd8ce61d35eb2d7a43c11d7e7adef268c1d09a0a40c1d881b8ea742'; PROFILE_SHA='fb48a1931828ee60ab3812faf795550bf4e672a54c56f78cd10dcf674c45560d'
ACTOR='8527916639'; OWNER='8693203710'; BOT_USERNAME='dual_coach_pilot_test_bot'
UNIT='hermes-gateway-dualcoachtest-strict-dac4e812-v2.service'; OBSERVER_UNIT='dualcoach-lifecycle-observer-b6d78bc1-v2.service'; BASE_SERVICE='hermes-gateway-dualcoachtest.service'
PROTECTED=('config.yaml','auth.json','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')
 if state(UNIT).get('ActiveState') not in (None,'inactive','failed') or state(OBSERVER_UNIT).get('ActiveState') not in (None,'inactive','failed'): raise Blocked('transient gateway/observer already active')
 forbidden=[PROFILE/x for x in ('customers/registry.json','gateway.lock','gateway.pid','gateway_state.json','state.db','sessions','data/onboarding','data/owner-actions','data/customers')]
 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')
 return {'service':'inactive/dead','authorities':'empty','flags':safe,'deliveries':0,'monitors':0}
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 sha(RUNTIME/'SEAL.json')!=RUNTIME_SEAL: raise Blocked('runtime seal drift')
 seal=json.loads((RUNTIME/'SEAL.json').read_text()); receipt=json.loads((RUNTIME/'runtime-receipt.json').read_text())
 if seal.get('status')!='READY_ISOLATED_RUNTIME' or seal.get('candidate')!=FULL or seal.get('preserved_product_candidate')!=PRODUCT: raise Blocked('runtime binding mismatch')
 if receipt.get('status')!='READY_ISOLATED_RUNTIME' or receipt.get('verification',{}).get('profile_config',{}).get('sha256')!=CONFIG_SHA: raise Blocked('runtime receipt mismatch')
 origins={}
 for name in ('gateway','gateway.platforms.telegram','gateway.platforms.dualcoach_admin','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={}
 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)):
  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.name!=wheel or sha(source)!=want: raise Blocked('direct_url wheel mismatch: '+dist)
  direct[dist]={'url':value['url'],'wheel_sha256':want}
 return {'interpreter':sys.executable,'origins':origins,'direct_url':direct,'runtime_seal_sha256':RUNTIME_SEAL}
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)
 return {'customer_id':customer,'session_id':made.session.session_id,'generation':1,'sid_hash':made.session.sid_hash,'customer_link':made.customer_link}
def start_runtime(temp:Path)->dict[str,Any]:
 ready=temp/'observer-ready.json'; events=temp/'observer-events.jsonl'
 command(['/usr/bin/systemd-run','--user',f'--unit={OBSERVER_UNIT}','--collect','--property=Type=exec',str(PYTHON),str(CANDIDATE/'lifecycle_observer_v7.py'),'observe','--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',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
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,'runtime_seal_sha256':RUNTIME_SEAL,'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_V2',marker=uuid.uuid4().hex)
  temp=Path(f'/run/user/{os.getuid()}/dualcoach-v2-{uuid.uuid4().hex}'); 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,prepared,runtime=ordered_pre_runtime(provider_action,provider_done,prepare_action,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')!='df51caefa45e62541f0656468195766ea6c4bb6f589b8b61ee16360befcda6f4' 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='Customer 8527916639: open the single DualCoach invite in this private DM and tap Start yourself; do not send onboarding answers yet.'
  append(receipt,'READY_CUSTOMER_CLAIM',status='READY_CUSTOMER_CLAIM',customer_id=prepared['customer_id'],session_id=prepared['session_id'],membership_evidence_sha256=member['evidence_sha256'],provider=provider_result,gateway=state(UNIT),observer=state(OBSERVER_UNIT),invite_message_id=sent['message_id'],invite_count=1,activation=False,delivery=False,exact_user_action=handoff)
  shutil.rmtree(temp); 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())