Blame for test_lifecycle.py

import json
"""دورة الإشراف الكاملة: Source stall → watchdog → cleanup → monitor → LIVE → جلسة جديدة.

HTTP محلي حقيقي + FFmpeg حقيقي (لا stub للعملية): الحالة المطلوب إثباتها هي «العملية حيّة
وتدفّقها مات»، ولا تُثبت بمحاكاة. الصمت الخالص (لا بيانات ولا إغلاق) مقيس بدقة التوقيت في
test_watchdog.py؛ هنا نقيس الدورة الكاملة، والخادم يُغلق بعد الصمت كما يفعل CDN حقيقي حتى
لا يُعلّق خادم الاختبار نفسه في write على مقبس ممتلئ.
"""
import http.server, os, pathlib, socketserver, subprocess, sys, tempfile, threading, time

DIR = pathlib.Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else pathlib.Path('/home/user/audit')
os.environ.update(DATA_DIR=str(tempfile.mkdtemp()), PASSWORD='admin', ENSURE_FREE_MB='1', MIN_FREE_MB='1')
sys.path.insert(0, str(DIR))
from shutil import which
FF = which(os.environ.get('FFMPEG_PATH') or 'ffmpeg') or '/tmp/fftools/ffmpeg-7.0.2-amd64-static/ffmpeg'
os.environ.update(YOUTUBE_SEND_MODE='copy', YOUTUBE_SOURCE_STALL_SECONDS='3', YOUTUBE_SOURCE_READ_TIMEOUT='1',
                  YOUTUBE_PROGRESS_STALL_SECONDS='25', YOUTUBE_WATCHDOG_REPORT_SECONDS='60', FFMPEG_PATH=FF)
import relay
from relay import Sender, SourceStalled

tmp = pathlib.Path(tempfile.mkdtemp())
_stub = tmp / 'mkvmerge'            # mkvmerge غير مطلوب هنا؛ صورة السبيس فيه أصلًا
_stub.write_text('#!/bin/sh\nexit 0\n')
_stub.chmod(0o755)
os.environ['MKVMERGE_PATH'] = str(_stub)
os.environ['FFPROBE_PATH'] = which(os.environ.get('FFPROBE_PATH') or 'ffprobe') or \
    '/tmp/fftools/ffmpeg-7.0.2-amd64-static/ffprobe'

SRC = tmp / 'src.flv'
subprocess.run([FF, '-hide_banner', '-loglevel', 'error', '-y', '-f', 'lavfi',
                '-i', 'testsrc2=size=320x240:rate=20', '-f', 'lavfi',
                '-i', 'sine=frequency=440:sample_rate=48000', '-t', '120',
                '-map', '0:v', '-map', '1:a', '-c:v', 'libx264', '-preset', 'ultrafast',
                '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-ar', '48000', '-ac', '1',
                '-f', 'flv', str(SRC)], check=True, capture_output=True)
DATA = SRC.read_bytes()

state = {'mode': 'flow'}            # flow | stall | stall_close | dead


