"""Original event-driven TaskGroup/gather failure and cleanup, no sleeps or external effects."""
import asyncio,json,sys

def leaves(err):
 if isinstance(err,BaseExceptionGroup):return sorted(x for child in err.exceptions for x in leaves(child))
 return [type(err).__name__]

async def worker(started,release,events,cleanup_failure=False):
 events.append('worker_started');started.set()
 try:await release.wait();events.append('worker_released');return 7
 finally:
  events.append('worker_cleanup')
  if cleanup_failure:raise RuntimeError('authored cleanup failure')

async def failer(started,events):
 await started.wait();events.append('child_raises_ValueError');raise ValueError('authored child failure')

async def group_failure(cleanup_failure=False):
 started=asyncio.Event();release=asyncio.Event();events=[];caught=[]
 try:
  async with asyncio.TaskGroup() as tg:
   w=tg.create_task(worker(started,release,events,cleanup_failure));f=tg.create_task(failer(started,events))
 except* Exception as err:caught=leaves(err);events.append('group_exception_caught_after_exit')
 assert w.done() and f.done() and 'worker_cleanup' in events and events.index('worker_cleanup')<events.index('group_exception_caught_after_exit')
 assert caught==(['RuntimeError','ValueError'] if cleanup_failure else ['ValueError'])
 assert w.cancelled() is (not cleanup_failure)
 return {'events':events,'exception_leaf_types':caught,'worker_done_after_exit':w.done(),'worker_cancelled':w.cancelled(),'all_children_done':w.done() and f.done()}

async def gather_failure():
 started=asyncio.Event();release=asyncio.Event();events=[];w=asyncio.create_task(worker(started,release,events));f=asyncio.create_task(failer(started,events))
 try:await asyncio.gather(w,f)
 except ValueError:
  events.append('gather_ValueError_caught');at_catch={'worker_done':w.done(),'worker_cancelled':w.cancelled(),'cleanup_happened':'worker_cleanup' in events};assert at_catch=={'worker_done':False,'worker_cancelled':False,'cleanup_happened':False}
 finally:
  if not w.done():w.cancel()
  try:await w
  except asyncio.CancelledError:events.append('harness_awaited_worker_cancellation')
  await asyncio.gather(f,return_exceptions=True)
 assert w.done() and f.done() and events.index('gather_ValueError_caught')<events.index('worker_cleanup')
 return {'events':events,'at_catch':at_catch,'all_children_done_after_explicit_harness_cleanup':True}

async def success_and_child_cancel():
 async def success(n):return n*n
 async with asyncio.TaskGroup() as tg:a=tg.create_task(success(2));b=tg.create_task(success(3))
 assert [a.result(),b.result()]==[4,9]
 started=asyncio.Event();release=asyncio.Event();events=[]
 async def cancel_self():raise asyncio.CancelledError()
 async with asyncio.TaskGroup() as tg:
  c=tg.create_task(cancel_self());d=tg.create_task(success(5))
 assert c.cancelled() and d.result()==25
 return {'successful_results':[a.result(),b.result()],'cancelled_child_is_cancelled':c.cancelled(),'sibling_success_not_cancelled':d.result()}

async def suite():
 before=set(asyncio.all_tasks());group=await group_failure();gather=await gather_failure();cleanup=await group_failure(True);success=await success_and_child_cancel();current=asyncio.current_task();new_live=[t for t in asyncio.all_tasks() if t not in before and t is not current and not t.done()];assert not new_live
 return {'TaskGroup_failure':group,'gather_default_failure':gather,'TaskGroup_cleanup_also_fails':cleanup,'success_and_cancelled_child':success,'new_live_tasks_left_after_harness':0,'event_controlled_no_sleep':True,'runtime':'.'.join(map(str,sys.version_info[:3])),'scope':'cooperative in-memory tasks only; CancelledError propagated in worker. No swallow-cancellation/CPU preemption/threads/timeout/network/external side-effect rollback or transactional guarantee. gather cancellation itself differs from sibling child failure.'}

def demo():
 first=asyncio.run(suite());second=asyncio.run(suite());assert first==second;return first
if __name__=='__main__':print(json.dumps(demo(),ensure_ascii=True,indent=2))