#!/usr/bin/env python3
"""One-use, fail-closed Task26 strict-A launch controller."""
from __future__ import annotations

import argparse
import asyncio
import hashlib
import importlib
import importlib.metadata
import importlib.util
import json
import os
import signal
import stat
import subprocess
import sys
import tempfile
import time
import uuid
import zipfile
from datetime import UTC, date, datetime, time as daytime
from pathlib import Path
from typing import Any, Callable

ROOT = Path('/home/cube/projects/richard/traning coach')
CANDIDATE = ROOT/'.omo/evidence/task26/task26-strict-final-candidate-st_01a00f35'
PROFILE = Path('/home/cube/.hermes/profiles/dualcoachtest')
HERMES = Path('/home/cube/projects/richard/hermes-agent')
PYTHON = HERMES/'.venv/bin/python'
SITE_PACKAGES = HERMES/'.venv/lib/python3.12/site-packages'
SERVICE = 'hermes-gateway-dualcoachtest.service'
PREDECESSOR = ROOT/'.omo/evidence/task26/task26-strict-launch-controller-v4-st_01a00f53'
V4_CONTROLLER_SEAL_SHA = '99c2343645e5238568d1263a347587461d48edead1f3ce30aa9d804cee9b7c0e'
V4_CLEAN_RUN_SEAL_SHA = '8f9111b2ac21ac1ae1fb3f0af50bd1ff9b522e3db14cf1cf886b13cb2ab9f4b0'
FULL = '573d19e464c7df2a0dcacddbb915447beb20c97aceb7c1d1306415014d5730d7'
CORE = '90b2557bb159dc7595ea6bcd3d76d1a0dabf77c51f78aa0ffcf237746d669b49'
HERMES_SHA = 'b87aeaca4abb39ce43f6c9dc35eb47e9c3665ec363882aa286d47fb74ec1b7dc'
PROFILE_SHA = 'fb48a1931828ee60ab3812faf795550bf4e672a54c56f78cd10dcf674c45560d'
PACKAGE_SHA = '262d345eb795c55ca8076abf96d2262be4a7fe9a05405fee7277dea7c59d0175'
OBSERVER_SHA = 'fb95738da52c4d0c6f3e9664f9ce2d657bba8845a555b08444acba61e346f570'
ACTOR = '8527916639'
OWNER = '8693203710'
BOT_USERNAME = 'dual_coach_pilot_test_bot'
BOUNDARIES = ('verified','baseline','prepared','deployed','started','subscribed','membership','provider','ready','invite_intent','invite_result')
FLAGS = os.O_CLOEXEC | getattr(os, 'O_NOFOLLOW', 0)
MODULE_MEMBERS = {
    'gateway.platforms.telegram': 'gateway/platforms/telegram.py',
    'gateway.platforms.telegram_customer_bootstrap': 'gateway/platforms/telegram_customer_bootstrap.py',
    'gateway.platforms.telegram_staff_membership_gate': 'gateway/platforms/telegram_staff_membership_gate.py',
    'gateway.platforms.dualcoach_admin': 'gateway/platforms/dualcoach_admin.py',
    'checkin_cli': 'checkin_cli/__init__.py',
    'checkin_cli.customer_admin': 'checkin_cli/customer_admin.py',
}
CHILDREN: list[subprocess.Popen[Any]] = []

class Blocked(RuntimeError): pass

def canon(value: Any) -> bytes:
    return (json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=True)+'\n').encode()