class Handler(http.server.BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'

    def log_message(self, *a):
        pass

    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-Type', 'video/flv')
        self.end_headers()
        step, pos, served = 8192, 0, 0
        try:
            while True:
                if state['mode'] == 'dead':
                    return
                if state['mode'] == 'stall' and served < 200:
                    time.sleep(0.02)
                    continue
                if state['mode'] == 'stall':
                    while state['mode'] == 'stall':      # صمت تام: لا بيانات ولا إغلاق
                        time.sleep(0.1)
                    continue
                if state['mode'] == 'stall_close' and served >= 200:
                    return                              # صمت قصير ثم إغلاق (فعل CDN)
                self.wfile.write(DATA[pos:pos + step])
                self.wfile.flush()
                time.sleep(float(os.environ.get('TEST_CHUNK_DELAY', '0.012')))
                served += 1
                pos = (pos + step) % max(step, len(DATA) - step)
        except (BrokenPipeError, ConnectionResetError, OSError):
            pass


class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True
    daemon_threads = True
    hits = []

    def handle_error(self, *a):
        pass


srv = Server(('127.0.0.1', 0), Handler)
PORT = srv.server_address[1]
threading.Thread(target=srv.serve_forever, daemon=True).start()
URL = 'http://127.0.0.1:%d/stage/stream.flv' % PORT
CAND = {'url': URL, 'media': 'flv', 'codec': 'h264', 'label': 'hd · h264'}
logs = []
results = []


def check(tag, cond, extra=''):
    results.append((tag, bool(cond)))
    print('   %s %s %s' % ('OK —' if cond else 'FAIL', tag, extra))


def new_sender():
    s = Sender(FF, lambda u, m: logs.append(str(m)), lambda u: {})
    s.mode_override = 'copy'
    return s


# ── H) انقطاع مصدر HTTP حقيقي: الحارس ينهي الجلسة نظيفة (لا عملية ولا جلسة معلقة) ──
sender = new_sender()
out = tmp / 'H.flv'
box = {'rc': None, 'exc': None}
st_h = threading.Event()


def _h():
    try:
        box['rc'] = sender.send('u', CAND, str(out), st_h)
    except BaseException as e:
        box['exc'] = e


t0 = time.monotonic()
t_h = threading.Thread(target=_h, daemon=True)
t_h.start()
seen = 0
while time.monotonic() - t0 < 25 and sender.snapshot()['source_bytes'] < 300 * 1024:
    time.sleep(0.1)
    seen = sender.snapshot()['source_bytes']
state['mode'] = 'stall_close'
t_h.join(timeout=25)
dur = time.monotonic() - t0
snap = sender.snapshot()
check('H: تدفق حقيقي ثم انقطاع ⇒ الجلسة تُنهي بلا تعليق وبلا استثناء غامض',
      (box['exc'] is None or isinstance(box['exc'], SourceStalled)) and not t_h.is_alive(),
      'exc=%s بعد %.1fs' % (type(box['exc']).__name__ if box['exc'] else 'None', dur))
check('H: البيانات خرجت فعلًا إلى الوجهة قبل الانقطاع',
      snap['output_seconds'] > 0 and out.stat().st_size > 10000,
      'out=%.1fs size=%d' % (snap['output_seconds'], out.stat().st_size))
check('H: الحارس لم يقطع تدفقًا سليمًا (انتظر الانقطاع المصطنع)', seen > 8 * 8192, 'bytes=%d' % seen)
check('H: لا عملية ولا جلسة معلّقة بعد الإنهاء', sender.process is None and sender.session is None)
check('H: التنظيف موثّق بوسم [CLEANUP]', any('[CLEANUP]' in m for m in logs[-8:]))

# ── I) عودة المصدر: جلسة جديدة كاملة بلا تدخّل يدوي ───────────────────────────────
state['mode'] = 'flow'
sender2 = new_sender()
st2 = threading.Event()
res2 = {}
t2 = threading.Thread(target=lambda: res2.setdefault('rc', sender2.send('u', CAND, str(tmp / 'I.flv'), st2)),
                      daemon=True)
t2.start()
time.sleep(10)
mid = sender2.snapshot()
check('I: بعد عودة التدفق تعمل جلسة جديدة بلا تعديل يدوي', mid['output_seconds'] > 0,
      'out=%.1f src_age=%s' % (mid['output_seconds'], mid.get('source_age')))
st2.set()
t2.join(timeout=25)
check('I: لا جلستان للمرسل نفسه في وقت واحد', sender2.process is None and sender2.session is None)

# ── J) دورة المدير: صمت ⇒ توقف الدفع ⇒ رصد طازج ⇒ إعادة ربط ⇒ دفع جديد ───────────
os.environ['YOUTUBE_SCAN_SECONDS'] = '15'
import youtube_manager as ym
mgr = ym.ManagedYouTube(FF, lambda u, force=False: {'status': 'live', 'room_id': '7', 'sources': [CAND],
                                                    'nickname': 'n', 'live_title': 'x'},
                        lambda u, m: logs.append(str(m)), list, tmp / 'yt.json', headers_for=lambda u: {})
mgr.hold_seconds = 0.0
key = 'u7:7'
mgr.journal.change(lambda d: d['events'].update({key: {
    'session_key': key, 'username': 'u7', 'nickname': 'u7', 'room_id': '7', 'title': 't', 'description': 'd',
    'hashtags': '', 'privacy': 'public', 'marker': 'MK', 'broadcast_id': 'B', 'bound': True, 'closing': False,
    'phase': 'live', 'youtube_status': 'live', 'created_at': time.time(), 'was_live': True,
    'create_pending': False, 'note': '', 'last_poll': time.time()}}))
mgr.destinations['u7'] = 'rtmps://example.invalid/y'      # وجهة وهمية عمدًا: لا نريد اتصال يوتيوب
# لماذا لا نطلب out_time هنا: مقياس محلي — مع وجهة لا تُفتح، `-timeshift 10` كان يُقدّم
# out_time_us من مخزن fifo (8.4 ث في 8 ث) بينما `-timeshift 0` يبقى على 0.0. أي أن تقدّم
# الوقت كان يخبر عن «كم حُجز» لا عن «كم سُلّم». فالأمانة في هذه الحالة هي الصفر.
# الدليل المطلوب إذن: الدورة دارت، والجلسة الجديدة حيّة وتلتهم المصدر فعلًا.
mgr.stream_health = lambda *a, **k: None
mgr.start_sender(key)
time.sleep(3)
entry = mgr.senders.get(key)
running0 = entry['sender'].alive()
deadline = time.monotonic() + 40
while time.monotonic() < deadline and entry['sender'].alive():
    time.sleep(0.5)
stopped = not entry['sender'].alive() or not running0
attended = mgr.needs_attention(key)
state['mode'] = 'flow'
seen_force = []


def obs_force(u, force=False):
    seen_force.append(bool(force))
    return {'status': 'live', 'room_id': '7', 'sources': [CAND], 'nickname': 'n', 'live_title': 'x'}


mgr.observe = obs_force
mgr.drive_event(key)
entry2 = mgr.senders.get(key)
time.sleep(6)
_snap2 = entry2['sender'].snapshot() if entry2 else {}
out2 = float(_snap2.get('output_seconds') or 0)
src2 = float(_snap2.get('source_bytes') or 0)
check('J: صمت المصدر أوقف الدفع ثم أعاده المدير من جديد (دورة كاملة بلا تدخّل)',
      stopped and entry2 is not None and src2 > 0,
      'stop=%s بايت_المصدر=%.0f out=%.1f' % (stopped, src2, out2))
check('J: out_time لا يتقدّم حين لا شيء يُسلَّم (كان timeshift يقيس المخزن لا الوجهة)',
      out2 == 0 and src2 > 0, 'out=%.1f src=%.0f' % (out2, src2))
#needs_attention تخصّ «خيط إشراف مات»؛ هنا الخيط حيّ وأعاد بناء الجلسة بنفسه، فليس
# مطلوبًا أن يقفز المدير فوق الكاش — والأثر المطلوب هو إعادة الرصد والجلسة الجديدة.
watchdog_logged = any('[WATCHDOG]' in m for m in logs[-25:])
check('J: عطل التدفق أُعلِن في السجل وأُعيد الرصد/الجلسة (خيط حيّ لا يحتاج قفز الكاش)',
      watchdog_logged and (attended or any(seen_force) or out2 > 0),
      'attended=%s force=%s log=%s' % (attended, seen_force[:2], watchdog_logged))
# الدورة كاملة تُقرأ من السجل وحده — هذا ما طلبه المستخدم: صمت ⇒ حارس ⇒ تنظيف ⇒ مراقبة ⇒
# LIVE ⇒ جلسة جديدة. الترتيب أدناه مقيس من logs نفسها، لا مفترض.
def _first(tag, after=-1):
    for i in range(after + 1, len(logs)):
        if tag in logs[i]:
            return i
    return -1


i_restart = _first('[RESTART]')
i_relay = _first('[RELAY]', i_restart)
i_watch = _first('[WATCHDOG]', i_relay)
i_clean = _first('[CLEANUP]', i_watch)
check('J: الدورة موثّقة بترتيبها في السجل: [RESTART]→[RELAY]→[WATCHDOG]→[CLEANUP]',
      -1 not in (i_restart, i_relay, i_watch, i_clean),
      'idx=%s/%s/%s/%s' % (i_restart, i_relay, i_watch, i_clean))

# الحمولة الحقيقية — لا قاموس محاكاة: تعليقٌ في نهاية سطر قاموس كان يبتلع المفاتيح التي بعده
# بصمت (الشيفرة تُترجَم واللوحة تفقد العمود). نقرأ snapshot() نفسه ونتأكد أن كل مفاتيح اللوحة حاضرة.
try:
    _snap = mgr.snapshot()
    _ev0 = (_snap.get('events') or [{}])[0]
    missing = [k for k in ('source_age_seconds', 'source_stall_limit', 'source_gap', 'output_seconds',
                           'youtube_health', 'phase') if k not in _ev0]
except Exception as exc:                                       # noqa: BLE001
    missing = ['snapshot رفع %s' % type(exc).__name__]
check('J: payload اللوحة فيه كل المفاتيح (لا مفتاح يبتلعه تعليق في نفس السطر)', not missing, str(missing))

check('J: لا جلسة مزدوجة ولا عمليات ffmpeg متراكمة',
      len(mgr.senders) == 1 and sum(1 for e in mgr.senders.values() if e['sender'].alive()) <= 1,
      'senders=%d' % len(mgr.senders))

# ── K) انتهاء البث: تعود الحلقة إلى المراقبة ولا يبقى المرسل عالقًا ────────────────
mgr.observe = lambda u, force=False: {'status': 'offline', 'room_id': '7', 'sources': [],
                                      'nickname': 'n', 'live_title': ''}
mgr.drive_event(key)
ev = mgr.journal.read()['events'].get(key)
check('K: offline ⇒ المرسل يتوقف والحدث يبقى للمراقبة (لا إنهاء ولا تعليق)',
      (not ev or not ev.get('closing')) and not any(e['sender'].alive() for e in mgr.senders.values()))
_dead = time.monotonic() + 8
while time.monotonic() < _dead and not any('[OFFLINE]' in m for m in logs):
    time.sleep(0.3)
check('K: نهاية بث تيك توك موسومة [OFFLINE] (تُفرَق في السجل عن صمت المصدر)',
      any('[OFFLINE]' in m for m in logs), logs[-1][:90])
for e in list(mgr.senders.values()):
    e['stop'].set()
    e['sender'].terminate()

# ── L) التطبيق يستورد، ونقاط الحالة تعمل بعد التعديل ───────────────────────────────
r = subprocess.run([sys.executable, '-c',
                    'import sys;sys.path.insert(0,%r);import app;print("IMPORT-OK")' % str(DIR)],
                   capture_output=True, text=True, env=dict(os.environ, RECORD_MODE='never'))
check('L: app.py يُستورد بلا استثناء', 'IMPORT-OK' in r.stdout, (r.stderr or '')[-140:])

# مفتاح اللوحة الجديد: عمر آخر بايت وصل من تيك توك يجب أن يعبر من snapshot المرسل إلى حدث الحالة،
# وإلا بقي المشاهد يرى «بث حيّ» بينما المصدر صامت منذ دقيقتين (نفس ما خفي وقت التجمّد).
class _FakeTh:
    def is_alive(self): return True
class _FakeSender:
    stall_limit = 45
    def snapshot(self): return {'status': 'يرسل', 'source': 'hd', 'output_seconds': 12.0, 'source_age': 3.0}
_snap = _FakeSender().snapshot()
_payload = {'source_age_seconds': _snap.get('source_age'), 'source_stall_limit': _FakeSender.stall_limit}
check('L: عمر آخر بايت من المصدر يمرّ إلى حالة اللوحة', _payload['source_age_seconds'] == 3.0
      and _payload['source_stall_limit'] == 45, str(_payload))
try:
    mgr.senders['panel-check'] = {'sender': _FakeSender(), 'thread': _FakeTh(), 'stop': type('S', (), {'is_set': lambda self: False})()}
    snap = mgr.snapshot()
    check('L: snapshot المدير يعمل مع المرسلات الحيّة (لا AttributeError على stall_limit)', isinstance(snap, dict), '')
except Exception as exc:                                        # noqa: BLE001
    check('L: snapshot المدير يعمل مع المرسلات الحيّة (لا AttributeError على stall_limit)', False, repr(exc)[:150])
finally:
    mgr.senders.pop('panel-check', None)
try:
    from fastapi.testclient import TestClient
    import app as appmod
    with TestClient(appmod.app) as c:
        r1 = c.get('/api/state', auth=('admin', 'admin'))
        r2 = c.get('/api/youtube', auth=('admin', 'admin'))
    check('L: نقاط الصحة/الحالة (FastAPI) تعمل', r1.status_code == 200 and r2.status_code == 200
      and 'version' in r1.json() and 'send_mode' in r2.json(),
          'state=%s youtube=%s ver=%s' % (r1.status_code, r2.status_code, r1.json().get('version')))
except ImportError as e:
    check('L: TestClient غير متاح (httpx ناقص) — فحص الاستيراد يكفي', True, str(e)[:60])

# أمانة في التقرير: لا Flask ولا /api/health في هذا التطبيق أصلًا (0 مواضع في كل كوميتات
# upstream) — فحص الصحة الفعلي هو /api/state و/api/youtube. نؤكد ذلك في الاختبار حتى لا يُقرأ
# مطلب «بقاء Flask /api/health عاملًا» على أنه شيء عُدِّل أو وُجد.
_src = (DIR / 'app.py').read_text(encoding='utf-8')
_routes = [seg.split(chr(39))[1] for seg in _src.splitlines() if seg.startswith('@app.') and chr(39) in seg]
check('L: الصحة = /api/state و/api/youtube (FastAPI) — لا Flask ولا /api/health في هذا المشروع',
      '/api/health' not in _routes and any(r.startswith('/api/state') for r in _routes)
      and 'flask' not in (DIR / 'requirements.txt').read_text(encoding='utf-8').lower(),
      'routes=%d' % len(_routes))

# ── M) لا عمليات متسربة بعد كل ما سبق ──────────────────────────────────────────────
ps = subprocess.run(['ps', '-eo', 'args'], capture_output=True, text=True).stdout.splitlines()
stray = [l for l in ps if 'testsrc2' in l or (str(tmp) in l and 'ffmpeg' in l)]
# ── N) الوجهة ترفض (187) ⇒ المشرف يبطئ النقر بدل إعادة المحاولة كل 15ث ──────────────
class _BoomSender:
    last_exit_code = 187
mgr.retry_min = 5.0
mgr.retry_max = 600.0
mgr.retry.clear()
mgr.hammer.clear()
logs_tail = []
mgr.log = lambda u, m: logs_tail.append(str(m))
for _ in range(3):
    mgr._note_early_exit('hammeruser', _BoomSender(), lasted=4.0)
third = mgr._backoff('hammeruser')
mgr._note_early_exit('hammeruser', _BoomSender(), lasted=4.0)
hammered = mgr._backoff('hammeruser')
# القفل الحقيقي: tick() يستدعي start_sender كل 15 ث؛ إن لم يُقفل retry_at ذلك الاستدعاء يبقى
# التأجيل تزيينًا. نُنشئ حدثًا وهميًا مربوطًا ونؤكد أن start_sender لا يبني شيئًا تحت القفل.
started_while_gated = 'لم يُنفَّذ'
gate_ok = False
try:
    mgr.journal.change(lambda d: d['events'].update({'gate:key': {'username': 'hammeruser', 'room_id': '7',
                                                                  'broadcast_id': 'X', 'bound': True,
                                                                  'marker': 'm'}}))
    mgr.hammer.clear(); mgr.retry_at.clear()
    for _ in range(4):
        mgr._note_early_exit('hammeruser', _BoomSender(), lasted=4.0)
    gate_ok = mgr.retry_at.get('gate:key', 0) > time.monotonic()
    mgr.start_sender('gate:key')
    e = mgr.senders.get('gate:key')
    started_while_gated = bool(e and e['thread'].is_alive())
    if e:
        e['stop'].set()
        e['sender'].terminate()
except Exception as exc:                    # noqa: BLE001
    started_while_gated = repr(exc)[:100]
check('N: التأجيل يقفل start_sender من tick() أيضًا (لا إعادة اتصال كل 15 ث على حافة رافضة)',
      gate_ok and started_while_gated is False, 'gated=%s started=%s' % (gate_ok, started_while_gated))

mgr.senders.pop('gate:key', None)
mgr.journal.change(lambda d: d['events'].pop('gate:key', None))
mgr.hammer.clear(); mgr.retry_at.clear()

# القفل يجب أن يكون مرئيًا وقابلاً للإلغاء: ملاحظة في الحدث، وclear_pacing يفتح الباب فورًا
_logs_gp = []
mgr.log = lambda u, m: _logs_gp.append(str(m))
mgr.journal.change(lambda d: d['events'].update({'gate:vis': {'username': 'visu', 'room_id': '9',
                                                              'title': 't', 'note': '', 'phase': 'live',
                                                              'broadcast_id': 'X', 'bound': True, 'marker': 'm'}}))
mgr.retry_at['gate:vis'] = time.monotonic() + 60
mgr.start_sender('gate:vis')
noted = ('إعادة الاتصال مؤجّلة' in ((mgr.journal.read()['events'].get('gate:vis') or {}).get('note') or ''))
n_cleared = mgr.clear_pacing('visu', 'اختبار')
vis_state = {k: v for k, v in (mgr.snapshot().get('events') or [{}])[0].items() if k in ('retry_in', 'note')}
mgr.journal.change(lambda d: d['events'].pop('gate:vis', None))
mgr.log = lambda u, m: logs.append(str(m))
check('N: القفل مرئي (ملاحظة «إعادة الاتصال مؤجّلة N ث» لا صمت) وقابل للإلغاء بقرار المشغّل',
      noted and n_cleared >= 1 and not mgr.retry_at, 'noted=%s cleared=%s' % (noted, n_cleared))
check('N: retry_in يصل إلى payload اللوحة (الحقل موجود ولو صفرًا)',
      all('retry_in' in ev for ev in (mgr.snapshot().get('events') or [{'retry_in': 0}])), str(vis_state)[:80])
_BoomSender.last_exit_code = 0
mgr.hammer.clear(); mgr.retry_at.clear()
mgr.journal.change(lambda d: d['events'].update({'gate:reset': {'username': 'hammeruser', 'room_id': '7'}}))
mgr._note_early_exit('hammeruser', _BoomSender(), lasted=40.0)
reset_ok = ('hammeruser' not in mgr.hammer and mgr._backoff('hammeruser') <= 60.0
            and 'gate:reset' not in mgr.retry_at)
mgr.journal.change(lambda d: d['events'].pop('gate:reset', None))
# retry_min في هذه الصورة =1 ث، فالأرضية (20 ث) هي الحدّ الأدنى المقصود: الطلعة الثالثة يجب
# ألا تكون «2 ث» كما كان قبل الأرضية، والرابعة يجب أن تصعد فوقها.
floor_ok = third >= 20 and hammered > third
logged_ok = any('تأجيل إعادة الاتصال إلى %d ث' % int(third) in m for m in logs_tail)
check('N: عمليات متتالية قفلتها الوجهة خلال ثوانٍ ⇒ أرضية 20 ث ثم تصاعد موثّق في السجل',
      floor_ok and logged_ok, 'third=%s fourth=%s logged=%s' % (third, hammered, logged_ok))
check('N: جلسة طويلة ⇒ يُصفَّر عدّاد النقر وتعود الوتيرة الطبيعية', reset_ok, str(mgr.hammer))

# N2: المصدر جائع لا الوجهة رافضة — العقاب يجب ألا يُطبَّق، بل يُسرَّع الرصد
class _HungrySender:
    last_exit_code = 187
    last_source_gap = 25.0


mgr.retry_min = 5.0
mgr.hammer.clear(); mgr.retry_at.clear(); mgr.retry.clear()
logs_n2 = []
mgr.log = lambda u, m: logs_n2.append(str(m))
for _ in range(4):
    mgr._note_early_exit('hungry', _BoomSender(), lasted=4.0)      # وجهة رافضة: عقاب مستحق
penalized = mgr._backoff('hungry') > mgr.retry_min
mgr.hammer.clear(); mgr.retry_at.clear(); mgr.retry.clear()
logs_n2.clear()
for _ in range(4):
    mgr._note_early_exit('hungry', _HungrySender(), lasted=4.0)    # نفس الرمز لكن المصدر جائع
unpenalized = (mgr._backoff('hungry') == mgr.retry_min and not mgr.hammer.get('hungry')
               and not mgr.retry_at and any('الجوع من جهة تيك توك' in m for m in logs_n2))
check('N2: جوع المصدر يُلغي عقاب الوجهة (تأجيل باطل يطيل صمت المشاهدين)',
      penalized and unpenalized, 'penalized=%s unpenalized=%s' % (penalized, unpenalized))
# N3: مصدر يرسل + ثلاث إغلاقات ⇒ المفتاح نفسه صار مشكوكًا فيه: يجب أن يُجهَّل ليُعاد من الـAPI
class _FeedingSender:
    last_exit_code = 187
    last_source_gap = 0.2


mgr.destinations['hammeruser'] = 'rtmps://old.example.invalid/live/OLDKEY'
seen_ensure = []
mgr.ensure_stream = lambda u: (seen_ensure.append(u), mgr.destinations.__setitem__(u, 'rtmps://new.example.invalid/live/NEWKEY_FROM_API'))
mgr.hammer.clear()
for _ in range(3):
    mgr._note_early_exit('hammeruser', _FeedingSender(), lasted=14.0)
# العنوان الجديد من الـAPI يجب أن يكون ما في الحوزة الآن (لا الأقدم، ولا فراغ بعد pop)
refreshed = (mgr.destinations.get('hammeruser', '').endswith('NEWKEY_FROM_API') and seen_ensure == ['hammeruser'])
check('N3: ثلاث إغلاقات مع مصدر يرسل تعيد قراءة عنوان الاستقبال فورًا (لا بعد موت الخيط)',
      refreshed and any('جُدِّد عنوان الاستقبال' in m for m in logs_n2)
      and mgr.destinations.get('hammeruser', '').endswith('NEWKEY_FROM_API'),
      'dest=%r logs=%s' % (mgr.destinations.get('hammeruser'), [m[:40] for m in logs_n2][-2:]))
mgr.retry_at.clear()
check('N3: sender_loop يقرأ الوجهة في كل دورة (النص SOURCE_OF_TRUTH هو dict العناوين)',
      'destination = self.destinations.get(user) or destination' in
      __import__('inspect').getsource(ym.ManagedYouTube.sender_loop), '')
mgr.destinations.pop('hammeruser', None)
# N4: التعافي يمسح ملاحظة العقاب من اللوحة — «مؤجّلة 27 ث» تحت «خرج 285 ث» كذب على المشغّل
mgr.journal.change(lambda dd: dd['events'].update({'rec:9': {
    'session_key': 'rec:9', 'username': 'recovered', 'nickname': 'r', 'room_id': '9', 'title': 't',
    'description': 'd', 'hashtags': '', 'privacy': 'public', 'marker': 'M', 'broadcast_id': 'BB',
    'bound': True, 'closing': False, 'phase': 'live', 'youtube_status': 'live',
    'created_at': time.time(), 'was_live': True, 'create_pending': False, 'last_poll': time.time(),
    'note': 'إعادة الاتصال مؤجّلة 68 ث (الوجهة أغلقت الاتصال ثلاث مرات والمصدر يرسل)'}}))
mgr.hammer['recovered'] = 5
mgr.retry['recovered'] = 90.0
mgr.retry_at['rec:9'] = time.monotonic() + 60
logs_n4 = []
mgr.log = lambda a, b: logs_n4.append(str(b))
mgr._note_early_exit('recovered', _FeedingSender(), lasted=40.0)
_note_now = (mgr.journal.read()['events'].get('rec:9') or {}).get('note') or ''
# N5: الوصول للتثبيت لا يحتاج موت جلسة — السند في start_sender موجود ويستدعي نفس الدالة
mgr.journal.change(lambda dd: dd['events'].update({'sta:8': {
    'session_key': 'sta:8', 'username': 'stableu', 'nickname': 's', 'room_id': '8', 'title': 't',
    'description': 'd', 'hashtags': '', 'privacy': 'public', 'marker': 'M2', 'broadcast_id': 'BS',
    'bound': True, 'closing': False, 'phase': 'live', 'youtube_status': 'live',
    'created_at': time.time(), 'was_live': True, 'create_pending': False, 'last_poll': time.time(),
    'note': 'إعادة الاتصال مؤجّلة 40 ث (الوجهة أغلقت الاتصال ثلاث مرات والمصدر يرسل)'}}))
mgr.destinations['stableu'] = 'rtmps://x.invalid/live/K'
mgr.hammer['stableu'] = 4
mgr.retry['stableu'] = 80.0
mgr.retry_at['sta:8'] = time.monotonic() + 40
logs_n5 = []
mgr.log = lambda a, b: logs_n5.append(str(b))
n_cleared = mgr.mark_stream_stable('stableu')
_ev = mgr.journal.read()['events'].get('sta:8') or {}
check('N5: mark_stream_stable يُلغي النافذة والملاحظة والعدّاد ويُنزل الوتيرة للأصل',
      n_cleared >= 3 and not mgr.retry_at and _ev.get('note', '') == '' and not mgr.hammer.get('stableu')
      and mgr._backoff('stableu') == mgr.retry_min and any('مُسح تأجيل العقاب' in x for x in logs_n5),
      'cleared=%s note=%r' % (n_cleared, _ev.get('note', '')[:24]))
mgr.journal.change(lambda dd: dd['events'].pop('sta:8', None))# N6: إغلاق على حدث لم يُشغَّل (ready) ⇒ لا تجديد مفتاح عبثي، والسبب يُكتب في الملاحظة والسجل
mgr.journal.change(lambda dd: dd['events'].update({'ns:6': {
    'session_key': 'ns:6', 'username': 'notstarted', 'nickname': 'n', 'room_id': '6', 'title': 't',
    'description': 'd', 'hashtags': '', 'privacy': 'public', 'marker': 'M6', 'broadcast_id': 'BN',
    'bound': True, 'closing': False, 'phase': 'bound', 'youtube_status': 'ready',
    'created_at': time.time(), 'was_live': False, 'create_pending': False, 'last_poll': time.time(),
    'note': ''}}))
mgr.destinations['notstarted'] = 'rtmps://x.invalid/live/K6'
seen_ns = []
mgr.ensure_stream = lambda u: seen_ns.append(u)
mgr.hammer.clear(); mgr.retry_at.clear(); mgr.retry.clear()
logs_n6 = []
mgr.log = lambda a, b: logs_n6.append(str(b))
for _ in range(3):
    mgr._note_early_exit('notstarted', _FeedingSender(), lasted=12.0)
_ev6 = mgr.journal.read()['events'].get('ns:6') or {}
check('N6: حدث بحالة ready — لا يُعاد قراءة المفتاح (كان قراءة API بلا معنى على سبب معروف)',
      not seen_ns, 'ensure_stream=%s' % seen_ns)
check('N6: السبب يصل اللوحة والسجل ببادئة تُمسح عند أول تدفق مقبول',
      str(_ev6.get('note') or '').startswith('إعادة الاتصال مؤجّلة') and 'لم يُشغَّل' in str(_ev6.get('note'))
      and any('لم يبدأ بعد' in x for x in logs_n6) and any('السبب المعلوم' in x for x in logs_n6),
      'note=%r' % str(_ev6.get('note'))[:60])
# والحدث الحيّ (was_live) لا يأخذ هذا المسار: يبقى تجديد المفتاح كما هو
mgr.journal.change(lambda dd: dd['events'].update({'ns:6': dict(dd['events']['ns:6'],
                                                                was_live=True, youtube_status='active')}))
mgr.hammer.clear(); mgr.retry_at.clear(); mgr.retry.clear()
seen_live = []
mgr.ensure_stream = lambda u: seen_live.append(u)
for _ in range(3):
    mgr._note_early_exit('notstarted', _FeedingSender(), lasted=12.0)
check('N6: بثّ كان حيًا — مسار تجديد المفتاح يعمل كما كان', seen_live == ['notstarted'], 'seen=%s' % seen_live)
mgr.journal.change(lambda dd: dd['events'].pop('ns:6', None))
mgr.hammer.clear(); mgr.retry_at.clear(); mgr.retry.clear()# N7: كتم سبام رسالة الفشل المتطابقة (توكن 403 كان يطبع 12/60 سطرًا) بلا كتم حالة اللوحة
import app as _app
_m1 = _app.throttle_failure('sp', 'تيك توك رفض الجلسة (403)', now=1000)
_m2 = _app.throttle_failure('sp', 'تيك توك رفض الجلسة (403)', now=1015)
_m3 = _app.throttle_failure('sp', 'تيك توك رفض الجلسة (403)', now=1300)
_m4 = _app.throttle_failure('sp', 'سبب آخر مختلف', now=1301)
_o1 = _app.throttle_failure('آخر', 'تيك توك رفض الجلسة (403)', now=1301)
check('N7: التطابق داخل دقيقتين يُكتم (لا سطر)، وغير المطابق يُكتب فورًا',
      _m1 and _m2 is None and _m3 and _m4, 'm1=%r m2=%r m3=%r m4=%r' % (_m1, _m2, _m3, _m4))
check('N7: عند أول كتابة بعد الكتم يُلحق عدد ما كُتم، وحساب آخر لا يتأثر',
      'كُتمت 1' in str(_m3) and 'كُتمت' not in str(_m4) and str(_o1) == 'تيك توك رفض الجلسة (403)',
      'm3=%r m4=%r o1=%r' % (_m3, _m4, _o1))
_src7 = open('/home/user/audit/app.py', encoding='utf-8').read()
check('N7: worker يستعمل الكتم والحالة تبقى صادقة (التحديث خارج الكتم)',
      'out=throttle_failure(user,msg)' in _src7
      and "update(user,status='انتظار إعادة المحاولة')" in _src7,
      'الاستدعاء أو تحديث الحالة مفقود')
_app._failure_notes.clear()# N8: نافذة تأجيل منتهية تمسح قفلها وملاحظتها المجمدة («مؤجّلة 17 ث») عند أول start_sender
mgr.journal.change(lambda dd: dd['events'].update({'ex:9': {
    'session_key': 'ex:9', 'username': 'expwin', 'nickname': 'e', 'room_id': '9', 'title': 't',
    'description': 'd', 'hashtags': '', 'privacy': 'public', 'marker': 'M9', 'broadcast_id': 'BE',
    'bound': True, 'closing': False, 'phase': 'live', 'youtube_status': 'active',
    'created_at': time.time(), 'was_live': True, 'create_pending': False, 'last_poll': time.time(),
    'note': 'إعادة الاتصال مؤجّلة 17 ث (الوجهة أغلقت الاتصال ثلاث مرات والمصدر يرسل)'}}))
mgr.retry_at['ex:9'] = time.monotonic() - 1.0
mgr._gate_note_at['ex:9'] = time.monotonic() - 30.0
mgr.senders.pop('ex:9', None)
mgr.destinations.pop('expwin', None)
mgr.ensure_stream = lambda u: (_ for _ in ()).throw(RuntimeError('لا عنوان في الاختبار'))
logs8 = []
mgr.log = lambda a2, b: logs8.append(str(b))
try:
    mgr.start_sender('ex:9')
except Exception:
    pass
_ev9 = mgr.journal.read()['events'].get('ex:9') or {}
check('N8: نافذة منتهية — القفل والملاحظة المجمدة يُمسحان ولا يعيقان المسار',
      'ex:9' not in mgr.retry_at and not str(_ev9.get('note') or ''), 'note=%r q=%s'
      % (str(_ev9.get('note'))[:40], 'ex:9' in mgr.retry_at))
check('N8: المسح يُعلَن في السجل مرة واحدة', sum('انتهت نافذة التأجيل' in x for x in logs8) == 1,
      'logs=%r' % [x[:60] for x in logs8])
# والنافذة الحيّة كما هي: تكتب ملاحظتها ولا تمسح القفل (حماية من انقلاب الإصلاح)
mgr.retry_at['ex:9'] = time.monotonic() + 60.0
mgr._gate_note_at.pop('ex:9', None)
try:
    mgr.start_sender('ex:9')
except Exception:
    pass
_ev9b = mgr.journal.read()['events'].get('ex:9') or {}
check('N8: نافذة حيّة — الملاحظة تُكتب والقفل يبقى',
      str(_ev9b.get('note') or '').startswith('إعادة الاتصال مؤجّلة') and 'ex:9' in mgr.retry_at,
      'note=%r' % str(_ev9b.get('note'))[:50])
mgr.journal.change(lambda dd: dd['events'].pop('ex:9', None))
mgr.retry_at.pop('ex:9', None)
mgr._gate_note_at.pop('ex:9', None)# N9: إيقاف الهوية — رفض 403 يُهدّئ رصد الحساب المرفوض وحده، وأي شيء آخر يُنهي الإيقاف
# (فحوص N4 الختامية تعتمد أحداثًا أنشأتها قبلنا ⇒ نحفظ الـjournal كاملًا ونسترجعه في النهاية)
_jb9 = json.loads(json.dumps(mgr.journal.read()))
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[]))
mgr.max_events = 3
calls = {'n': 0}
def _obs(u):
    calls['n'] += 1
    raise RuntimeError('تيك توك رفض الجلسة (403) — توكن sessionid منتهٍ')
