#!/usr/bin/env python3
"""Archive-first controller for Telegram update 629525116 only."""
from __future__ import annotations
import argparse, datetime, fcntl, hashlib, json, os, stat, subprocess, sys, time
from pathlib import Path
from typing import Any

TARGET=629525116; UNIT="hermes-gateway-dualcoachtest.service"
PROFILE=Path("/home/cube/.hermes/profiles/dualcoachtest")
HERMES=Path("/home/cube/projects/richard/hermes-agent"); PYTHON=HERMES/".venv/bin/python"
SITE=HERMES/".venv/lib/python3.12/site-packages"
RECEIPT=PROFILE/"data/telegram-ingress-receipts-v1-d0aacf0f4bdbb7c0.json"; LOCK=Path(str(RECEIPT)+".lock")
CUSTOMER="task26_live_2e_r2_20260815_8527916639"
WHEEL="2110071bc2761e6cbd149de4694dd7328148ec7e5b2e3c7c10a2709082d5e161"
CANDIDATE="f38d0373a58877806ff64a9cac54ab01e6e6e47f7101e3dcfd5ffbf05f8bb101"
TELEGRAM_MODULE="5d3103b744b6e92ddbc576a77df433c2127d6df17260a3c78d2000d7b90d2f97"
PROFILE_WHEEL="f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6"
AUTHORITY_DIGEST="8856ca5146cd74a6e9ed05dcd06747a694cd4f8415ee0575d673d04c9e310f09"
OBSERVER_AUTHORITY="791f79115641c3d3fda1b754f0cebaf986e51a855cd5289a24c9b636f7e015b0"
DEPLOYMENT_RECEIPT="8cd6edb7775804302703e20153f23127dd709adbcc24fe81bffb89e5f33f7136"
DIRECT_URL_HASH="f4bb54f774200c04d7f6d2f5566a4a47237d81672277da52428527aa9641b5fc"
EDITABLE={"__editable__.physique_checkin_cli-0.1.0.pth":"c469ba412e431af58b157129ad3e6e3e3229638eceab22baa09fdff155f4252c","__editable___physique_checkin_cli_0_1_0_finder.py":"2dc57c797ed4b59b7a2d08a9bc1b241e1527d3dad2809b8b6b43a67a975b95fb","physique_checkin_cli-0.1.0.dist-info/direct_url.json":"498294b3c9066aae5f8cb07b5ba63a53d404cc2c03145f468cc79bf76317f92c"}
OBSERVER=Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-continuous-lifecycle-observer-v6.2-f38d0373")
CHECKLIST=Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-activation-checklist-6aaef77d-task26-live-2e-r2/activation-checklist-receipt.json")
CONTINUATION=Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/task26-telegram-poll-recovery-st_01a00d2c/run-2/application-watch.log")
CONTINUATION_HASH="687b4c053b424f7a4d2fa8875e01e4338410dba63c0481077959587b96ca4bf1"

