"""Logged, fresh-context provider invocations. No invocation occurs on import."""
from __future__ import annotations
import asyncio
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import time
from datetime import datetime, timezone
from uuid import uuid4

ROOT=Path(__file__).resolve().parent
PROFILES={
 'astra':{'provider':'codex','model':'gpt-6-astra','effort':'medium','family':'openai'},
 'mini':{'provider':'codex','model':'gpt-5.4-mini','effort':'medium','family':'openai'},
 'fable':{'provider':'claude','model':'claude-fable-5-1','effort':'medium','family':'anthropic'},
 'haiku':{'provider':'claude','model':'claude-haiku-4-5','effort':'medium','family':'anthropic'},
 'pro':{'provider':'agy','model':'gemini-3.1-pro-high','effort':'high','family':'google'},
 'flash':{'provider':'agy','model':'gemini-3.8-flash-low','effort':'low','family':'google'},
}

def utc(): return datetime.now(timezone.utc).isoformat()
def dump(path,data):
 path=Path(path); path.parent.mkdir(parents=True,exist_ok=True)
 path.write_text(json.dumps(data,ensure_ascii=False,indent=2,allow_nan=False),encoding='utf-8')
def sha(data): return hashlib.sha256(data if isinstance(data,bytes) else data.encode()).hexdigest()

def parse_json(text):
 def unique(pairs):
  d={}
  for k,v in pairs:
   if k in d: raise ValueError('duplicate JSON key: '+k)
   d[k]=v
  return d
 def invalid(c): raise ValueError('invalid JSON constant: '+c)
 text=text.strip()
 match=re.fullmatch(r'```(?:json)?[ \t]*\r?\n([\s\S]*?)\r?\n```',text,re.I)
 if match: text=match[1]
 return json.loads(text,object_pairs_hook=unique,parse_constant=invalid)

def command(profile,folder,prompt):
 cfg=PROFILES[profile]; p=cfg['provider']; effort=cfg['effort']
 common=(ROOT/'agent-instructions.txt').read_text(encoding='utf-8')
 if p=='codex':
  args=[shutil.which('codex'),'exec','--model',cfg['model'],'--ignore-user-config','--ephemeral','--json','--skip-git-repo-check','--sandbox','read-only','--cd',str(folder/'workspace'),'--output-last-message',str(folder/'answer.txt')]
  settings=[f'model_reasoning_effort="{effort}"','features.shell_tool=false','features.shell_snapshot=false','features.multi_agent=false','agents.enabled=false','features.apps=false','features.plugins=false','features.remote_plugin=false','features.hooks=false','features.memories=false','features.skill_mcp_dependency_install=false','web_search="disabled"','project_doc_max_bytes=0',f'model_instructions_file={json.dumps(str(ROOT/"agent-instructions.txt"))}']
  for c in settings: args+=['-c',c]
  return args+['-']
 if p=='claude':
  return [shutil.which('claude'),'--print','--model',cfg['model'],'--output-format','json','--safe-mode','--tools','','--strict-mcp-config','--mcp-config','{"mcpServers":{}}','--permission-prompts','none','--no-chrome','--no-session-persistence','--effort',effort,'--system-prompt',common]
 return [shutil.which('agy'),'--print',prompt,'--model',cfg['model'],'--effort',effort,'--mode','plan','--output-format','stream-json','--print-timeout','180s','--log-file',str(folder/'internal-agy.log')]