mgr._observe = _obs
notes9 = []
mgr.log = lambda a2, b: notes9.append(str(b))
mgr.accounts = lambda: ['pa']
mgr.auth_pause.clear()
mgr.tick()
first = calls['n']
mgr.tick()
check('N9: رفض 403 يضبط إيقاف الرصد للحساب المرفوض وحده، والدورة التالية لا تسأل تيك توك',
      first == 1 and calls['n'] == 1 and mgr.auth_pause.get('pa', 0) > time.monotonic(),
      'calls=%s pause=%s' % (calls['n'], 'yes' if mgr.auth_pause.get('pa', 0) > time.monotonic() else 'no'))
mgr.auth_pause['pa'] = time.monotonic() - 1.0
mgr.tick()
check('N9: إيقاف منتهٍ يعيد السؤال (لا جمود دائم)', calls['n'] == 2, 'calls=%s' % calls['n'])
mgr.auth_pause['pa'] = time.monotonic() - 1.0   # tick2 أعاد تسليح الإيقاف (403 ثانية) فتُبَطَل قبل دورة الخطأ المختلف
def _obs5(u):
    calls['n'] += 1
    raise RuntimeError('تيك توك أعاد 500؛ سيُعاد المحاولة لاحقًا')
mgr._observe = _obs5
mgr.tick()
check('N9: خطأ غير هوية لا يضبط الإيقاف (ويُنهي ما ضُبط سابقًا) — الرصد يبقى بكامل سرعته',
      calls['n'] == 3 and not mgr.auth_pause.get('pa'), 'calls=%s pause=%s'
      % (calls['n'], mgr.auth_pause.get('pa')))
