#!/usr/bin/env python3
"""Reproduce a resume-path admission gap using unchanged upstream handler functions.
Controlled stand-ins replace HTTP, DB lookup, and inference. No live service is called.
"""
from pathlib import Path
import ast,asyncio,contextvars,functools,hashlib,json,os,time,types,uuid
from contextlib import suppress
R=Path(__file__).resolve().parent
SRC=Path(os.environ.get('HERMES_SOURCE_DIR',str(R/'hermes-source')))
EXPECTED={'gateway/platforms/api_server.py':'6a7e6fb0a0d5ea25ddc47d36a64e2c0bc50ded331d49e1c4521d801ca7189043','gateway/platforms/api_server_runs.py':'8a7fcfd56183c32a14b41c7c43ce36b21b37bf49f2ed035dafe48899c0e8cbe7'}
for name,digest in EXPECTED.items():
 assert hashlib.sha256((SRC/name).read_bytes()).hexdigest()==digest,'Requires pinned Hermes d923a5eb487a17c0a68801bba9ef0f56c81040a8'
scope={'wraps':functools.wraps,'suppress':suppress,'time':time,'json':json,'hashlib':hashlib,'uuid':uuid,'asyncio':asyncio}
def extract(path,names):
 tree=ast.parse((SRC/path).read_text());nodes=[]
 for node in ast.walk(tree):
  if isinstance(node,(ast.FunctionDef,ast.AsyncFunctionDef)) and node.name in names:nodes.append(node)
 assert {n.name for n in nodes}==set(names)
 mod=ast.Module(body=[ast.ImportFrom(module='__future__',names=[ast.alias(name='annotations')],level=0)]+nodes,type_ignores=[])
 exec(compile(ast.fix_missing_locations(mod),path,'exec'),scope)
extract('gateway/platforms/api_server.py',['_admit_api_agent_request','_release_pending_api_work','_activate_admitted_request','active_agent_work_count','_concurrency_limited_response'])
extract('gateway/platforms/api_server_runs.py',['_handle_runs'])
reservation=contextvars.ContextVar('reservation',default=None)
scope['_api_agent_request_reservation']=reservation
scope['_api_runs']=types.SimpleNamespace(_uses_room_run_auth=lambda *a:False)
scope['_error_response']=lambda message,status,**kw:{'status':status,**kw}
scope['_resolve_conversation_history']=lambda *a,**kw:([],None,None,None)
scope['_RunStream']=lambda:object()
scope['_RunLaunch']=lambda *a,**kw:types.SimpleNamespace(run_id=a[1])
scope['_accepted_response']=lambda run_id,status,*a,**kw:{'status':202,'run_id':run_id}
async def resolve_session(self,session):return session
scope['_resolve_live_session_id']=resolve_session
async def experiment(existing_session):
 gate=asyncio.Event();reached=asyncio.Event();workers_gate=asyncio.Event();calls=0
 async def execute(self,run,**kw):await workers_gate.wait()
 scope['_execute_run']=execute
 api=types.SimpleNamespace(_openai_error=lambda *a,**kw:None,_request_turn_author=lambda *a:None,_request_agent_overrides=lambda *a,**kw:{})
 for name in ['_api_request_profile','_api_request_browser_control_principal','_api_request_browser_control_transport_family']:
  setattr(api,name,contextvars.ContextVar(name,default=None))
 class Adapter:
  _max_concurrent_runs=1;_inflight_agent_runs=0;_pending_agent_requests=0;_model_name='synthetic'
  active_agent_work_count=scope['active_agent_work_count']
  _concurrency_limited_response=scope['_concurrency_limited_response']
  _activate_admitted_request=scope['_activate_admitted_request']
  def __init__(self):
   for name in ['_active_run_tasks','_run_owners','_run_streams','_run_streams_created','_run_approval_sessions','_run_statuses']:setattr(self,name,{})
   self._background_tasks=set()
  def _check_auth(self,*a):return None
  def _draining_response(self):return None
  def _parse_session_key_header(self,*a):return None,None
  async def _normalize_room_dispatch(self,request,body):return body,None
  def _resolve_route(self,*a):return None
  def _request_route_conflict_error(self,**kw):return None
  def _run_idempotency_scope(self,*a):return 'synthetic-user'
  def _set_run_status(self,run,status,**kw):return {'status':status,**kw}
  async def _conversation_history_for_session(self,*a):return []
  async def _admit_to_live_bot_chat(self,*a):
   nonlocal calls
   calls+=1;reached.set();await gate.wait();return None
 a=Adapter()
 class Request:
  headers={}
  def __init__(self,session):self.session=session
  async def json(self):return {'input':'synthetic','session_id':self.session} if existing_session else {'input':'synthetic'}
 handler=scope['_admit_api_agent_request'](scope['_handle_runs'])
 first=asyncio.create_task(handler(a,Request('existing-one'),_api_server=api))
 if existing_session:await reached.wait()
 else:await first
 count_at_pause=a.active_agent_work_count()
 second=asyncio.create_task(handler(a,Request('existing-two'),_api_server=api))
 # Allow the second handler to reach either refusal or the same awaited owner lookup.
 for _ in range(3):await asyncio.sleep(0)
 gate.set();responses=await asyncio.gather(first,second)
 active_after_admission=a.active_agent_work_count()
 workers_gate.set();await asyncio.gather(*a._active_run_tasks.values())
 return {'existing_session':existing_session,'configured_cap':1,'count_while_first_owner_lookup_paused':count_at_pause,'response_statuses':[r['status'] for r in responses],'active_runs_after_admission':active_after_admission}
async def main():
 results=[await experiment(False),await experiment(True)]
 assert results[0]['response_statuses']==[202,429] and results[0]['active_runs_after_admission']==1
 assert results[1]['response_statuses']==[202,202] and results[1]['active_runs_after_admission']==2
 out={'source_commit':'d923a5eb487a17c0a68801bba9ef0f56c81040a8','kind':'unchanged source-handler concurrency reproduction with controlled HTTP/DB/worker stand-ins; not a live HTTP test','results':results}
 (R/'api-admission-gap.json').write_text(json.dumps(out,indent=2)+'\n');print(json.dumps(out,indent=2))
asyncio.run(main())