def invoke(profile,prompt,folder,timeout=210):
 folder=Path(folder); folder.mkdir(parents=True,exist_ok=False); (folder/'workspace').mkdir()
 cfg=PROFILES[profile]; args=command(profile,folder,prompt)
 (folder/'input.txt').write_text(prompt,encoding='utf-8')
 meta={'id':folder.name,'profile':profile,**cfg,'started_at':utc(),'prompt_sha256':sha(prompt),'command':args,'timeout_seconds':timeout,'tool_policy':'none requested; Codex and Claude capabilities disabled; Gemini plan-mode and no-tool instruction, actual tool events audited','fresh_context':True}
 dump(folder/'started.json',meta)
 env=os.environ.copy(); env['PYTHONIOENCODING']='utf-8'
 start=time.monotonic(); failure=None; answer=''; raw=''; stderr=''; code=None
 p=subprocess.Popen(args,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,cwd=folder/'workspace',env=env,encoding='utf-8',errors='replace',creationflags=subprocess.CREATE_NO_WINDOW if os.name=='nt' else 0)
 try:
  raw,stderr=p.communicate(input='' if cfg['provider']=='agy' else prompt,timeout=timeout); code=p.returncode
 except subprocess.TimeoutExpired:
  if os.name=='nt': subprocess.run(['taskkill','/PID',str(p.pid),'/T','/F'],capture_output=True,creationflags=subprocess.CREATE_NO_WINDOW)
  else: p.kill()
  raw,stderr=p.communicate(); code=p.returncode; failure='infrastructure_timeout'
 (folder/'stdout.txt').write_text(raw,encoding='utf-8'); (folder/'stderr.txt').write_text(stderr,encoding='utf-8')
 events=[]
 for line in raw.splitlines():
  try: events.append(json.loads(line))
  except ValueError: pass
 meta['elapsed_seconds']=time.monotonic()-start; meta['exit_code']=code
 tool_events=[]
 if cfg['provider']=='codex':
  answer=(folder/'answer.txt').read_text(encoding='utf-8') if (folder/'answer.txt').exists() else ''
  meta['session_id']=next((e.get('thread_id') for e in events if e.get('type')=='thread.started'),None)
  meta['usage']=next((e.get('usage') for e in events if e.get('type')=='turn.completed'),{})
  tool_events=[e for e in events if e.get('item',{}).get('type') in ['command_execution','mcp_tool_call','web_search','file_change']]
  meta['model_identity_evidence']='Requested CLI identifier; raw events do not independently attest backend identity.'
 elif cfg['provider']=='claude':
  try:
   obj=json.loads(raw); answer=obj.get('result',''); meta['usage']=obj.get('usage',{}); meta['model_usage']=obj.get('modelUsage',{}); meta['session_id']=obj.get('session_id'); meta['reported_cost_usd']=obj.get('total_cost_usd'); meta['num_turns']=obj.get('num_turns')
   if obj.get('is_error'): failure=failure or 'provider_error'
  except ValueError: failure=failure or 'provider_envelope_invalid'
  meta['model_identity_evidence']='CLI identifier plus provider modelUsage, including auxiliary models.'
 else:
  results=[e.get('result',{}) if e.get('event')=='result' else e for e in events if e.get('event')=='result' or e.get('type')=='result' or e.get('status') in ['SUCCESS','ERROR']]
  obj=results[-1] if results else {}
  answer=obj.get('response',obj.get('result',''))
  if not isinstance(answer,str): answer=json.dumps(answer,ensure_ascii=False)
  if not answer:
   chunks=[]
   for e in events:
    if e.get('type')=='assistant':
     msg=e.get('message',{})
     if isinstance(msg,dict): chunks += [c.get('text','') for c in msg.get('content',[]) if c.get('type')=='text']
   answer=''.join(chunks)
  meta['provider_result']=obj; meta['usage']=obj.get('usage',obj.get('stats',{})); meta['session_id']=obj.get('conversation_id',obj.get('session_id'))
  meta['provider_init']=next((e.get('init') for e in events if e.get('event')=='init'),{})
  if obj.get('status')!='SUCCESS': failure=failure or 'provider_status_not_success'
  tool_events=[e for e in events if e.get('type') in ['tool_use','tool_result'] or any(c.get('type')=='tool_use' for c in e.get('message',{}).get('content',[]) if isinstance(c,dict))]
  tool_events += [e for e in events if e.get('event')=='step_update' and e.get('step_update',{}).get('step_type') not in ['user_input','agent_response','checkpoint','error','conversation_checkpoint']]
  meta['model_identity_evidence']='Requested Antigravity CLI identifier; inspect stream metadata separately; no independent attestation.'
 if tool_events: failure=failure or 'unexpected_tool_use'
 meta['tool_event_count']=len(tool_events); meta['event_types']=[e.get('type',e.get('event')) for e in events]
 if code!=0 or not answer: failure=failure or 'provider_error_or_empty'
 meta['status']='failed' if failure else 'succeeded'; meta['failure']=failure; meta['ended_at']=utc(); meta['answer_sha256']=sha(answer)
 (folder/'answer.txt').write_text(answer,encoding='utf-8'); dump(folder/'metadata.json',meta)
 if failure: raise RuntimeError(f'{profile}: {failure} ({folder.name})')
 return answer,meta

class Pool:
 def __init__(self,concurrency_per_provider=2): self.limits={k:asyncio.Semaphore(concurrency_per_provider) for k in ['codex','claude','agy']}
 async def call(self,profile,prompt,directory):
  async with self.limits[PROFILES[profile]['provider']]:
   return await asyncio.to_thread(invoke,profile,prompt,Path(directory)/str(uuid4()))

if __name__=='__main__':
 import sys
 key=sys.argv[1]; print(invoke(key,'Reply exactly {"ready":true}. Do not use tools. Engineering test only.',ROOT/'private'/'engineering-clean'/(key+'-'+str(uuid4())[:8]))[0])