def _obsok(u):
    calls['n'] += 1
    return {'status': 'offline', 'room_id': 'x'}
mgr._observe = _obsok
mgr.tick()
check('N9: نجاح الرصد يمسح الإيقاف', calls['n'] == 4 and not mgr.auth_pause.get('pa'), 'calls=%s' % calls['n'])
def _restore9(dd):
    dd.clear(); dd.update(_jb9)
mgr.journal.change(_restore9)
mgr.auth_pause.clear()
# N10: تناوب النقل — جوعان على FLV يقدّم HLS أولًا، وأول 25 ث مُسلَّمة تعيد الترتيب الأصلي
_obs10 = {'sources': [
    {'url': 'https://x/flv-a', 'media': 'flv', 'codec': 'h264', 'label': 'hd · h264'},
    {'url': 'https://x/pl.m3u8', 'media': 'hls', 'codec': 'h264', 'label': 'hd · h264 · HLS'}]}
mgr.starve.pop('u10', None)
base10 = mgr._select_source('u10', _obs10)
check('N10: بلا جوع — FLV أولًا كما هي', base10[0]['media'] == 'flv', base10[0]['media'])
mgr.starve['u10'] = 2
rot10 = mgr._select_source('u10', _obs10)
check('N10: جوعان ⇒ أول مرشح غير FLV يُقدَّم أولًا (اختبار النقل البديل)',
      rot10[0]['media'] == 'hls' and rot10[-1]['media'] == 'flv',
      [c['media'] for c in rot10])
_sd10 = _FeedingSender(); _sd10.last_source_gap = 12.0
mgr.log = lambda a2, b: None
mgr._note_early_exit('u10', _sd10, lasted=12.0)
check('N10: خروج جوع يزيد عدّاد التناوب', int(mgr.starve.get('u10') or 0) >= 3,
      'starve=%s' % mgr.starve.get('u10'))
mgr.mark_stream_stable('u10')
check('N10: 25 ث مُسلَّمة تُلغي التناوب فيعود FLV أولًا',
      not mgr.starve.get('u10') and mgr._select_source('u10', _obs10)[0]['media'] == 'flv',
      'starve=%s' % mgr.starve.get('u10'))
mgr.starve.pop('u10', None)# N11: كتم سبام رسالة الإمساك — متطابقة داخل 120 ث تُكتم، وعند الكتابة يُلحق العدّاد
logs11 = []
mgr.log = lambda a2, b: logs11.append(str(b))
mgr._hold_notes.pop('hz', None)
mgr._log_throttled('hz', 'نتمسك بآخر مصدر صالح (100s)')
mgr._log_throttled('hz', 'نتمسك بآخر مصدر صالح (133s)')
check('N11: الرسالة المتطابقة داخل النافذة تُكتم (سطر واحد فقط)', len(logs11) == 1, 'logs=%d' % len(logs11))
_mh = mgr._hold_notes.get('hz')
_mh[1] -= 121.0                    # أقدم النافذة صناعيًا
mgr._log_throttled('hz', 'نتمسك بآخر مصدر صالح (254s)')
check('N11: بعد انتهاء النافذة تُكتب وتُلحق عدد ما كُتم',
      len(logs11) == 2 and 'كُتمت 1' in logs11[-1], 'last=%r' % logs11[-1][:60])
mgr._log_throttled('hz', 'رسالة أخرى مختلفة')
check('N11: رسالة غير متطابقة تُكتب فورًا', len(logs11) == 3, 'logs=%d' % len(logs11))
mgr._hold_notes.pop('hz', None)# N12: حجب المصادر ⇒ ملاحظة صادقة تُكتب مرة، والخيط يخرج بلا عوائق
import threading as _th
mgr.journal.change(lambda dd: dd['events'].update({'bl:1': {
    'session_key': 'bl:1', 'username': 'blocked', 'nickname': 'b', 'room_id': '5', 'title': 't',
    'description': 'd', 'hashtags': '', 'privacy': 'public', 'marker': 'MB', 'broadcast_id': 'BB',
    'bound': True, 'closing': False, 'phase': 'live', 'youtube_status': 'active',
    'created_at': time.time(), 'was_live': True, 'create_pending': False, 'last_poll': time.time(),
    'note': 'يوتيوب يستقبل؛ طُلب بدء الحدث للعامة حسب الخصوصية'}}))
mgr._observe = lambda u: {'status': 'unknown', 'room_id': '', 'sources': [], 'nickname': 'b', 'live_title': 't'}
_stop12 = _th.Event()
_sd12 = _FeedingSender()
_th12 = _th.Thread(target=mgr.sender_loop, args=('bl:1', 'blocked', '5', 'rtmps://x/k', _stop12, _sd12), daemon=True)
_th12.start(); _th12.join(timeout=6)
_ev12 = mgr.journal.read()['events'].get('bl:1') or {}
check('N12: حجب المصادر — الملاحظة الصادقة تُكتب مرة والخيط يخرج',
      _th12.is_alive() is False
      and str(_ev12.get('note') or '') == 'تيك توك لا يرجع مصادر لهذا الخادم الآن؛ الحدث محفوظ والمحاولة كل دورة رصد',
      'alive=%s note=%r' % (_th12.is_alive(), str(_ev12.get('note'))[:50]))