def sha_bytes(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest()
def sha(path: Path) -> str: return sha_bytes(path.read_bytes())

def validate_interpreter_record(record: dict[str, Any]) -> None:
    if record.get('executable') != str(PYTHON): raise Blocked('wrong interpreter')
    if record.get('prefix') != str(HERMES/'.venv'): raise Blocked('wrong virtual environment')
    if record.get('telegram_origin') != str(SITE_PACKAGES/'telegram/__init__.py'): raise Blocked('telegram missing or path drift')
    if record.get('execstart_python') != str(PYTHON): raise Blocked('service ExecStart interpreter drift')
    if record.get('service_virtual_env') != str(HERMES/'.venv'): raise Blocked('service environment drift')
    if record.get('working_directory') != str(PROFILE): raise Blocked('service working-directory drift')
    modules=record.get('modules',{})
    for name,member in MODULE_MEMBERS.items():
        row=modules.get(name,{})
        if row.get('origin') != str(SITE_PACKAGES/member): raise Blocked(f'import path drift: {name}')
        if row.get('loaded_sha256') != row.get('wheel_sha256'): raise Blocked(f'import byte drift: {name}')
    if record.get('hermes_wheel_sha256') != HERMES_SHA or record.get('profile_wheel_sha256') != PROFILE_SHA: raise Blocked('installed wheel drift')
    if record.get('v4_controller_seal_sha256') != V4_CONTROLLER_SEAL_SHA: raise Blocked('v4 controller seal drift')
    if record.get('v4_clean_run_seal_sha256') != V4_CLEAN_RUN_SEAL_SHA: raise Blocked('v4 clean run seal drift')

def interpreter_proof() -> dict[str, Any]:
    if sys.executable != str(PYTHON): raise Blocked(f'wrong interpreter: {sys.executable}')
    try:
        telegram=importlib.import_module('telegram')
    except ImportError as exc:
        raise Blocked('telegram unavailable in required interpreter') from exc
    wheels={
        'gateway': CANDIDATE/'artifacts/hermes_agent-0.17.0-py3-none-any.whl',
        'checkin_cli': CANDIDATE/'artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl',
    }
    modules: dict[str, Any]={}
    archives={key:zipfile.ZipFile(path) for key,path in wheels.items()}
    try:
        for name,member in MODULE_MEMBERS.items():
            module=importlib.import_module(name); module_file=getattr(module,'__file__',None)
            if not isinstance(module_file,str): raise Blocked(f'import has no file: {name}')
            origin=Path(module_file).resolve()
            archive=archives['gateway' if name.startswith('gateway') else 'checkin_cli']
            modules[name]={'origin':str(origin),'loaded_sha256':sha(origin),'wheel_sha256':sha_bytes(archive.read(member))}
    finally:
        for archive in archives.values(): archive.close()
    show=run(['/usr/bin/systemctl','--user','show',SERVICE,'-p','ExecStart','-p','Environment','-p','WorkingDirectory']).stdout
    fields=dict(x.split('=',1) for x in show.splitlines() if '=' in x)
    execstart=fields.get('ExecStart',''); environment=fields.get('Environment','')
    telegram_file=getattr(telegram,'__file__',None)
    if not isinstance(telegram_file,str): raise Blocked('telegram import has no file')
    record={'schema':'task26-v5-interpreter-proof-v1','executable':sys.executable,'prefix':sys.prefix,'telegram_origin':str(Path(telegram_file).resolve()),'execstart_python':execstart.split('path=',1)[1].split(' ',1)[0] if 'path=' in execstart else None,'service_virtual_env':next((x.split('=',1)[1] for x in environment.split() if x.startswith('VIRTUAL_ENV=')),None),'working_directory':fields.get('WorkingDirectory'),'modules':modules,'hermes_wheel_sha256':sha(wheels['gateway']),'profile_wheel_sha256':sha(wheels['checkin_cli']),'v4_controller_seal_sha256':sha(PREDECESSOR/'SEAL-v4.json'),'v4_clean_run_seal_sha256':sha(PREDECESSOR/'run-20260817-strict-a-v4/RUN-SEAL-v4.json')}
    validate_interpreter_record(record); return record

def verify_permission(path: Path) -> dict[str, Any]:
    value=json.loads(path.read_text())
    expected={'status':'AUTHORIZED_ONE_USE_V5','controller_sha256':sha(Path(__file__)),'required_interpreter':str(PYTHON),'v4_controller_seal_sha256':V4_CONTROLLER_SEAL_SHA,'v4_clean_run_seal_sha256':V4_CLEAN_RUN_SEAL_SHA}
    if any(value.get(k)!=v for k,v in expected.items()): raise Blocked('v5 permission mismatch')
    return value

def resolver_contract() -> dict[str, Any]:
    import yaml
    raw=yaml.safe_load((PROFILE/'config.yaml').read_text())
    telegram_cfg=raw.get('platforms',{}).get('telegram',{})
    if not isinstance(telegram_cfg,dict) or 'token' in telegram_cfg: raise Blocked('literal config token forbidden')
    env_loader=importlib.import_module('hermes_cli.env_loader'); gateway_config=importlib.import_module('gateway.config')
    env_module_file=getattr(env_loader,'__file__',None); config_module_file=getattr(gateway_config,'__file__',None)
    if not isinstance(env_module_file,str) or not isinstance(config_module_file,str): raise Blocked('authoritative credential resolver path unavailable')
    env_file=Path(env_module_file).resolve(); config_file=Path(config_module_file).resolve()
    if not callable(getattr(env_loader,'load_hermes_dotenv',None)) or not callable(getattr(gateway_config,'load_gateway_config',None)): raise Blocked('authoritative credential resolver unavailable')
    return {'schema':'task26-v5-resolver-contract-v1','literal_config_token':False,'network_used':False,'resolver':'hermes_cli.env_loader.load_hermes_dotenv','bootstrap':'gateway.config.load_gateway_config','resolver_sha256':sha(env_file),'bootstrap_sha256':sha(config_file)}

def extract_telegram_credential(config: Any, platform: Any) -> str:
    platforms=getattr(config,'platforms',None)
    telegram=getattr(platform,'TELEGRAM',None)
    resolved=platforms.get(telegram) if isinstance(platforms,dict) else None
    token=getattr(resolved,'token',None)
    if not isinstance(token,str) or not token.strip(): raise Blocked('telegram credential unavailable')
    return token

def resolve_telegram_credential() -> tuple[str,dict[str,Any]]:
    contract=resolver_contract()
    try:
        env_loader=importlib.import_module('hermes_cli.env_loader'); gateway_config=importlib.import_module('gateway.config')
        old_home=os.environ.get('HERMES_HOME'); os.environ['HERMES_HOME']=str(PROFILE)
        try:
            loaded=env_loader.load_hermes_dotenv(hermes_home=PROFILE,project_env=HERMES/'.env')
            config=gateway_config.load_gateway_config()
        finally:
            if old_home is None: os.environ.pop('HERMES_HOME',None)
            else: os.environ['HERMES_HOME']=old_home
        token=extract_telegram_credential(config,gateway_config.Platform)
    except Blocked: raise
    except Exception as exc: raise Blocked('authoritative credential resolution failed') from exc
    proof={**contract,'credential_resolved':True,'credential_stored':False,'credential_source_files':[str(Path(p).resolve()) for p in loaded]}
    return token,proof

def validate_network_identity(identity: Any, readiness_bot_id: Any) -> tuple[int,str]:
    bot_id=getattr(identity,'id',None); username=getattr(identity,'username',None)
    if type(bot_id) is not int or bot_id<=0 or username!=BOT_USERNAME or readiness_bot_id!=bot_id: raise Blocked('wrong Telegram bot identity or administrator readiness')
    return bot_id,username

async def telegram_network_preflight(token: str) -> dict[str,Any]:
    import yaml
    Bot=importlib.import_module('telegram').Bot; gate=importlib.import_module('gateway.platforms.telegram_staff_membership_gate')
    raw=yaml.safe_load((PROFILE/'config.yaml').read_text())
    synthetic={'version':1,'registry_mode':'ordinary_v1','diagnostic_session_digest':None,'owner':{'user_id':OWNER,'chat_id':OWNER,'topic_id':'0'},'customers':[]}
    inventory=gate.build_staff_chat_inventory(synthetic,raw)
    try:
        async with Bot(token) as bot:
            identity=await asyncio.wait_for(bot.get_me(),timeout=15)
            readiness_id=await asyncio.wait_for(gate.verify_subscription_readiness(bot,inventory),timeout=30)
    except Exception as exc: raise Blocked('bounded read-only Telegram identity/admin preflight failed') from exc
    bot_id,username=validate_network_identity(identity,readiness_id)
    return {'schema':'task26-v5-telegram-preflight-v1','status':'PASS_READ_ONLY','operations':['get_me','get_chat_member(bot_admin)'],'bot_id':str(bot_id),'bot_username':username,'staff_chat_inventory_sha256':inventory.sha256,'staff_chat_count':len(inventory.rows),'credential_stored':False}

def write_all(fd: int, raw: bytes) -> None:
    view = memoryview(raw)
    while view:
        n = os.write(fd, view)
        if type(n) is not int or n <= 0 or n > len(view): raise Blocked('incomplete write')
        view = view[n:]

def atomic(path: Path, value: Any, *, exclusive: bool = False) -> None:
    path.parent.mkdir(parents=True, mode=0o700, exist_ok=True); path.parent.chmod(0o700)
    if path.is_symlink(): raise Blocked(f'symlink output: {path}')
    if exclusive and path.exists(): raise Blocked(f'one-use output exists: {path}')
    tmp = path.with_name('.'+path.name+'.'+uuid.uuid4().hex+'.tmp')
    fd = os.open(tmp, os.O_WRONLY|os.O_CREAT|os.O_EXCL|FLAGS, 0o600)
    try:
        os.fchmod(fd, 0o600); write_all(fd, canon(value)); os.fsync(fd)
    finally: os.close(fd)
    if exclusive and path.exists(): tmp.unlink(); raise Blocked(f'one-use output exists: {path}')
    os.replace(tmp, path); os.chmod(path, 0o600)
    dfd=os.open(path.parent,os.O_RDONLY|os.O_DIRECTORY|FLAGS)
    try: os.fsync(dfd)
    finally: os.close(dfd)

def run(command: list[str], **kw: Any) -> subprocess.CompletedProcess[str]:
    p=subprocess.run(command,text=True,capture_output=True,check=False,**kw)
    if p.returncode: raise Blocked(f'command failed ({p.returncode}): {command!r}: {p.stderr.strip()}')
    return p

def service() -> dict[str,str]:
    p=run(['/usr/bin/systemctl','--user','show',SERVICE,'-p','ActiveState','-p','SubState','-p','MainPID'])
    return dict(x.split('=',1) for x in p.stdout.splitlines() if '=' in x)

def config_flags() -> dict[str,bool]:
    import yaml
    x=yaml.safe_load((PROFILE/'config.yaml').read_text()); extra=x['platforms']['telegram']['extra']; a=extra['adaptive_nutrition']
    return {'delivery_enabled':a['delivery_enabled'],'activation':a['activation'],'delivery':a['delivery']}

def verify_candidate() -> dict[str,Any]:
    seal=json.loads((CANDIDATE/'SEAL.json').read_text()); inv=json.loads((CANDIDATE/'PACKAGE-INVENTORY.json').read_text())
    expected={'full_candidate_digest':FULL,'core_candidate_digest':CORE,'hermes_wheel_sha256':HERMES_SHA,'profile_wheel_sha256':PROFILE_SHA,'package_inventory_sha256':PACKAGE_SHA,'lifecycle_observer_v7_sha256':OBSERVER_SHA}
    if seal.get('status')!='READY_STRICT_REHEARSAL' or any(seal.get(k)!=v for k,v in expected.items()): raise Blocked('candidate seal mismatch')
    if sha(CANDIDATE/'PACKAGE-INVENTORY.json')!=PACKAGE_SHA: raise Blocked('package inventory drift')
    for row in inv:
        p=CANDIDATE/row['path']
        if not p.is_file() or p.stat().st_size!=row['size'] or sha(p)!=row['sha256']: raise Blocked(f'candidate inventory mismatch: {row["path"]}')
    for name in ('objective','security','trainer-free','quality','qa','provenance'):
        r=json.loads((CANDIDATE/f'review-{name}.json').read_text())
        if r.get('status') != 'READY_STRICT_REHEARSAL': raise Blocked(f'review not ready: {name}')
    run([str(PYTHON),str(CANDIDATE/'verify_candidate.py')])
    return expected

def baseline() -> dict[str,Any]:
    if service()!= {'MainPID':'0','ActiveState':'inactive','SubState':'dead'}: raise Blocked('service not inactive/dead')
    forbidden=[PROFILE/'customers/registry.json',PROFILE/'gateway.lock',PROFILE/'gateway.pid',PROFILE/'gateway_state.json',PROFILE/'state.db',PROFILE/'sessions',PROFILE/'data/onboarding',PROFILE/'data/owner-actions',PROFILE/'data/customers']
    unexpected=[str(p) for p in forbidden if p.exists()]
    jobs=PROFILE/'cron/jobs.json'
    if jobs.exists() and json.loads(jobs.read_text()).get('jobs',[])!=[]: unexpected.append(str(jobs))
    if unexpected: raise Blocked('unexpected live bytes must be archived by cleanup controller: '+','.join(unexpected))
    flags=config_flags()
    if flags!={'delivery_enabled':False,'activation':False,'delivery':False}: raise Blocked('config safety flags mismatch')
    ps=run(['/usr/bin/ps','-u',str(os.getuid()),'-o','args=']).stdout
    if any(('dualcoachtest' in l or 'task26' in l) and any(x in l for x in ('observer','journalctl -f','gateway run')) for l in ps.splitlines() if str(Path(__file__).resolve()) not in l): raise Blocked('stale target process')
    return {'service':'inactive/dead','flags':flags,'unexpected_live_bytes':0}

def prepare(lineage: str) -> dict[str,Any]:
    customer='task26_strict_a_v5_'+lineage
    registry=PROFILE/'customers/registry.json'; registry.parent.mkdir(mode=0o700)
    initial={'version':1,'owner':{'user_id':OWNER,'chat_id':OWNER,'topic_id':'0'},'customers':[],'registry_mode':'ordinary_v1','diagnostic_session_digest':None}
    atomic(registry,initial,exclusive=True)
    admin=importlib.import_module('checkin_cli.customer_admin')
    bootstrap=importlib.import_module('gateway.platforms.telegram_customer_bootstrap')
    today=date.today()
    admin.register_customer(registry,admin.CustomerDraft(customer,'Task26 strict A v5',ACTOR,ACTOR,'0',today,daytime(8),0,1,2000,150,('meal_1','meal_2','meal_3')))
    draft=bootstrap.CustomerDraft(customer,'Task26 strict A v5',today.isoformat(),'08:00',0,1,2000,150,('meal_1','meal_2','meal_3'))
    prepared=bootstrap.RoomBootstrapStore(bootstrap.room_bootstrap_state_dir(PROFILE)).prepare_rehearsal_customer_invite(draft,bot_username=BOT_USERNAME,owner_id=OWNER)
    return {'customer_id':customer,'session_id':prepared.session.session_id,'sid_hash':prepared.session.sid_hash,'generation':1,'customer_link':prepared.customer_link}

def deploy(out: Path) -> dict[str,Any]:
    hw=CANDIDATE/'artifacts/hermes_agent-0.17.0-py3-none-any.whl'; pw=CANDIDATE/'artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl'
    run([str(PYTHON),'-m','pip','install','--no-deps','--no-compile','--force-reinstall',str(hw),str(pw)])
    code='''import hashlib,importlib.metadata as m,json\nfrom pathlib import Path\nimport gateway.platforms.telegram as t,checkin_cli\ndef d(n):\n x=m.distribution(n).read_text("direct_url.json");return json.loads(x)\nprint(json.dumps({"hermes_origin":str(Path(t.__file__).resolve()),"profile_origin":str(Path(checkin_cli.__file__).resolve()),"hermes_direct":d("hermes-agent"),"profile_direct":d("physique-checkin-cli")},sort_keys=True))'''
    proof=json.loads(run([str(PYTHON),'-c',code]).stdout)
    for key,want in (('hermes_direct',HERMES_SHA),('profile_direct',PROFILE_SHA)):
        got=proof[key]['archive_info']['hashes']['sha256']
        if got!=want: raise Blocked(f'installed {key} mismatch')
    loaded=interpreter_proof()
    receipt={'schema':'task26-strict-deployment-v5','candidate_digest':FULL,'candidate_core_digest':CORE,'candidate_inventory_digest':PACKAGE_SHA,'hermes_wheel_sha256':HERMES_SHA,'profile_wheel_sha256':PROFILE_SHA,'loaded_byte_proof':loaded['modules'],**proof}
    atomic(out,receipt,exclusive=True); return receipt

def wait_file(path: Path, timeout: float, predicate: Callable[[Any],bool]) -> Any:
    deadline=time.monotonic()+timeout
    while time.monotonic()<deadline:
        if path.exists():
            try:
                value=json.loads(path.read_text())
                if predicate(value): return value
            except (OSError,json.JSONDecodeError): pass
        time.sleep(.05)
    raise Blocked(f'bounded event timeout: {path}')

def start_runtime(root: Path) -> dict[str,Any]:
    journal=root/'journal.txt'; jf=open(journal,'w',encoding='utf-8')
    monitor=subprocess.Popen(['/usr/bin/journalctl','--user','-u',SERVICE,'-f','-n0','--no-pager','-o','cat'],stdout=jf,stderr=subprocess.STDOUT,start_new_session=True)
    events=root/'observer-events.jsonl'; ready=root/'observer-ready.json'
    observer=subprocess.Popen([str(PYTHON),str(CANDIDATE/'lifecycle_observer_v7.py'),'observe','--events',str(events),'--ready',str(ready)],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,start_new_session=True)
    CHILDREN.extend((monitor,observer))
    wait_file(ready,5,lambda x:x.get('status')=='ARMED')
    run(['/usr/bin/systemctl','--user','start',SERVICE])
    deadline=time.monotonic()+90
    while time.monotonic()<deadline:
        s=service(); state=PROFILE/'gateway_state.json'
        if s.get('ActiveState')=='active' and s.get('SubState')=='running' and state.exists():
            x=json.loads(state.read_text())
            if x.get('gateway_state')=='running' and x.get('platforms',{}).get('telegram',{}).get('state')=='connected': break
        time.sleep(.2)
    else: raise Blocked('gateway readiness timeout')
    membership=PROFILE/'data/onboarding/telegram-staff-membership-v1/events.jsonl'
    deadline=time.monotonic()+10
    armed=None
    while time.monotonic()<deadline:
        if membership.exists():
            rows=[json.loads(x) for x in membership.read_text().splitlines()]
            armed=next((x for x in reversed(rows) if x.get('event')=='subscription_armed'),None)
            if armed: break
        time.sleep(.05)
    if not armed: raise Blocked('membership subscription not armed')
    return {'service_pid':int(service()['MainPID']),'observer_pid':observer.pid,'monitor_pid':monitor.pid,'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(receipt: Path,deployment: 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=json.loads((PROFILE/'customers/registry.json').read_text()); config=yaml.safe_load((PROFILE/'config.yaml').read_text()); inventory=gate.build_staff_chat_inventory(registry,config)
    if inventory.sha256!=runtime['staff_chat_inventory_sha256']: raise Blocked('staff inventory drift')
    async with Bot(token) as bot:
        return await gate.create_pre_activation_evidence(bot,inventory,output_path=receipt,deployment_receipt_path=deployment,registry_path=PROFILE/'customers/registry.json',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'])

def provider(receipt_dir: Path) -> dict[str,Any]:
    provider_auth_check=importlib.import_module('gateway.platforms.dualcoach_admin').provider_auth_check
    r=provider_auth_check(receipt_directory=receipt_dir,allow_billable_active_probe=True)
    if r.exit!=0: raise Blocked('provider readiness failed: '+r.result.value)
    return {'result':r.result.value,'receipt_sha256':sha(r.receipt_path),'receipt_path':str(r.receipt_path)}

async def send_once(root: Path, prepared: dict[str,Any], token: str) -> dict[str,Any]:
    import yaml
    Bot=importlib.import_module('telegram').Bot
    intent=root/'invite-intent.json'; result=root/'invite-result.json'
    if intent.exists() or result.exists(): raise Blocked('invite was already attempted; retry forbidden')
    message='DualCoach 시작 초대입니다. 아래 링크를 눌러 본인이 직접 시작해 주세요.\n'+prepared['customer_link']
    atomic(intent,{'schema':'task26-invite-intent-v1','status':'ATTEMPTING_NEVER_RETRY','idempotency_key':sha_bytes((FULL+prepared['session_id']).encode()),'customer_id':prepared['customer_id'],'session_id':prepared['session_id'],'destination':ACTOR,'message_sha256':sha_bytes(message.encode())},exclusive=True)
    config=yaml.safe_load((PROFILE/'config.yaml').read_text())
    try:
        async with Bot(token) as bot:
            sent=await bot.send_message(chat_id=int(ACTOR),text=message,disable_web_page_preview=True)
    except BaseException as exc:
        atomic(root/'invite-unknown.json',{'schema':'task26-invite-unknown-v1','status':'UNKNOWN_NO_RETRY','error_type':type(exc).__name__},exclusive=True); raise Blocked('invite outcome unknown; no retry permitted') from exc
    value={'schema':'task26-invite-result-v1','status':'SENT_EXACTLY_ONCE','message_id':str(sent.message_id),'chat_id':str(sent.chat_id),'idempotency_key':sha_bytes((FULL+prepared['session_id']).encode())}
    atomic(result,value,exclusive=True); return value

def rollback(root: Path, reason: str) -> None:
    try: subprocess.run(['/usr/bin/systemctl','--user','stop',SERVICE],capture_output=True,timeout=220)
    except Exception: pass
    for child in CHILDREN:
        if child.poll() is None:
            try: os.killpg(child.pid,signal.SIGTERM); child.wait(timeout=10)
            except Exception:
                try: os.killpg(child.pid,signal.SIGKILL)
                except Exception: pass
    subprocess.run(['/usr/bin/systemctl','--user','reset-failed',SERVICE],capture_output=True)
    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'):
        p=PROFILE/rel
        if p.exists() and not p.is_symlink():
            q=archive/rel; q.parent.mkdir(parents=True,mode=0o700,exist_ok=True); os.rename(p,q)
    atomic(root/'blocked.json',{'status':'BLOCKED_CLEAN_ROLLBACK','reason':reason})

def assert_secret_absent(root: Path, token: str) -> None:
    needle=token.encode()
    for path in root.rglob('*'):
        if path.is_file() and not path.is_symlink():
            try: raw=path.read_bytes()
            except OSError: continue
            if needle in raw: raise Blocked(f'credential leaked into evidence: {path.name}')

def execute(root: Path, interpreter: dict[str,Any], permission: dict[str,Any], token: str, resolver: dict[str,Any], network: dict[str,Any]) -> dict[str,Any]:
    marker=root/'ONE-USE-v5'; atomic(marker,{'status':'CLAIMED','controller_sha256':sha(Path(__file__)),'permission_sha256':sha(Path(permission['_path']))},exclusive=True)
    proof={k:v for k,v in interpreter.items()}; atomic(root/'00-interpreter-v5.json',proof,exclusive=True)
    atomic(root/'00-secret-resolution-v5.json',resolver,exclusive=True); atomic(root/'00-network-preflight-v5.json',network,exclusive=True)
    lineage=datetime.now(UTC).strftime('%Y%m%d%H%M%S')+'_'+uuid.uuid4().hex[:8]
    invited=False
    try:
        pins=verify_candidate(); atomic(root/'01-verified.json',pins)
        base=baseline(); atomic(root/'02-baseline.json',base)
        prepared=prepare(lineage); private_link=prepared.pop('customer_link'); atomic(root/'03-prepared.json',prepared)
        prepared['customer_link']=private_link
        deployment=deploy(root/'04-deployment.json')
        runtime=start_runtime(root); atomic(root/'05-runtime.json',runtime)
        member=asyncio.run(membership(root/'06-membership.json',root/'04-deployment.json',prepared,runtime,token))
        membership_rows=member.get('membership_results')
        private_rows=member.get('private_dm_results')
        if not isinstance(membership_rows,list) or not membership_rows or any(not isinstance(row,dict) or row.get('status') not in {'left','kicked'} for row in membership_rows): raise Blocked('fresh left/kicked membership evidence missing')
        if not isinstance(private_rows,list) or any(not isinstance(row,dict) or row.get('identity_separated') is not True for row in private_rows): raise Blocked('private staff identity separation missing')
        provider_result=provider(root/'provider')
        if verify_candidate()!=pins or config_flags()!=base['flags']: raise Blocked('candidate/config changed after startup')
        atomic(root/'07-ready.json',{'status':'READY_TO_INVITE','membership_evidence_sha256':member['evidence_sha256'],'provider':provider_result})
        assert_secret_absent(root,token)
        invite=asyncio.run(send_once(root,prepared,token)); invited=True
        assert_secret_absent(root,token)
        handoff='Customer 8527916639: open the single DualCoach invite in this private DM and tap Start yourself; do not send onboarding answers yet.'
        final={'schema':'task26-strict-launch-result-v5','status':'READY_CUSTOMER_CLAIM','interpreter':str(PYTHON),'v4_controller_seal_sha256':V4_CONTROLLER_SEAL_SHA,'v4_clean_run_seal_sha256':V4_CLEAN_RUN_SEAL_SHA,'candidate':FULL,'customer_id':prepared['customer_id'],'session_id':prepared['session_id'],'service':service(),'observer_pid':runtime['observer_pid'],'membership_evidence_sha256':member['evidence_sha256'],'invite_message_id':invite['message_id'],'invite_count':1,'first_user_instruction':handoff,'activation':False,'delivery':False,'delivery_capability_issued':False}
        atomic(root/'READY_CUSTOMER_CLAIM.json',final,exclusive=True); return final
    except Exception as exc:
        if not invited: rollback(root,str(exc))
        raise

def dry_run(root: Path, interpreter: dict[str,Any], resolver: dict[str,Any]) -> dict[str,Any]:
    before=tree_snapshot(PROFILE); pins=verify_candidate(); base=baseline(); after=tree_snapshot(PROFILE)
    if before!=after: raise Blocked('dry-run mutated profile')
    value={'schema':'task26-strict-launch-dry-run-v5','status':'PASS_ZERO_MUTATIONS','candidate':FULL,'interpreter_proof':interpreter,'secret_resolution_contract':resolver,'checks':pins,'baseline':base,'planned_boundaries':BOUNDARIES}
    atomic(root/'dry-run.json',value,exclusive=True); return value

def tree_snapshot(root: Path) -> str:
    rows=[]
    for p in sorted(root.rglob('*')):
        try: st=p.lstat()
        except FileNotFoundError: continue
        rel=p.relative_to(root).as_posix()
        if p.is_symlink(): rows.append((rel,'L',os.readlink(p)))
        elif p.is_file(): rows.append((rel,'F',st.st_size,st.st_mtime_ns,sha(p)))
        elif p.is_dir(): rows.append((rel,'D',stat.S_IMODE(st.st_mode)))
    return sha_bytes(canon(rows))

class BoundaryModel:
    """Deterministic model used to prove ordering/crash/unknown-outcome policy."""
    def __init__(self) -> None: self.index=0; self.attempted=False; self.sent=0
    def cross(self,name:str,*,crash_before:bool=False,crash_after:bool=False,unknown:bool=False)->None:
        if name!=BOUNDARIES[self.index]: raise Blocked('order violation')
        if crash_before: raise RuntimeError('crash-before')
        if name=='invite_intent':
            if self.attempted:
                raise Blocked('duplicate invite')
            self.attempted=True
        if name=='invite_result':
            if not self.attempted: raise Blocked('missing intent')
            if unknown: raise Blocked('unknown-no-retry')
            self.sent+=1
        self.index+=1
        if crash_after: raise RuntimeError('crash-after')

def main() -> int:
    p=argparse.ArgumentParser(); p.add_argument('mode',choices=('dry-run','execute')); p.add_argument('--root',type=Path,required=True); p.add_argument('--permission',type=Path,required=True); a=p.parse_args()
    try:
        interpreter=interpreter_proof(); permission=verify_permission(a.permission); permission['_path']=str(a.permission); resolver=resolver_contract()
        if a.mode=='execute':
            token,resolver=resolve_telegram_credential(); network=asyncio.run(telegram_network_preflight(token))
            a.root.mkdir(mode=0o700,parents=True,exist_ok=True); result=execute(a.root,interpreter,permission,token,resolver,network)
        else:
            a.root.mkdir(mode=0o700,parents=True,exist_ok=True); result=dry_run(a.root,interpreter,resolver)
        print(json.dumps(result,sort_keys=True)); return 0
    except Exception as exc: print('BLOCKED:',exc,file=sys.stderr); return 2
if __name__=='__main__': raise SystemExit(main())