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

import argparse
import asyncio
import hashlib
import json
import os
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'
SERVICE = 'hermes-gateway-dualcoachtest.service'
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)

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 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 extract_wheels(tmp: Path) -> tuple[Path,Path]:
    h=tmp/'hermes'; p=tmp/'profile'; h.mkdir();p.mkdir()
    with zipfile.ZipFile(CANDIDATE/'artifacts/hermes_agent-0.17.0-py3-none-any.whl') as z:z.extractall(h)
    with zipfile.ZipFile(CANDIDATE/'artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl') as z:z.extractall(p)
    return h,p

def prepare(lineage: str) -> dict[str,Any]:
    customer='task26_strict_a_'+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)
    wheel_paths: list[str] = []
    with tempfile.TemporaryDirectory(prefix='task26-wheel-api-') as d:
        h,p=extract_wheels(Path(d)); wheel_paths=[str(p),str(h)]; sys.path[:0]=wheel_paths
        from checkin_cli.customer_admin import CustomerDraft as AdminDraft, register_customer
        from gateway.platforms.telegram_customer_bootstrap import CustomerDraft,RoomBootstrapStore,room_bootstrap_state_dir
        today=date.today()
        register_customer(registry,AdminDraft(customer,'Task26 strict A',ACTOR,ACTOR,'0',today,daytime(8),0,1,2000,150,('meal_1','meal_2','meal_3')))
        draft=CustomerDraft(customer,'Task26 strict A',today.isoformat(),'08:00',0,1,2000,150,('meal_1','meal_2','meal_3'))
        prepared=RoomBootstrapStore(room_bootstrap_state_dir(PROFILE)).prepare_rehearsal_customer_invite(draft,bot_username=BOT_USERNAME,owner_id=OWNER)
    sys.path[:len(wheel_paths)] = []
    for module_name in tuple(sys.modules):
        if module_name == 'checkin_cli' or module_name.startswith('checkin_cli.') or module_name == 'gateway' or module_name.startswith('gateway.'):
            del sys.modules[module_name]
    row={'customer_id':customer,'session_id':prepared.session.session_id,'sid_hash':prepared.session.sid_hash,'generation':1,'customer_link':prepared.customer_link}
    return row

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')
    receipt={'schema':'task26-strict-deployment-v1','candidate_digest':FULL,'candidate_core_digest':CORE,'candidate_inventory_digest':PACKAGE_SHA,'hermes_wheel_sha256':HERMES_SHA,'profile_wheel_sha256':PROFILE_SHA,**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)
    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]) -> dict[str,Any]:
    import yaml
    from telegram import Bot
    from gateway.platforms.telegram_staff_membership_gate import build_staff_chat_inventory,create_pre_activation_evidence
    registry=json.loads((PROFILE/'customers/registry.json').read_text()); config=yaml.safe_load((PROFILE/'config.yaml').read_text()); inventory=build_staff_chat_inventory(registry,config)
    if inventory.sha256!=runtime['staff_chat_inventory_sha256']: raise Blocked('staff inventory drift')
    token=config['platforms']['telegram']['token']
    async with Bot(token) as bot:
        return await 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]:
    from gateway.platforms.dualcoach_admin import 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]) -> dict[str,Any]:
    import yaml
    from telegram import 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(config['platforms']['telegram']['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
    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 execute(root: Path) -> dict[str,Any]:
    marker=root/'ONE-USE'; atomic(marker,{'status':'CLAIMED','controller_sha256':sha(Path(__file__))},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))
        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})
        invite=asyncio.run(send_once(root,prepared)); invited=True
        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-v1','status':'READY_CUSTOMER_CLAIM','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) -> 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-v1','status':'PASS_ZERO_MUTATIONS','candidate':FULL,'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); a=p.parse_args()
    try:
        a.root.mkdir(mode=0o700,parents=True,exist_ok=True); result=dry_run(a.root) if a.mode=='dry-run' else execute(a.root); 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())