_stop12.set()
mgr.journal.change(lambda dd: dd['events'].pop('bl:1', None))# N13: المسار الصامت مغلق — «غير مباشر» يُكتب عند التحوّل فقط، وstreak جديد يعيده
st13 = []
mgr.max_events = 9        # استرجاع N9 أعاد أحداث N4 — بلا رفع السقف تُقطع حلقة الرصد قبل 'pa'
_pre13 = mgr.journal.read().get('enabled')   # مراحل سابقة قد تترك الأتمتة معطلة ⇒ tick يرجع قبل الرصد
mgr.journal.change(lambda dd: dd.update(enabled=True))
mgr.log = lambda a2, b: st13.append(str(b))
mgr.last_seen_status.pop('pa', None)
mgr._observe = lambda u: {'status': 'offline', 'room_id': '', 'sources': [], 'nickname': 'x', 'live_title': 't'}
mgr.tick()
n13a = sum('غير مباشر الآن' in x for x in st13)
mgr.tick()
n13b = sum('غير مباشر الآن' in x for x in st13)
check('N13: أول تقرير غير-حيّ يُكتب مرة', n13a == 1, 'n=%d' % n13a)
check('N13: تكرار نفس الحالة لا يعيد السطر', n13b == 1, 'n=%d' % n13b)
mgr._observe = lambda u: {'status': 'live', 'room_id': '13', 'sources': [], 'nickname': 'x', 'live_title': 't'}
mgr.tick()
mgr._observe = lambda u: {'status': 'offline', 'room_id': '', 'sources': [], 'nickname': 'x', 'live_title': 't'}
mgr.tick()
n13c = sum('غير مباشر الآن' in x for x in st13)
check('N13: streak جديد (عاد حيًّا ثم انطفأ) يُكتب من جديد', n13c == 2, 'n=%d' % n13c)
mgr.last_seen_status.pop('pa', None)
mgr.max_events = 3
mgr.journal.change(lambda dd: dd.update(enabled=_pre13))
# N14: تعدد المذيعين المتزامن — حسابان حيّان في نفس الدورة ⇒ حدثان بلا تلوّث، والسقف يُلتزم
_jb14 = json.loads(json.dumps(mgr.journal.read()))
_pre14_acc = mgr.accounts
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[]))
mgr.max_events = 3
mgr.accounts = lambda: ['m1', 'm2']
mgr.last_seen_status.pop('m1', None); mgr.last_seen_status.pop('m2', None)
def _obs14(u):
    return {'status': 'live', 'room_id': 'R1' if u == 'm1' else 'R2',
            'sources': [{'url': 'https://x/%s.flv' % u, 'media': 'flv', 'codec': 'h264',
                         'label': 'hd · h264'}],
            'nickname': u, 'live_title': 't'}
mgr._observe = _obs14
seen14, opened14, driven14 = [], [], []
mgr.ensure_stream = lambda u: seen14.append(u)
def _open14(u, o):
    opened14.append((u, o.get('room_id')))
    key = u + ':' + o['room_id']
    mgr.journal.change(lambda dd: dd['events'].update({key: {
        'session_key': key, 'username': u, 'nickname': u, 'room_id': o['room_id'],
        'title': 't', 'description': 'd', 'hashtags': '', 'privacy': 'public',
        'marker': 'M', 'broadcast_id': 'B-' + u, 'bound': True, 'closing': False,
        'phase': 'live', 'youtube_status': 'active', 'created_at': time.time(),
        'was_live': True, 'create_pending': False, 'last_poll': time.time(), 'note': ''}}))
    return True
mgr.open_event = _open14
mgr.drive_event = lambda key: driven14.append(key)
mgr.tick()
_keys14 = set(mgr.journal.read()['events'].keys())
check('N14: مذيعان حيّان في دورة واحدة ⇒ حدثان متزامنان بلا تلوّث مفاتيح',
      _keys14 == {'m1:R1', 'm2:R2'}, 'keys=%s' % sorted(_keys14))
check('N14: ensure_stream/open_event لكل حساب باسمه (لا تبادل أدوار)',
      seen14 == ['m1', 'm2'] and opened14 == [('m1', 'R1'), ('m2', 'R2')]
      and driven14 == ['m1:R1', 'm2:R2'],
      'seen=%s opened=%s driven=%s' % (seen14, opened14, driven14))
# السقف: حساب ثالث/ثانٍ يُؤجَّل عند بلوغ الحد، والحالة تعلن الانتظار
mgr.journal.change(lambda dd: dd.update(events={}, history={}.copy() or dd.get('history', [])))
mgr.journal.change(lambda dd: dd.update(history=[]))
mgr.max_events = 1
mgr.accounts = lambda: ['m1', 'm2']
mgr.tick()
check('N14: السقف max_events=1 يُلتزم (المذيع الثاني ينتظر ولا يُخترق)',
      set(mgr.journal.read()['events'].keys()) == {'m1:R1'} and 'الحد الأقصى' in mgr.status_text,
      'keys=%s status=%r' % (sorted(mgr.journal.read()['events'].keys()), mgr.status_text[:40]))
def _restore14(dd):
    dd.clear(); dd.update(_jb14)
mgr.journal.change(_restore14)
mgr.accounts = _pre14_acc
mgr.max_events = 3
# N15: علامة ملكية مفقودة عند الإنهاء ⇒ حلّ لا عالق: بث فعلي ⇒ إنهاء+أرشفة، غير مبدوء ⇒ أرشفة بلا مساس
_jb15 = json.loads(json.dumps(mgr.journal.read()))
_orig_remote15, _orig_api15, _orig_log15, _orig_stop15 = mgr.remote, mgr.api, mgr.log, mgr.stop_sender
def _mk15(status):
    mgr.journal.change(lambda dd: dd['events'].update({'mm:1': {
        'session_key': 'mm:1', 'username': 'mm1', 'nickname': 'm', 'room_id': '8', 'title': 't',
        'description': 'd', 'hashtags': '', 'privacy': 'public', 'marker': 'MK-EVENT-x',
        'broadcast_id': 'BB15', 'bound': True, 'closing': True, 'phase': 'closing',
        'youtube_status': 'active', 'created_at': time.time(), 'was_live': True,
        'create_pending': False, 'last_poll': time.time(), 'note': ''}}))
    calls15 = []
    class _API15:
        def call(self, *a, **k): calls15.append(a)
    mgr.api = _API15()
    mgr.remote = lambda res, rid: {'id': rid, 'snippet': {'description': 'وصف بلا علامة'},
                                   'status': {'lifeCycleStatus': status}}
    logs15 = []
    mgr.log = lambda a2, b: logs15.append(str(b))
    return calls15, logs15
calls15, logs15 = _mk15('live')
ok15 = mgr.finish_event('mm:1')
_h15 = mgr.journal.read().get('history') or []
check('N15: علامة مفقودة وكان البث فعليًا ⇒ transition complete + أرشفة (لا عالق في closing)',
      ok15 and 'mm:1' not in mgr.journal.read()['events']
      and any('complete' in str(a) for a in calls15)
      and _h15 and _h15[-1].get('outcome') == 'completed_marker_lost'
      and any('WARN' in x for x in logs15),
      'ok=%s calls=%s outcome=%s' % (ok15, calls15, _h15[-1].get('outcome') if _h15 else '—'))
calls15b, _ = _mk15('ready')
ok15b = mgr.finish_event('mm:1')
_h15b = mgr.journal.read().get('history') or []
check('N15: علامة مفقودة وغير مبدوء ⇒ أرشفة محلية فقط (لا حذف ولا إنهاء لبث قد يكون للغير)',
      ok15b and not calls15b and _h15b and str(_h15b[-1].get('outcome')).startswith('marker_mismatch'),
      'ok=%s calls=%s outcome=%s' % (ok15b, calls15b, _h15b[-1].get('outcome') if _h15b else '—'))
def _restore15(dd):
    dd.clear(); dd.update(_jb15)
mgr.journal.change(_restore15)
mgr.remote, mgr.api, mgr.log, mgr.stop_sender = _orig_remote15, _orig_api15, _orig_log15, _orig_stop15
# N16: 403 من يوتيوب عند فتح الحدث ⇒ إنهاء بنظام بملاحظة صريحة (لا planned معلقة تحرق مقعدًا)
_jb16 = json.loads(json.dumps(mgr.journal.read()))
_pre16_acc, _pre16_me = mgr.accounts, mgr._observe
_pre16_es, _pre16_oe, _pre16_api = mgr.ensure_stream, mgr.open_event, mgr.api
_pre16_log, _pre16_fin = mgr.log, mgr.finish_event
_pre16_arch = mgr.archive_event
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[]))
mgr.max_events = 3
mgr.accounts = lambda: ['rj1']
def _obs16(u, *a, **k):
    return {'status': 'live', 'room_id': 'R9',
            'sources': [{'url': 'https://x/rj.flv', 'media': 'flv', 'codec': 'h264', 'label': 'hd'}],
            'nickname': u, 'live_title': 't'}
mgr._observe = _obs16
mgr.ensure_stream = lambda u: None
class _APIErr(Exception):
    pass
from youtube_api import APIError as _APIErrCls
def _open16(u, o):
    mgr._last_obs = {'username': u, 'room_id': o.get('room_id')}
    raise _APIErrCls(403, 'livePermissionBlocked')
mgr.open_event = _open16
logs16 = []
mgr.log = lambda a2, b: logs16.append(str(b))
mgr.last_seen_status.pop('rj1', None)
mgr.auth_pause.pop('rj1', None)
mgr.tick()
check('N16: رفض 403 عند الفتح ⇒ السجل يسمّي الصلاحية والحدث يُغلق بنظام (لا planned معلّقة)',
      any('403/livePermissionBlocked' in str(x) for x in logs16)
      and all(str(e.get('note') or '').find('صلاحية') < 0 for e in mgr.journal.read().get('events', {}).values()),
      'logs=%s events=%s' % ([str(x)[:60] for x in logs16], list(mgr.journal.read().get('events', {}).keys())))
def _restore16(dd):
    dd.clear(); dd.update(_jb16)
mgr.journal.change(_restore16)
mgr.accounts, mgr._observe = _pre16_acc, _pre16_me
mgr.ensure_stream, mgr.open_event, mgr.api = _pre16_es, _pre16_oe, _pre16_api
mgr.log, mgr.finish_event, mgr.archive_event = _pre16_log, _pre16_fin, _pre16_arch
# N16b: قفل المفتاح بعد رفض 403 — open_event الحقيقي يرفض إعادة فتح نفس مفتاح الجلسة داخل النافذة
_jb16b = json.loads(json.dumps(mgr.journal.read()))
_pre16b_es, _pre16b_log, _pre16b_probe = mgr.ensure_stream, mgr.log, mgr.probe
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[]))
mgr.max_events = 3
mgr.accounts = lambda: ['rj1']
mgr._observe = _obs16
mgr.probe = lambda u, c: {'video': 'h264', 'audio': 'aac'}
# كتل سابقة (N14) تترك رقعها دون استعادة — نربط الحقيقي من الصنف صراحةً
_pre16b_oe, _pre16b_sn = mgr.open_event, mgr._skip_note
mgr.open_event = ym.ManagedYouTube.open_event.__get__(mgr)
mgr._skip_note = ym.ManagedYouTube._skip_note.__get__(mgr)
_es16b_calls = {'n': 0}
def _es16b(u):
    # الدورة الأولى: يوتيوب يرفض قبل إنشاء الحدث (الرفض يسبق open_event) — القفل
    # يُضبط من _last_obs الذي يضبطه tick قبل ensure_stream. الدورة الثانية: المصدر
    # يُسلَّم طبيعيًا ويجب أن يمنع القفل في open_event الحقيقي إعادة الفتح.
    _es16b_calls['n'] += 1
    if _es16b_calls['n'] == 1:
        raise _APIErrCls(403, 'livePermissionBlocked')
    return None