class Closed(RuntimeError): pass
def sha(b:bytes)->str:return hashlib.sha256(b).hexdigest()
def fsha(p:Path)->str:return sha(p.read_bytes())
def load(p:Path)->Any:return json.loads(p.read_text())
def run(*a:str,check:bool=True):return subprocess.run(a,text=True,capture_output=True,check=check)
def create(p:Path,b:bytes)->None:
 p.parent.mkdir(mode=0o700,parents=True,exist_ok=True); fd=os.open(p,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
 try:
  if os.write(fd,b)!=len(b):raise Closed("partial evidence write")
  os.fsync(fd)
 finally:os.close(fd)
 d=os.open(p.parent,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW);os.fsync(d);os.close(d)
def emit(p:Path,v:object)->None:create(p,json.dumps(v,indent=2,sort_keys=True).encode()+b"\n")
def safe_file(p:Path)->os.stat_result:
 s=p.lstat()
 if not stat.S_ISREG(s.st_mode) or s.st_nlink!=1 or stat.S_IMODE(s.st_mode)!=0o600:raise Closed(f"unsafe metadata: {p}")
 return s
def read_safe(p:Path)->bytes:
 safe_file(p);fd=os.open(p,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
 try:
  s=os.fstat(fd)
  if not stat.S_ISREG(s.st_mode) or s.st_nlink!=1 or stat.S_IMODE(s.st_mode)!=0o600:raise Closed("unsafe descriptor")
  out=[]
  while b:=os.read(fd,65536):out.append(b)
  return b"".join(out)
 finally:os.close(fd)

def fields()->dict[str,str]:
 r=run("systemctl","--user","show",UNIT,"-p","ActiveState","-p","SubState","-p","MainPID","-p","InvocationID","-p","Result")
 return dict(x.split("=",1) for x in r.stdout.splitlines() if "=" in x)
def stop()->None:
 run("systemctl","--user","stop",UNIT);end=time.monotonic()+45
 while time.monotonic()<end:
  s=fields()
  if s.get("ActiveState")!="active" and s.get("MainPID")=="0":return
  time.sleep(.1)
 raise Closed("service stop timeout")

def parse_receipt(raw:bytes)->dict[str,Any]:
 try:v=json.loads(raw)
 except Exception as e:raise Closed("receipt JSON invalid") from e
 if not isinstance(v,dict) or v.get("version") not in (1,2,3):raise Closed("receipt version invalid")
 keys={1:{"version","receipts"},2:{"version","receipts","terminal_receipts"},3:{"version","receipts","terminal_receipts","terminal_failures"}}[v["version"]]
 if set(v)!=keys or any(not isinstance(v.get(k),dict) for k in keys-{"version"}):raise Closed("receipt schema keys invalid")
 return v
def validate_target(v:dict[str,Any])->None:
 t=str(TARGET)
 if v.get("version")!=3 or v.get("terminal_failures")!={t:{"reason_code":"handler_exception","stage":"blocked_receipt"}}:raise Closed("target is not sole handler_exception")
 if t in v["receipts"] or t in v["terminal_receipts"]:raise Closed("target receipt conflict")
 for k,x in v["receipts"].items():
  if not k.isdecimal() or x!={"reason_code":"handled","stage":"receipt"}:raise Closed("normal receipt invalid")
 for k,x in v["terminal_receipts"].items():
  if not k.isdecimal() or not isinstance(x,dict) or x.get("reason_code")!="business_commit_reconciled" or x.get("stage")!="recovery_receipt" or len(str(x.get("provenance_digest","")))!=64:raise Closed("terminal receipt invalid")

def replace(original:bytes,restore:bool=False)->str:
 d=os.open(RECEIPT.parent,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW);lf=os.open(LOCK.name,os.O_RDWR|os.O_NOFOLLOW|os.O_NONBLOCK,dir_fd=d);tmp=f".{RECEIPT.name}.incident-{TARGET}.tmp"
 try:
  fcntl.flock(lf,fcntl.LOCK_EX|fcntl.LOCK_NB);current=read_safe(RECEIPT)
  if restore: payload=original
  else:
   v=parse_receipt(current);validate_target(v);del v["terminal_failures"][str(TARGET)];del v["terminal_failures"];v["version"]=2 if v["terminal_receipts"] else 1
   if v["version"]==1:del v["terminal_receipts"]
   payload=json.dumps(v,indent=2,sort_keys=True).encode()
  fd=os.open(tmp,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600,dir_fd=d)
  try:os.write(fd,payload);os.fsync(fd)
  finally:os.close(fd)
  os.replace(tmp,RECEIPT.name,src_dir_fd=d,dst_dir_fd=d);os.fsync(d);return sha(payload)
 finally:
  try:os.unlink(tmp,dir_fd=d)
  except FileNotFoundError:pass
  fcntl.flock(lf,fcntl.LOCK_UN);os.close(lf);os.close(d)

def static_gates()->dict[str,object]:
 direct=SITE/"hermes_agent-0.17.0.dist-info/direct_url.json"
 if fsha(direct)!=DIRECT_URL_HASH or load(direct).get("archive_info",{}).get("hashes",{}).get("sha256")!=WHEEL:raise Closed("installed wheel mismatch")
 if fsha(SITE/"gateway/platforms/telegram.py")!=TELEGRAM_MODULE:raise Closed("Telegram module mismatch")
 for p,h in EDITABLE.items():
  if fsha(SITE/p)!=h:raise Closed("editable mapping drift")
 if str(PROFILE/"workspace/checkin_cli/checkin_cli") not in (SITE/"__editable___physique_checkin_cli_0_1_0_finder.py").read_text():raise Closed("editable target mismatch")
 ready=load(OBSERVER/"live-v62-ready.json");b=ready.get("bindings",{})
 expected={"successor":CANDIDATE,"wheel":WHEEL,"profile_wheel":PROFILE_WHEEL,"authority_digest":AUTHORITY_DIGEST,"deployment_receipt":DEPLOYMENT_RECEIPT}
 if ready.get("status")!="READY_CONTINUOUS_LIFECYCLE" or ready.get("authority_sha256")!=OBSERVER_AUTHORITY or any(b.get(k)!=v for k,v in expected.items()):raise Closed("observer authority mismatch")
 if fsha(CHECKLIST)!="b093aeb888ce7325cddd7d56d52a7e4a7ec8a180ac80eef21fea9453a8ef0e8e" or load(CHECKLIST).get("lifecycle_bindings",{}).get("authority_digest")!=AUTHORITY_DIGEST:raise Closed("activation authority mismatch")
 if fsha(CONTINUATION)!=CONTINUATION_HASH:raise Closed("continuation replay evidence mismatch")
 prior=CONTINUATION.read_text(errors="strict").splitlines();needle=f"telegram_ingress stage=receipt update_id={TARGET}"
 if sum(needle in x and "reason_code=" not in x for x in prior)!=1:raise Closed("continuation replay evidence ambiguous")
 needle=f"{OBSERVER/'observer_v62.py'} observe --profile {PROFILE}"
 if sum(needle in x for x in run("ps","-eo","args=").stdout.splitlines())!=1:raise Closed("observer v6.2 not uniquely armed")
 return {"wheel":WHEEL,"candidate":CANDIDATE,"authority_digest":AUTHORITY_DIGEST,"observer_authority":OBSERVER_AUTHORITY}

def authority()->dict[str,object]:
 r=load(PROFILE/"customers/registry.json");m=[x for x in r.get("customers",[]) if x.get("customer_key")==CUSTOMER]
 if len(m)!=1 or m[0].get("enabled") is not True or m[0].get("telegram")!={"user_id":"8527916639","chat_id":"8527916639","topic_id":"0"}:raise Closed("customer/DM authority mismatch")
 b=load(PROFILE/"data/onboarding/telegram-customer-bootstrap-v1/ledger.json");a=[x for x in b.get("sessions",[]) if x.get("state")=="ACTIVE"]
 if len(a)!=1 or a[0].get("generation")!=6 or a[0].get("customer_draft",{}).get("customer_key")!=CUSTOMER:raise Closed("ACTIVE bootstrap mismatch")
 import yaml
 ad=yaml.safe_load((PROFILE/"config.yaml").read_text())["platforms"]["telegram"]["extra"]["adaptive_nutrition"]
 if any(ad.get(k) is not False for k in ("activation","delivery","delivery_enabled")):raise Closed("adaptive delivery/activation enabled")
 delivery=[PROFILE/"data/owner-actions/draft-deliveries.json",PROFILE/"data/scheduled-deliveries.jsonl"]
 if any(x.exists() for x in delivery):raise Closed("delivery count is not zero")
 return {"customer_enabled":True,"bootstrap":"generation6/ACTIVE","adaptive_activation":False,"adaptive_delivery":False,"delivery_count":0,"dm_topic":0}

def inventory()->dict[str,dict[str,object]]:
 roots=[PROFILE/"customers",PROFILE/"data/customers"/CUSTOMER,PROFILE/"data/onboarding/telegram-customer-bootstrap-v1",PROFILE/"data/onboarding/telegram-publication-outbox-v1",PROFILE/"data/owner-actions"]
 files={PROFILE/"config.yaml",PROFILE/"data/customer-activation-journal.json",PROFILE/"data/customer-activation-audit.jsonl",PROFILE/"data/nutrition-onboarding-projection-journal.jsonl"}
 for root in roots:
  if root.exists():files.update(p for p in root.rglob("*") if p.is_file() and not p.name.endswith(".lock"))
 files.discard(RECEIPT);return {str(p.relative_to(PROFILE)):{"sha256":fsha(p),"bytes":p.stat().st_size} for p in sorted(files) if p.exists()}

def cli_proof(root:Path)->None:
 h=run(str(PYTHON),"-m","hermes_cli.main","--profile","dualcoachtest","gateway","--help").stdout;s=(HERMES/"hermes_cli/subcommands/gateway.py").read_text();commands=["run","start","stop","restart","status","install","uninstall","list","setup","migrate-legacy","enroll"]
 if not all(x in h for x in commands) or any(x in (h+s).lower() for x in ("handler_exception","telegram-ingress","poll replay")):raise Closed("CLI surface proof failed")
 emit(root/"cli-recovery-surface.json",{"schema":"task26-cli-recovery-surface-v1","supported_gateway_commands":commands,"non_committed_handler_exception_recovery":False,"help_sha256":sha(h.encode()),"parser_sha256":sha(s.encode())})

def watchers(root:Path):
 cursor=run("journalctl","--user","-u",UNIT,"-n","0","--show-cursor","--no-pager").stdout.split("-- cursor: ")[-1].strip();jf=os.open(root/"journal-watch.log",os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600);af=os.open(root/"application-watch.log",os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
 j=subprocess.Popen(["journalctl","--user","-u",UNIT,"--after-cursor",cursor,"-f","-o","short-iso"],stdout=jf,stderr=subprocess.STDOUT);a=subprocess.Popen(["tail","-n","0","-F",str(PROFILE/"logs/gateway.log")],stdout=af,stderr=subprocess.STDOUT);os.close(jf);os.close(af);time.sleep(.2)
 if j.poll() is not None or a.poll() is not None:raise Closed("watcher arm failed")
 emit(root/"watchers-armed.json",{"schema":"task26-watchers-armed-v1","journal_pid":j.pid,"application_pid":a.pid,"cursor_sha256":sha(cursor.encode()),"armed_before_restart":True,"armed_at_local":datetime.datetime.now().isoformat()});return j,a
def kill(ws)->None:
 for p in ws:
  if p.poll() is None:p.terminate()
 for p in ws:
  try:p.wait(timeout=3)
  except subprocess.TimeoutExpired:p.kill();p.wait(timeout=3)
def wait(root:Path,timeout:float)->dict[str,object]:
 end=time.monotonic()+timeout;needle=f"telegram_ingress stage=receipt update_id={TARGET}";bad=f"telegram_handler_exception update_id={TARGET}";armed=datetime.datetime.fromisoformat(load(root/"watchers-armed.json")["armed_at_local"]);healthy_since=None
 while time.monotonic()<end:
  text=(root/"journal-watch.log").read_text(errors="replace")+"\n"+(root/"application-watch.log").read_text(errors="replace");lines=[]
  for x in text.splitlines():
   try:stamp=datetime.datetime.strptime(x[:23],"%Y-%m-%d %H:%M:%S,%f")
   except ValueError:continue
   if stamp>=armed.replace(tzinfo=None):lines.append(x)
  normal=[x for x in lines if needle in x and "reason_code=" not in x];fail=[x for x in lines if bad in x or (needle in x and "reason_code=handler_exception" in x)];v=parse_receipt(read_safe(RECEIPT));s=fields();g=load(PROFILE/"gateway_state.json");absent=all(str(TARGET) not in v.get(k,{}) for k in ("receipts","terminal_receipts","terminal_failures"));healthy=s.get("ActiveState")=="active" and s.get("SubState")=="running" and int(s.get("MainPID","0"))>1 and g.get("gateway_state")=="running" and g.get("platforms",{}).get("telegram",{}).get("state")=="connected"
  healthy_since=healthy_since or (time.monotonic() if healthy else None)
  if not normal and not fail and absent and healthy and healthy_since is not None and time.monotonic()-healthy_since>=15:return {"normal_handled_log_count":1,"handler_exception_count":0,"offset_lower_bound":TARGET+1,"offset_proof":"pinned prior successful handling plus no redelivery during a fresh healthy polling cycle proves Telegram accepted offset > target","continuation_log_sha256":CONTINUATION_HASH,"service":s,"telegram":"connected"}
  if normal or fail:raise Closed("target redelivered during continuation")
  time.sleep(.1)
 raise Closed("continuation offset/health timeout")
def seal(root:Path)->str:
 rows=[{"path":str(p.relative_to(root)),"sha256":fsha(p),"bytes":p.stat().st_size} for p in sorted(root.rglob("*")) if p.is_file() and p.name not in ("SEAL.json","SEAL.json.sha256")];emit(root/"SEAL.json",{"schema":"task26-telegram-poll-recovery-seal-v1","candidate":CANDIDATE,"target_update_id":TARGET,"files":rows});h=fsha(root/"SEAL.json");create(root/"SEAL.json.sha256",f"{h}  SEAL.json\n".encode());return h

def execute(root:Path,timeout:float)->int:
 root.mkdir(mode=0o700,parents=True,exist_ok=True)
 if stat.S_IMODE(root.stat().st_mode)!=0o700:raise Closed("evidence root mode")
 cli_proof(root);static=static_gates();initial=fields()
 if not ((initial.get("ActiveState")=="active" and initial.get("SubState")=="running") or (initial.get("ActiveState")!="active" and initial.get("MainPID")=="0")):raise Closed("initial service state unsafe")
 auth=authority();original=read_safe(RECEIPT);validate_target(parse_receipt(original));pre=inventory();emit(root/"pre-inventory.json",pre);emit(root/"preconditions.json",{"schema":"task26-recovery-preconditions-v1","static":static,"authority":auth,"service":initial,"receipt":{"path":str(RECEIPT),"sha256":sha(original),"bytes":len(original),"version":3,"target_reason":"handler_exception"}})
 changed=False;ws=();stopped=False
 try:
  stop();stopped=True
  if authority()!=auth or inventory()!=pre or read_safe(RECEIPT)!=original:raise Closed("drift while stopping")
  create(root/"original-receipt.byte-exact.json",original)
  removed=replace(original);changed=True;v=parse_receipt(read_safe(RECEIPT))
  if any(str(TARGET) in v.get(k,{}) for k in ("receipts","terminal_receipts","terminal_failures")):raise Closed("bounded removal failed")
  ws=watchers(root);run("systemctl","--user","start",UNIT);replay=wait(root,timeout);time.sleep(.3);kill(ws);ws=()
  post=inventory();emit(root/"post-inventory.json",post)
  if post!=pre or authority()!=auth:raise Closed("business authority mutated")
  postraw=read_safe(RECEIPT);postv=parse_receipt(postraw)
  if any(str(TARGET) in postv.get(k,{}) for k in ("receipts","terminal_receipts","terminal_failures")):raise Closed("target remains in receipt")
  result={"schema":"task26-telegram-poll-recovery-result-v1","status":"PASS_RECOVERED","target_update_id":TARGET,"candidate":CANDIDATE,"receipt":{"pre_sha256":sha(original),"removed_sha256":removed,"post_sha256":sha(postraw),"post_version":postv["version"]},"replay":replay,"authority":auth,"business_state_unchanged":True,"safe_fallback":{"application_handlings":1,"maximum_customer_sends":1,"bot_api_message_id":None,"limitation":"Bot API has no outgoing-history endpoint; privacy-safe application logs omit response content and message ID"},"user_may_send_today_checkin":True};emit(root/"recovery-result.json",result);seald=seal(root);print(json.dumps({"status":"PASS_RECOVERED","evidence_root":str(root),"seal":seald,"pre_receipt_sha256":sha(original),"post_receipt_sha256":sha(postraw),"offset_lower_bound":TARGET+1,"service":replay["service"],"user_may_send_today_checkin":True},sort_keys=True));return 0
 except BaseException as e:
  kill(ws)
  if changed:
   try:
    stop();stopped=True;cur=parse_receipt(read_safe(RECEIPT))
    if not any(str(TARGET) in cur.get(k,{}) for k in ("receipts","terminal_receipts","terminal_failures")):replace(original,True)
   except BaseException:pass
  try:emit(root/"failure-result.json",{"schema":"task26-recovery-failure-v1","status":"FAIL_CLOSED","error_class":type(e).__name__,"error":str(e),"service_left_stopped":stopped,"original_receipt_restored":RECEIPT.exists() and read_safe(RECEIPT)==original})
  except BaseException:pass
  raise

def main()->int:
 p=argparse.ArgumentParser();p.add_argument("command",choices=("execute",));p.add_argument("--evidence-root",type=Path,required=True);p.add_argument("--timeout",type=float,default=90);a=p.parse_args()
 try:return execute(a.evidence_root.resolve(),a.timeout)
 except Closed as e:print(json.dumps({"status":"FAIL_CLOSED","reason":str(e)},sort_keys=True),file=sys.stderr);return 2
if __name__=="__main__":raise SystemExit(main())