mgr.ensure_stream = _es16b
logs16b = []
mgr.log = lambda a2, b: logs16b.append(str(b))
mgr.last_seen_status.pop('rj1', None)
mgr.auth_pause.pop('rj1', None)
mgr._blocked403.pop('rj1:R9', None)
mgr.tick()
_events16b_after1 = len(mgr.journal.read().get('events') or {})
mgr.tick()
_rejects16b = sum(1 for x in logs16b if '403/livePermissionBlocked' in str(x))
check('N16b: بعد رفض 403 يُقفل المفتاح (rj1:R9) — open_event الحقيقي يرفض إعادة الفتح بملاحظة انتظار، بلا planned ولا رفض متكرر',
      'rj1:R9' in mgr._blocked403
      and _events16b_after1 == 0
      and _rejects16b == 1
      and any('يوتيوب رفض إنشاء حدث لهذه الجلسة' in str(v) for v in (mgr._skip_last or {}).values()),
      'blocked=%s events=%d rejects=%d skip=%s' % (list(mgr._blocked403.keys()), _events16b_after1,
                                                   _rejects16b,
                                                   [str(v)[:42] for v in (mgr._skip_last or {}).values()]))
mgr.ensure_stream = _pre16b_es
mgr.log = _pre16b_log
mgr.probe = _pre16b_probe
mgr.open_event, mgr._skip_note = _pre16b_oe, _pre16b_sn
def _restore16b(dd):
    dd.clear(); dd.update(_jb16b)
mgr.journal.change(_restore16b)
mgr._blocked403.pop('rj1:R9', None)

# N16c: 403 من حلقة القيادة (POST liveBroadcasts في ensure_broadcast) — كانت تُدفن في
# _event_error فتبقى الحدث planned يعيد POST كل دورة مسح (حالة الإنتاج 20:09)؛ الآن يُغلق
# بنظام عبر _youtube_rejected ويُقفل مفتاح الجلسة. الحدث يُبنى عبر open_event الحقيقي.
_jb16c = json.loads(json.dumps(mgr.journal.read()))
_pre16c_api, _pre16c_log = mgr.api, mgr.log
_pre16c_es, _pre16c_oe, _pre16c_probe = mgr.ensure_stream, mgr.open_event, mgr.probe
mgr.journal.change(_restore16b)
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[]))
mgr.max_events = 3
mgr.accounts = lambda: ['rj1']
def _obs16c(u, *a, **k):
    return {'status': 'live', 'room_id': 'R7',
            'sources': [{'url': 'https://x/rj.flv', 'media': 'flv', 'codec': 'h264', 'label': 'hd'}],
            'nickname': u, 'live_title': 't'}
mgr._observe = _obs16c
mgr.ensure_stream = lambda u: None
mgr.probe = lambda u, c: {'video': 'h264', 'audio': 'aac'}
_pre16c_oe2, _pre16c_de = mgr.open_event, mgr.drive_event
_pre16c_fe, _pre16c_ae = mgr.finish_event, mgr.archive_event
mgr.open_event = ym.ManagedYouTube.open_event.__get__(mgr)
mgr.drive_event = ym.ManagedYouTube.drive_event.__get__(mgr)
mgr.finish_event = ym.ManagedYouTube.finish_event.__get__(mgr)
mgr.archive_event = ym.ManagedYouTube.archive_event.__get__(mgr)
def _api16c_call(*a, **k):
    raise _APIErrCls(403, 'livePermissionBlocked')
mgr.api = type('FakeAPI403', (), {'call': staticmethod(_api16c_call)})()
logs16c = []
mgr.log = lambda a2, b: logs16c.append(str(b))
mgr.last_seen_status.pop('rj1', None)
mgr.auth_pause.pop('rj1', None)
mgr._blocked403.pop('rj1:R7', None)
mgr.tick()
_ev16c = (mgr.journal.read().get('events') or {}).get('rj1:R7')
_hist16c = mgr.journal.read().get('history') or []
_rejects16c = sum(1 for x in logs16c if '403/livePermissionBlocked' in str(x))
_note16c = next((str(h.get('note') or '') for h in _hist16c if h.get('session_key') == 'rj1:R7'), '')
check('N16c: رفض 403 في drive_event (POST الحدث) ⇒ إغلاق بنظام بملاحظة صادقة + قفل المفتاح — لا planned تعيد POST كل دورة',
      _ev16c is None
      and any(h.get('session_key') == 'rj1:R7' and h.get('outcome') == 'cancelled_before_create' for h in _hist16c)
      and 'صلاحية' in _note16c
      and 'rj1:R7' in mgr._blocked403
      and _rejects16c == 1,
      'in_events=%s hist=%s note=%r blocked=%s rejects=%d' % (
          _ev16c is not None,
          [(h.get('session_key'), h.get('outcome')) for h in _hist16c],
          _note16c[:60], list(mgr._blocked403.keys()), _rejects16c))
mgr.api, mgr.log = _pre16c_api, _pre16c_log
mgr.ensure_stream, mgr.open_event, mgr.probe = _pre16c_es, _pre16c_oe, _pre16c_probe
mgr.drive_event, mgr.finish_event, mgr.archive_event = _pre16c_de, _pre16c_fe, _pre16c_ae
def _restore16c(dd):
    dd.clear(); dd.update(_jb16c)
mgr.journal.change(_restore16c)
mgr._blocked403.pop('rj1:R7', None)

# N17: تبديل القناة المتعمد — تمرير معرف القناة الجديدة في start هو تأكيد المشغّل الصريح:
# أرشفة محلية لأحداث القناة السابقة (بلا تعديل بعيد) وتسجيد مصادرها؛ وبلا تأكيد يبقى التوقف الآمن
# إيقاف خيط المدير الخلفي أولًا: أُطلق من اختبارات الخدمة أعلاه وتداخله مع tick الاختبار
# (دورة خلفية متزامنة) كان يبتلع أحداث الاختبار ويرصد غرفًا خطأ — ثبت بالتحسسات (l71)
_ms17 = getattr(mgr, 'manager_stop', None)
if _ms17 is not None:
    _ms17.set()
    _mt17 = getattr(mgr, 'manager_thread', None)
    if _mt17 is not None and _mt17.is_alive():_mt17.join(timeout=5)
_jb17 = json.loads(json.dumps(mgr.journal.read()))
_pre17_api, _pre17_log, _pre17_stop = mgr.api, mgr.log, mgr.stop_sender
def _restore17(dd):
    dd.clear(); dd.update(_jb17)
mgr.journal.change(_restore17)
mgr.journal.change(lambda dd: dd.update(enabled=False, channel={'id': 'OLDCH', 'title': 'القناة القديمة'},
                                        streams={'rj1': {'id': 'oldstream'}}, events={
    'rj1:R5': {'session_key': 'rj1:R5', 'name': 'rj1', 'title': 't', 'description': 'm17', 'privacy': 'unlisted',
               'date': 'd', 'time': 't', 'username': 'rj1', 'room_id': 'R5', 'broadcast_id': None,
               'create_pending': False, 'phase': 'planned', 'bound': False, 'was_live': False, 'closing': False,
               'last_poll': 0, 'offline_since': None, 'offline_count': 0, 'note': ''}}, history=[]))
_stop17 = []
mgr.stop_sender = lambda k: _stop17.append(k) or True
logs17 = []
mgr.log = lambda a2, b: logs17.append(str(b))
class _ChAPI:
    def __init__(self, cid):
        self.cid = cid
    def call(self, resource, method='GET', params=None, body=None):
        if resource == 'channels':
            return {'items': [{'id': self.cid, 'snippet': {'title': 'قناة جديدة'}}]}
        raise AssertionError('نداء غير متوقع ' + resource)
mgr.api = _ChAPI('NEWCH')
_safe17 = False
try:
    mgr.check_channel()
except RuntimeError as e:
    _safe17 = 'القناة المفوضة تغيرت' in str(e)
check('N17a: تغيّر القناة بلا تأكيد يبقى توقفًا آمنًا — لا هجرة صامتة لأحداث لا تخص القناة', _safe17, 'safe=%r' % _safe17)
_rep17 = mgr.check_channel(report_only=True)
_d17b = mgr.journal.read()
check('N17c: فحص الربط يبلّغ عن القناة الجديدة دون اعتمادها — التوقف الآمن وسجل القناة وأحداثها كما هي',
      _rep17.get('id') == 'NEWCH' and (_rep17.get('previous') or {}).get('id') == 'OLDCH'
      and (_d17b.get('channel') or {}).get('id') == 'OLDCH'
      and any(e.get('username') == 'rj1' for e in (_d17b.get('events') or {}).values())
      and (_d17b.get('streams') or {}).get('rj1', {}).get('id') == 'oldstream',
      'rep=%r channel=%s events=%d streams=%s' % (_rep17, (_d17b.get('channel') or {}).get('id'),
                                                  len(_d17b.get('events') or {}), list((_d17b.get('streams') or {}).keys())))
_ch17 = mgr.check_channel(allow_switch=True, expected='NEWCH')
_d17 = mgr.journal.read()
_h17 = [h for h in (_d17.get('history') or []) if h.get('outcome') == 'local_cancelled_channel_switch']
check('N17b: التبديل المتعمد يؤرشف أحداث القناة السابقة محليًا ويُسجّد مصادرها ويعتمد الجديدة بسجل صادق',
      _ch17.get('id') == 'NEWCH' and not _d17.get('events') and not _d17.get('streams')
      and len(_h17) == 1 and _h17[0].get('session_key') == 'rj1:R5'
      and 'القناة السابقة' in str(_h17[0].get('note') or '')
      and (_d17.get('channel') or {}).get('id') == 'NEWCH'
      and any('تم اعتماد قناة جديدة' in x for x in logs17),
      'ch=%r events=%s streams=%s hist=%s stop=%s' % (_ch17, list((_d17.get('events') or {}).keys()),
                                                      list((_d17.get('streams') or {}).keys()),
                                                      [(h.get('session_key'), h.get('outcome')) for h in (_d17.get('history') or [])], _stop17))
mgr.api, mgr.log, mgr.stop_sender = _pre17_api, _pre17_log, _pre17_stop
mgr.journal.change(_restore17)

# N18: إعلان «ليس محتوى أطفال» للحسابات المخصصة — قيد 18+ غير مدعوم في واجهة يوتيوب
# الرسمية؛ POST الحدث يحمل selfDeclaredMadeForKids=false لهذه الحسابات فقط
_jb18 = json.loads(json.dumps(mgr.journal.read()))
_pre18_api, _pre18_log, _pre18_es, _pre18_ob, _pre18_probe = mgr.api, mgr.log, mgr.ensure_stream, mgr._observe, mgr.probe
def _restore18(dd):
    dd.clear(); dd.update(_jb18)
_posts18 = []
class _FakeAPI18:
    def call(self, resource, method='GET', params=None, body=None):
        if resource == 'liveBroadcasts' and method == 'POST':
            _posts18.append(body)
            return {'id': 'bb18'}
        return {'items': []}
def _obs18(u, *a, **k):
    return {'status': 'live', 'room_id': 'R1x',
            'sources': [{'url': 'https://x/rj.flv', 'media': 'flv', 'codec': 'h264', 'label': 'hd'}],
            'nickname': u, 'live_title': 't'}
_au18 = mgr._adult_account('AHMEDBARBOOKA1') and not mgr._adult_account('muhajer_iraqi')
mgr.journal.change(_restore18)
mgr.api = _FakeAPI18()
mgr._observe = _obs18
mgr.ensure_stream = lambda u: None
mgr.probe = lambda u, c: {'video': 'h264', 'audio': 'aac'}
# بقايا رقع N14 تعود عبر استعادات الكتل — نربط الحقيقي صراحةً (درس N16b نفسه)
mgr.open_event = ym.ManagedYouTube.open_event.__get__(mgr)
mgr.drive_event = ym.ManagedYouTube.drive_event.__get__(mgr)
logs18 = []
mgr.log = lambda a2, b: logs18.append(str(b))
mgr.last_seen_status.pop('ahmedbarbooka1', None); mgr.auth_pause.pop('ahmedbarbooka1', None)
mgr._blocked403.pop('ahmedbooka1:R1x', None); mgr._blocked403.pop('ahmedbarbooka1:R1x', None)
mgr.accounts = lambda: ['ahmedbarbooka1']
mgr.max_events = 3
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[]))
mgr.tick()
_post_adult = [b for b in _posts18 if isinstance(b, dict)]
_ok18 = bool(_post_adult) and any(
    (b.get('status') or {}).get('selfDeclaredMadeForKids') is False for b in _post_adult)
mgr.journal.change(lambda dd: dd.update(events={}, history=[]))
mgr.accounts = lambda: ['rj1']
mgr.last_seen_status.pop('rj1', None); mgr.auth_pause.pop('rj1', None)
mgr._blocked403.pop('rj1:R1x', None)
mgr.tick()
_ok18b = any((b.get('status') or {}).get('selfDeclaredMadeForKids') is None for b in _posts18[len(_post_adult):])
check('N18: حسابات المذيع المخصص تُنشأ بإعلان selfDeclaredMadeForKids=false — والحسابات العادية بلا إعلان',
      _au18 and _ok18 and _ok18b,
      'adult_helper=%r adult_post=%s normal_post=%s' % (_au18,
          [(b.get('status') or {}) for b in _post_adult], [b.get('status') for b in _posts18[len(_post_adult):]]))
mgr.api, mgr.log, mgr.ensure_stream = _pre18_api, _pre18_log, _pre18_es
mgr._observe, mgr.probe = _pre18_ob, _pre18_probe
mgr.journal.change(_restore18)

# N23: حدث عالق في الإنهاء — بث liveStarting لم يُشغَّل أبدًا (لا was_live ولا
# actualStartTime) يُحذف بأمان ويُؤرشف empty_deleted بدل «انتظار اكتمال الانتقال» للأبد
_jb23 = json.loads(json.dumps(mgr.journal.read()))
_pre23_api, _pre23_log, _pre23_stop = mgr.api, mgr.log, mgr.stop_sender
def _restore23(dd):
    dd.clear(); dd.update(_jb23)
_deleted23 = []
class _DelAPI23:
    def call(self, resource, method='GET', params=None, body=None):
        if resource == 'liveBroadcasts' and method == 'DELETE':
            _deleted23.append(params.get('id'))
            return {}
        if resource == 'liveBroadcasts' and method == 'GET':
            return {'items': [{'id': 'bb23', 'status': {'lifeCycleStatus': 'liveStarting'},
                               'snippet': {'description': 'desc\nMARK23'}}]}
        return {'items': []}
mgr.journal.change(_restore23)
mgr.stop_sender = lambda k: True
logs23 = []
mgr.log = lambda a2, b: logs23.append(str(b))
mgr.api = _DelAPI23()
mgr.journal.change(lambda dd: dd['events'].update({'rj1:S23': {
    'session_key': 'rj1:S23', 'name': 'rj1', 'title': 't', 'description': 'desc\nMARK23', 'privacy': 'unlisted',
    'date': 'd', 'time': 't', 'username': 'rj1', 'room_id': 'S23', 'broadcast_id': 'bb23',
    'create_pending': False, 'phase': 'closing', 'bound': True, 'was_live': False, 'closing': True,
    'last_poll': 0, 'offline_since': None, 'offline_count': 0, 'note': '', 'marker': 'MARK23'}}))
mgr.finish_event('rj1:S23')
_d23 = mgr.journal.read()
_hist23 = [h for h in (_d23.get('history') or []) if h.get('session_key') == 'rj1:S23']
check('N23: بث liveStarting لم يُشغَّل أبدًا يُحذف ويُؤرشف empty_deleted — لا إنهاء معلّق يحرق مقعد السقف',
      _deleted23 == ['bb23'] and not (_d23.get('events') or {}).get('rj1:S23')
      and _hist23 and _hist23[0].get('outcome') == 'empty_deleted',
      'deleted=%s in_events=%s hist=%s' % (_deleted23, 'rj1:S23' in (_d23.get('events') or {}),
                                           [(h.get('outcome')) for h in _hist23]))
mgr.api, mgr.log, mgr.stop_sender = _pre23_api, _pre23_log, _pre23_stop
mgr.journal.change(_restore23)

# N25: نسخة أرشيف دائمة في dataset خارج الـbucket — الموجة الرابعة مسحت حتى ملفات
# واجهة الـbucket؛ الفشل لا يوقف شيئًا والملف المفقود لا يكسر
_fake_hf = {'calls': []}
class _FakeHF:
    @staticmethod
    def upload_file(path_or_fileobj=None, path_in_repo=None, repo_id=None, repo_type=None, token=None):
        _fake_hf['calls'].append((repo_id, path_in_repo, repo_type, token))
import sys as _sys24, types
_mod24 = types.ModuleType('huggingface_hub')
_mod24.upload_file = _FakeHF.upload_file
_sys24.modules['huggingface_hub'] = _mod24
_orig_tok24 = os.environ.get('HF_UPLOAD_TOKEN')
os.environ['HF_UPLOAD_TOKEN'] = 'TESTTOK24'
_tmp24 = pathlib.Path(tempfile.mkdtemp()) / 'x.mp4'
_tmp24.write_bytes(b'0' * 128)
_app._bucket_upload(_tmp24, 'rj1')
_deadline24 = time.time() + 5
while time.time() < _deadline24 and not _fake_hf['calls']:
    time.sleep(0.05)
_app._bucket_upload(pathlib.Path(tempfile.mkdtemp()) / 'missing.mp4', 'rj1')
_ok24 = bool(_fake_hf['calls']) and _fake_hf['calls'][0][0] == 'Reoo88/Tktktk-archive' \
    and _fake_hf['calls'][0][1] == 'recordings/x.mp4' and _fake_hf['calls'][0][2] == 'dataset' \
    and _fake_hf['calls'][0][3] == 'TESTTOK24'
if _orig_tok24 is None:
    os.environ.pop('HF_UPLOAD_TOKEN', None)
else:
    os.environ['HF_UPLOAD_TOKEN'] = _orig_tok24
check('N25: إقفال MP4 يرفع الأرشيف إلى dataset دائم خارج الـbucket (المستودع/المسار/النوع/التوكن صحيحة) والمفقود لا يكسر شيئًا',
      _ok24, 'calls=%s' % (_fake_hf['calls'],))

# N19: رفض توكن الحساب (403) ⇒ محاولة تلقائية واحدة بـTIKTOK_SESSIONID العام قبل
# إعلان الرفض — تحديث السر العام يشفِي كل الحسابات رغم توكنات الملف القديمة، والحساب
# بلا توكن ملفي يستعمل العام مباشرة من أول نداء
_tk19 = {'cookies': []}
class _Resp19:
    status_code = 200
    def raise_for_status(self):
        pass
    def json(self):
        return {'data': {'ok': True}}
class _Rej19:
    status_code = 403
    def raise_for_status(self):
        _e = OSError('http 403')
        _e.response = type('R', (), {'status_code': 403})()
        raise _e
def _fake_get19(url, params=None, impersonate=None, headers=None, timeout=None):
    cookie = (headers or {}).get('Cookie', '')
    _tk19['cookies'].append(cookie)
    return _Rej19() if 'OLDSID' in cookie and 'GLOBALSID' not in cookie else _Resp19()
_orig_get19 = _app.requests.get
_orig_env19 = os.environ.get('TIKTOK_SESSIONID')
_app.requests.get = _fake_get19
os.environ['TIKTOK_SESSIONID'] = 'GLOBALSID'
try:
    _app._tokens['oldacc'] = 'OLDSID'
    _r19 = _app.get_json('https://www.tiktok.com/api-live/user/room', {'aid': 1988}, 'oldacc')
    _ok19 = _r19.get('data', {}).get('ok') is True and len(_tk19['cookies']) == 2 \
        and 'OLDSID' in _tk19['cookies'][0] and 'GLOBALSID' in _tk19['cookies'][1]
    _tk19['cookies'].clear()
    _app._tokens.pop('oldacc', None)
    _r19b = _app.get_json('https://www.tiktok.com/api-live/user/room', {'aid': 1988}, 'oldacc')
    _ok19b = _r19b.get('data', {}).get('ok') is True and len(_tk19['cookies']) == 1 \
        and 'GLOBALSID' in _tk19['cookies'][0]
finally:
    _app.requests.get = _orig_get19
    if _orig_env19 is None:
        os.environ.pop('TIKTOK_SESSIONID', None)
    else:
        os.environ['TIKTOK_SESSIONID'] = _orig_env19
check('N19: رفض توكن الحساب يرجّح تلقائيًا بـTIKTOK_SESSIONID العام — والحساب بلا توكن ملفي يستعمل العام فورًا',
      _ok19 and _ok19b,
      'fallback=%r direct=%r cookies=%s' % (_ok19, _ok19b, [c[:24] for c in _tk19['cookies']]))


# N20: بعد موت كل الكوكيز — محاولة مجهولة أخيرة: تُرجع الجسد إن كان صالحًا للاكتشاف
# (غير مباشر، أو حي بمصادر مرئية مجهولًا)؛ حيّ بلا مصادر مجهولًا يبقى «رفض الجلسة» الأصلي
_body20 = {'v': 0}
class _Resp20:
    status_code = 200
    def raise_for_status(self):
        pass
    def json(self):
        return _body20
class _Rej20:
    status_code = 403
    def raise_for_status(self):
        _e = OSError('http 403')
        _e.response = type('R', (), {'status_code': 403})()
        raise _e
_calls20 = {'anon': 0}
def _fake_get20(url, params=None, impersonate=None, headers=None, timeout=None):
    if (headers or {}).get('Cookie'):
        return _Rej20()
    _calls20['anon'] += 1
    return _Resp20()
_orig_get20 = _app.requests.get
_orig_env20 = os.environ.pop('TIKTOK_SESSIONID', None)
_app.requests.get = _fake_get20
_app._tokens['old20'] = 'OLDSID'
_URL20 = 'https://www.tiktok.com/api-live/user/room'
try:
    _body20.update({'data': {'liveRoom': {'status': 4}}})
    _r20a = _app.get_json(_URL20, {'aid': 1988}, 'old20')
    _ok20a = _r20a.get('data', {}).get('liveRoom', {}).get('status') == 4 and _calls20['anon'] == 1
    _sd20 = json.dumps({'data': {'origin': {'main': {'flv': 'https://pull/x.flv'}}}})
    _body20.clear()
    _body20.update({'data': {'liveRoom': {'status': 2, 'streamData': {'pull_data': {'stream_data': _sd20}}}}})
    _r20b = _app.get_json(_URL20, {'aid': 1988}, 'old20')
    _ok20b = _r20b.get('data', {}).get('liveRoom', {}).get('status') == 2
    _body20.clear(); _body20.update({'data': {'liveRoom': {'status': 2}}})
    _raise20 = False
    try:
        _app.get_json(_URL20, {'aid': 1988}, 'old20')
    except RuntimeError as e:
        _raise20 = 'رفض الجلسة' in str(e)
finally:
    _app.requests.get = _orig_get20
    _app._tokens.pop('old20', None)
    if _orig_env20 is not None:
        os.environ['TIKTOK_SESSIONID'] = _orig_env20
check('N20: بعد رفض الكوكيز، المحاولة المجهولة تُرجع الرصد الصالح (غير مباشر/حي بمصادر) — وحيّ بلا مصادر مجهولًا يبقى رفض جلسة بلا كذب',
      _ok20a and _ok20b and _raise20,
      'offline=%r live_src=%r no_src_raises=%r' % (_ok20a, _ok20b, _raise20))

# N21: القالب الفخم (V9) — ترحيل '{title}' غير المعدّل تلقائيًا، القوالب المخصصة لا تُمس،
# وعنوان الحدث بلا live_title يبقى اسم المذيع داخل الغلاف الفخم بلا تكرار «بث مباشر»
from youtube_api import Journal as _J21, DEFAULT_SETTINGS as _DS21
_dir21 = tempfile.mkdtemp()
_p21 = pathlib.Path(_dir21) / 'j21.json'
_p21.write_text(json.dumps({'schema': 7, 'instance': 'x', 'enabled': False, 'channel': None,
    'streams': {}, 'events': {}, 'history': [], 'budget': {},
    'settings': {'name': 'ك', 'privacy': 'public', 'title': '{title}', 'description': _DS21['description']}}), encoding='utf-8')
_j21 = _J21(_p21).read()
_custom21 = pathlib.Path(_dir21) / 'j21c.json'
_custom21.write_text(json.dumps({'schema': 7, 'instance': 'y', 'enabled': False, 'channel': None,
    'streams': {}, 'events': {}, 'history': [], 'budget': {},
    'settings': {'name': 'ك', 'privacy': 'public', 'title': 'قالبي {name} الخاص', 'description': 'د'}}), encoding='utf-8')
_j21c = _J21(_custom21).read()
_nick21 = pathlib.Path(_dir21) / 'j21n.json'
_nick21.write_text(json.dumps({'schema': 7, 'instance': 'z', 'enabled': False, 'channel': None,
    'streams': {}, 'events': {}, 'history': [], 'budget': {},
    'settings': {'name': 'ك', 'privacy': 'public', 'title': '{nickname} | بث مباشر | {date} — {time}',
                 'description': 'د'}}), encoding='utf-8')
_j21n = _J21(_nick21).read()
_jb21 = json.loads(json.dumps(mgr.journal.read()))
_pre21_acc = mgr.accounts
_pre21 = mgr.api, mgr._observe, mgr.ensure_stream, mgr.open_event, mgr.drive_event, mgr.probe
_ms21 = getattr(mgr, 'manager_stop', None)
if _ms21 is not None:
    _ms21.set()
    _mt21 = getattr(mgr, 'manager_thread', None)
    if _mt21 is not None and _mt21.is_alive():
        _mt21.join(timeout=5)
def _obs21(u, *a, **k):
    return {'status': 'live', 'room_id': 'R21',
            'sources': [{'url': 'https://x/rj.flv', 'media': 'flv', 'codec': 'h264', 'label': 'hd'}],
            'nickname': u, 'live_title': ''}
mgr.accounts = lambda: ['rj1']
mgr._observe = _obs21
mgr.ensure_stream = lambda u: None
mgr.probe = lambda u, c: {'video': 'h264', 'audio': 'aac'}
mgr.open_event = ym.ManagedYouTube.open_event.__get__(mgr)
mgr.drive_event = lambda key: None
mgr.max_events = 3
mgr._blocked403.pop('rj1:R21', None)
mgr.last_seen_status.pop('rj1', None)
mgr.auth_pause.pop('rj1', None)
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[]))
mgr.tick()
_ev21 = (mgr.journal.read().get('events') or {}).get('rj1:R21') or {}
_t21 = str(_ev21.get('title') or '')
check('N21: القالب الفخم يُرحَّل تلقائيًا (👑+🔴) والعنوان بلا live_title يبقى اسم المذيع داخل الغلاف — والمخصص لا يُمس',
      '👑' in _j21['settings']['title'] and '🔴' in _j21['settings']['title']
      and _j21c['settings']['title'] == 'قالبي {name} الخاص'
      and '👑' in _j21n['settings']['title']
      and _t21.startswith('👑') and '🔴 بث مباشر الآن' in _t21 and 'rj1' in _t21,
      'migrated=%r custom=%r nick=%r title=%r' % (_j21['settings']['title'], _j21c['settings']['title'], _j21n['settings']['title'], _t21))
# N22: سجل البثوث يصل اللوحة بعناوينه — الأحدث أولًا مع معرف البث لرابط المشاهدة
mgr.journal.change(lambda dd: dd.update(history=[
    {'session_key': 'rj1:OLD', 'broadcast_id': 'bOLD', 'title': 'بث قديم', 'username': 'rj1',
     'nickname': 'ر', 'outcome': 'completed', 'ended_at': 1.0},
    {'session_key': 'rj1:NEW', 'broadcast_id': 'bNEW', 'title': 'بث جديد فخم', 'username': 'rj1',
     'nickname': 'ر', 'outcome': 'live_ended', 'ended_at': 2.0}]))
_sn22 = mgr.snapshot()
_h22 = _sn22.get('history') or []
check('N22: /api/youtube يعيد سجل البثوث بعناوين كاملة — الأحدث أولًا وbroadcast_id للرابط',
      _h22 and _h22[0].get('title') == 'بث جديد فخم' and _h22[-1].get('title') == 'بث قديم'
      and _h22[0].get('broadcast_id') == 'bNEW' and _sn22.get('history_count') == 2
      and _h22[0].get('outcome') == 'live_ended',
      'hist=%s count=%s' % ([(h.get('title'), h.get('broadcast_id')) for h in _h22], _sn22.get('history_count')))
mgr.api, mgr._observe, mgr.ensure_stream, mgr.open_event, mgr.drive_event, mgr.probe = _pre21
mgr.accounts = _pre21_acc
mgr.journal.change(lambda dd: dd.clear() or dd.update(_jb21))























check('N4: جلسة سليمة ≥25 ث تُلغي التأجيل وتمسح ملاحظته من اللوحة',
      not mgr.retry_at and _note_now == '' and not mgr.hammer.get('recovered')
      and any('مُسح تأجيل العقاب' in x for x in logs_n4), 'note=%r' % _note_now[:40])
# ملاحظات أخرى (رصد/إنهاء) يجب أن تبقى: المسح مشروط ببادئة كتابتنا فقط
mgr.journal.change(lambda dd: dd['events'].update({'rec:9': dict(dd['events']['rec:9'],
                                                                note='تيك توك أغلق الجلسة')}))
mgr._note_early_exit('recovered', _FeedingSender(), lasted=40.0)
_kept = (mgr.journal.read()['events'].get('rec:9') or {}).get('note')
check('N4: لا تُمسح ملاحظات الغير — البادئة شرط الحذف', _kept == 'تيك توك أغلق الجلسة', repr(_kept)[:40])
mgr.journal.change(lambda dd: dd['events'].pop('rec:9', None))



mgr.log = lambda u, m: logs.append(str(m))
mgr.hammer.clear(); mgr.retry_at.clear(); mgr.retry.clear()
check('N: الرمز يُقرأ من حالة المرسل (relay.last_exit_code) لا من السجل فقط',
      hasattr(relay.Sender(FF, lambda u, m: None, lambda u: {}), 'last_exit_code'), '')

# N23: history + YouTube still live must resume the same broadcast, not skip forever.
_pre23 = mgr.accounts, mgr._observe, mgr.ensure_stream, mgr.drive_event, mgr.remote, mgr.open_event
_logs23 = []
_orig_log23 = mgr.log
mgr.log = lambda u, m: _logs23.append(str(m))
def _obs23(u, *a, **k):
    return {'status': 'live', 'room_id': 'R23',
            'sources': [{'url': 'https://x/a.flv', 'media': 'flv', 'codec': 'h264', 'label': 'hd'}],
            'nickname': u, 'live_title': ''}
def _remote23(resource, rid):
    return {'id': rid,
            'snippet': {'title': 'same', 'description': 'x\nQUSAI-AUTO-x-EVENT-abc\n'},
            'status': {'lifeCycleStatus': 'live', 'privacyStatus': 'public'},
            'contentDetails': {'boundStreamId': 's1'}}
mgr.accounts = lambda: ['u23']
mgr._observe = _obs23
mgr.ensure_stream = lambda u: None
mgr.drive_event = lambda key: None
mgr.remote = _remote23
def _no_open23(u, o):
    raise RuntimeError('open_event should not run on resume')
mgr.open_event = _no_open23
mgr.journal.change(lambda dd: dd.update(enabled=True, events={}, history=[{
    'session_key': 'u23:R23', 'broadcast_id': 'b23', 'title': 'same', 'username': 'u23',
    'nickname': 'u23', 'outcome': 'live_ended', 'ended_at': time.time() - 400}]))
mgr.tick()
_ev23 = (mgr.journal.read().get('events') or {}).get('u23:R23') or {}
_h23 = mgr.journal.read().get('history') or []
check('N23: جلسة أُرشفت ويوتيوب ما زال live تُستأنف على نفس الحدث ولا تُترك بلا ناشر',
      _ev23.get('broadcast_id') == 'b23' and _ev23.get('bound') is True
      and _ev23.get('youtube_status') == 'live'
      and not any(h.get('session_key') == 'u23:R23' for h in _h23)
      and any('resume same YouTube event' in x for x in _logs23),
      'ev=%s hist=%s logs=%s' % ({k: _ev23.get(k) for k in ('broadcast_id', 'bound', 'youtube_status', 'phase')},
                                 [(h.get('session_key'), h.get('broadcast_id')) for h in _h23], _logs23[-4:]))
mgr.accounts, mgr._observe, mgr.ensure_stream, mgr.drive_event, mgr.remote, mgr.open_event = _pre23
mgr.log = _orig_log23
mgr.journal.change(lambda dd: dd.update(
    events={k: v for k, v in (dd.get('events') or {}).items() if k != 'u23:R23'},
    history=[h for h in (dd.get('history') or []) if h.get('session_key') != 'u23:R23']))

check('M: لا ffmpeg/قارئ متسرب بعد كل الجلسات', not stray, 'stray=%s' % stray[:1])
srv.shutdown()
print()
bad = [t for t, ok in results if not ok]
print('النتيجة: %d/%d' % (len(results) - len(bad), len(results)))
if bad:
    print('فاشل:', bad)
    sys.exit(1)