Persist live stream transition diagnostics

AUDIT_AR.md

@@ -1,6 +1,6 @@
 # تقرير تدقيق وإصلاح Space `Reoo88/On`
 
-**التاريخ:** 2026-10-07 — **النسخة المنشورة قبل هذا التغيير:** `0a92b5b` — **النسخة الجديدة المرشحة:** `relay-v2.19-config-refresh`
+**التاريخ:** 2026-10-07 — **النسخة المنشورة قبل هذا التغيير:** `fa14eaf` — **النسخة الجديدة المرشحة:** `relay-v2.20-stream-monitor`
 
 > هذا تقرير تقني وتشغيلي، وليس وعدًا بمنع كل عطل مستقبلي. TikTok وYouTube وHugging Face خدمات خارجية تتغير دون تحكم المشروع؛ الهدف هو منع الأعطال المعروفة، كشف الأعطال الجديدة بسرعة، والتعافي الآمن دون إنشاء أحداث مكررة.
 
@@ -185,6 +185,22 @@ Enhanced RTMP v2 يعرّف multitrack، لكن ذلك لا يجعل track وا
 
 أي أن فيديو H.264 ينسخ، وفيديو H.265 ينسخ مع إعلان `hvc1`، ولا يوجد فك/scale/x264 في هذا الإصلاح. يبقى AAC normalization منفصلًا ولا يغيّر إطارات الفيديو. مسار x264 التاريخي ما زال في المصدر كخيار يدوي فقط؛ رفع `YOUTUBE_LIGHT_MAX` هو opt-in صريح.
 
+#### سجل أحداث البث الدائم
+
+أضيف `stream_monitor.py` ويكتب مصفوفة JSON محدودة افتراضيًا إلى `logs/stream_events.json` تحت `DATA_DIR`/Storage Bucket؛ يضيف كل حدث بكلفة ثابتة تقريبًا ويضغط الملف دوريًا عبر atomic replace بدل إعادة كتابة الملف كاملًا مع كل health poll. كل سجل يحوي `timestamp` بتوقيت UTC و`sequence` و`broadcaster` و`event` و`details`. لا تُكتب فيديوهات أو access units أو URLs أو stream keys أو cookies/tokens؛ المرشح يحذف هذه المفاتيح قبل الكتابة، والملف `0600` ويحتفظ بآخر 10,000 حدث فقط.
+
+الأحداث المتصلة فعليًا بمصادرها:
+
+- `CODEC_CHANGE`: من مقارنة FourCC في SequenceStart داخل `FLVBridge`.
+- `RESOLUTION_CHANGE`: من parsing بايتي لـSPS داخل avcC/hvcC، من دون فك إطار.
+- `BITRATE_CHANGE`: نافذة أربع ثوانٍ من أحجام video tags وDTS، بعتبة نسبة 35% وفارق 50 kbit/s لتجنب ضجيج VBR العادي.
+- `KEYFRAME_CHANGE`: من keyframe flag وفرق DTS بين الإطارات المفتاحية، مع rate limit.
+- `RECONNECT`: يميز `scope=tiktok_http` مع `outbound_rtmp_reused=true` عن `scope=youtube_rtmp` عند فتح ناشر خرج ثانٍ فعليًا.
+- `CONNECTION_ERROR`: من فشل الفتح أو حارس source/output أو خروج FFmpeg على جهة YouTube.
+- `YOUTUBE_HEALTH`: من `status.healthStatus` وقائمة configuration issues كل 20 ثانية (الحد الأدنى 15 ثانية).
+
+يمكن للمشغل قراءة السجل من endpoint محمي بالمصادقة: `GET /api/stream-events?limit=200&broadcaster=user`. هذا السجل هو دليل اختبار PK والخلفية والإشعارات؛ لكنه لا يستبدل مشاهدة عميل YouTube فعلية لأن العطل الأصلي كان player-side بينما VOD سليم.
+
 #### المراجع التي حُكم بها
 
 - Enhanced RTMP v2 الرسمي: <https://veovera.org/docs/enhanced/enhanced-rtmp-v2.pdf>
@@ -208,13 +224,13 @@ YouTube يسرد H.265 ضمن codecs المقبولة حاليًا لـRTMP/RTMP
 
 ## 5. اختبارات حقيقية ومحاكاة واقعية
 
-نفذ `verify.sh` بعد إصلاح config passthrough، والنتيجة **314/314** فحصًا ناجحًا:
+نفذ `verify.sh` بعد إصلاح config passthrough وإضافة المراقبة، والنتيجة **325/325** فحصًا ناجحًا:
 
 - فحص بنيوي واحد يؤكد غياب التسجيل وواجهة تنزيل الفيديو.
 - 25 فحص تزامن وعزل حسابات.
-- 28 regression للدخول والخصوصية وحدود الكوكي وفصل عطل RTMPS عن جوع المصدر.
+- 29 regression للدخول والخصوصية وحدود الكوكي وفصل عطل RTMPS عن جوع المصدر، ومنها API سجل الأحداث المحمي.
 - 83 فحص relay/API/UI، ومنها غياب أي runtime caller لـ`FormatChanged` وافتراضات Docker الخاصة بالنسخ.
-- 23 فحصًا جديدًا خاصًا بـFLVBridge: AVC/HEVC، CTS/DTS، reconnect، multi-SPS/PPS، transition مرصود، وتمرير FFmpeg الحقيقي للسجلين ولـaccess units بالبايت.
+- 33 فحصًا خاصًا بـFLVBridge/StreamMonitor: AVC/HEVC، CTS/DTS، reconnect، multi-SPS/PPS، parsing دقة SPS، codec/resolution/bitrate/keyframe events، JSON وصلاحية `0600` وتنقية الأسرار، وتمرير FFmpeg الحقيقي للسجلين ولـaccess units بالبايت.
 - 40 فحص completion، منها FFmpeg حقي��ي ومقارنة مهلات الصمت 12/30/45 ثانية.
 - 38 فحص watchdog بشبكة HTTP محلية وFFmpeg حقيقي، ومنها أن HEVC يستخدم `-c:v copy` ولا يدخل `libx264`.
 - 76 فحص lifecycle، منها source stall → cleanup → re-observe → session جديدة، استئناف حدث YouTube نفسه، حدود التزامن، 403، offline، HLS fallback، وعدم تراكم العمليات.
@@ -242,6 +258,8 @@ YouTube يسرد H.265 ضمن codecs المقبولة حاليًا لـRTMP/RTMP
 | `YOUTUBE_FORCE_TRANSCODE_USERS` | احذف المتغير | لا يُجبر أحمد أو Strong على `matchgop` |
 | `YOUTUBE_ENHANCED_RTMP` | `auto` | يعلن `hvc1` فقط عندما يكون payload المنسوخ HEVC، ولا يعلنه كذبًا على H.264 |
 | `YOUTUBE_VIDEO_MODE` | `original` | يبقي H.264 وH.265 ضمن المرشحين؛ يفضل H.264 مفيدًا عند وجوده ويقبل HEVC عند غيابه |
+| `YOUTUBE_HEALTH_INTERVAL` | `20` | يسجل healthStatus أثناء سيناريو الاختبار؛ راقب quota عند 3–5 أحداث متزامنة |
+| `STREAM_EVENT_MAX` | `10000` | يمنع نمو `stream_events.json` بلا حد |
 | `YOUTUBE_DAILY_BUDGET` | اترك القيمة الحالية المدروسة | منع استنزاف API |
 
 في سقفي الناشرين/الأحداث تعني `0` بلا سقف، فلا تضعها بدل `4`. أما `YOUTUBE_LIGHT_MAX=0` فمعناه المقصود **صفر عمليات x264**. Variables في الـSpace تتقدم على Dockerfile، لذا يجب حذف القيم القديمة `2` وقائمة الحسابات المجبرة عند النشر.

Dockerfile

@@ -5,7 +5,7 @@ RUN ffmpeg -hide_banner -h protocol=rtmps 2>&1 | grep -q rtmp_enhanced_codecs
 WORKDIR /app
 COPY requirements.txt .
 RUN pip install --no-cache-dir -U -r requirements.txt
-COPY app.py relay.py youtube_api.py youtube_manager.py flv_bridge.py tiktok_auth.py ingest.py index.html ./
+COPY app.py relay.py youtube_api.py youtube_manager.py flv_bridge.py stream_monitor.py tiktok_auth.py ingest.py index.html ./
 COPY verify.sh verify_tokens.py probe_live.py test_watchdog.py test_lifecycle.py test_relay.py test_concurrency.py test_flv_bridge.py ./
 
 # TikTok Live Monitor → Direct Relay → YouTube. No local video files.
@@ -35,7 +35,8 @@ ENV TZ=Asia/Baghdad \
     YOUTUBE_COPY_RECOVERY_SECONDS=60 \
     YOUTUBE_X264_PRESET=ultrafast \
     YOUTUBE_FIFO_TIMESHIFT=0 \
-    YOUTUBE_HEALTH_INTERVAL=60 \
+    YOUTUBE_HEALTH_INTERVAL=20 \
+    STREAM_EVENT_MAX=10000 \
     YOUTUBE_OBS_HOLD_SECONDS=900 \
     YOUTUBE_SOURCE_RETRY=15 \
     YOUTUBE_FORCE_GAP=5 \

README.md

@@ -8,7 +8,7 @@ pinned: false
 app_port: 7860
 ---
 
-# Reoo88/On — relay-v2.19-config-refresh
+# Reoo88/On — relay-v2.20-stream-monitor
 
 رصد TikTok LIVE → FFmpeg → YouTube RTMPS.
 **لا تسجيل محلي، لا أرشيف فيديو، ولا ملفات فيديو على القرص.**
@@ -43,8 +43,10 @@ TikTok H.264 أو H.265/FLV
 - تغيّر `AVCDecoderConfigurationRecord` أو `HEVCDecoderConfigurationRecord` لا يرمي `FormatChanged` ولا يقتل FFmpeg: يمر السجل الجديد قبل الإطار الذي يحتاجه عبر RTMPS المفتوح نفسه.
 - `FLVBridge` يحافظ على بايتات الصورة وCTS، ويصل أزمنة DTS عند تجديد رابط HTTP-FLV من دون رأس FLV ثانٍ. سجل جديد كامل يحل محل السابق؛ لا تُخلط مجموعات قد تحمل IDs متعارضة.
 - FFmpeg 7 ينقل تغيير extradata من demuxer إلى muxer ويعيد كتابة SequenceStart؛ يثبت الاختبار الحتمي ذلك لـH.264 وH.265.
+- يسجل `stream_monitor.py` أحداث `CODEC_CHANGE` و`RESOLUTION_CHANGE` و`BITRATE_CHANGE` و`KEYFRAME_CHANGE` و`RECONNECT` و`CONNECTION_ERROR` و`YOUTUBE_HEALTH` في `logs/stream_events.json` داخل Storage Bucket، من دون وسائط أو روابط أو مفاتيح.
+- يمكن قراءة آخر الأحداث من المسار المحمي `GET /api/stream-events?limit=200&broadcaster=user`، والملف محدود افتراضيًا إلى 10,000 سجل وصلاحيته `0600`.
 - إذا بقي خيط الإشراف حيًا ومات FFmpeg لسبب آخر، يعيد watchdog تدويره بدل بقاء YouTube في `noData`.
-- روابط المصدر الموقعة تُرصد من جديد، مع حارس صمت افتراضي 12 ثانية. الوضع الافتراضي يمنع تحويل الفيديو؛ خيارات `matchgop` القديمة لا تعمل إلا إذا فعّلها المشغل ورفع `YOUTUBE_LIGHT_MAX` عمدًا.
+- روابط المصدر الموقعة تُرصد من جديد، مع حارس صمت افتراضي 12 ثانية. تُسحب صحة YouTube كل 20 ثانية أثناء الحدث وتدخل السجل. الوضع الافتراضي يمنع تحويل الفيديو؛ خيارات `matchgop` القديمة لا تعمل إلا إذا فعّلها المشغل ورفع `YOUTUBE_LIGHT_MAX` عمدًا.
 - الحماية من تكرار مذيع واحد ومن خلط مذيعين على مفتاح RTMPS بيئي واحد مفعلة.
 - الحالة والتوكنات تحفظ في Storage Bucket المركب على `/data`؛ كوكيز TikTok المخزنة تُكتب بصلاحية `0600`.
 - كوكي TikTok لا يُرسل إلى روابط CDN الموقعة خارج `tiktok.com`؛ طلبات الوسائط العامة تستعمل `Referer` فقط.

app.py

@@ -47,6 +47,7 @@ try:
     ROOT.mkdir(parents=True,exist_ok=True)
 except OSError as e:
     logging.error('cannot create %s (%s) — starting read-only', ROOT, type(e).__name__)
+os.environ.setdefault('STREAM_EVENTS_PATH',str(ROOT/'logs'/'stream_events.json'))
 
 def _ensure_root():
     try:
@@ -441,6 +442,7 @@ def rank_sources(source):
 
 from youtube_manager import ManagedYouTube
 from relay import Sender, SourceStalled, RelayStartError
+from stream_monitor import read_events
 def enabled_accounts():
     with lock:return [u for u,t in tasks.items() if t.get('enabled')]
 youtube=ManagedYouTube(FFMPEG,observe_session,log,enabled_accounts,ROOT/'youtube-api-state.json',headers_for=tiktok_headers)
@@ -906,6 +908,10 @@ def health_detail():
 
 @app.get('/api/youtube')
 def youtube_state():return youtube.snapshot()
+@app.get('/api/stream-events')
+def stream_events(limit:int=200,broadcaster:str=''):
+    rows=read_events(limit,broadcaster.strip().lstrip('@') or None)
+    return {'count':len(rows),'events':rows,'file':'logs/stream_events.json'}
 @app.post('/api/youtube/check')
 def youtube_check():
     try:return {'channel':youtube.check()}

flv_bridge.py

@@ -23,6 +23,11 @@ References:
 - FFmpeg 7.1 ``flvdec.c``/``flvenc.c`` NEW_EXTRADATA handling
 - GStreamer bugs 797255/797256 (stream headers/caps changes in streamable FLV)
 """
+import os
+from collections import deque
+from stream_monitor import log_event, video_dimensions
+
+
 class FormatChanged(Exception):
     """Compatibility name; decoder-configuration changes no longer raise it."""
 
@@ -50,7 +55,7 @@ def _set_tag_ts(header, ts):
 
 
 class FLVBridge:
-    def __init__(self,detect_format_change=False,allow_raise=None):
+    def __init__(self,detect_format_change=False,allow_raise=None,user='',event_logger=None):
         """Create a streaming bridge.
 
         ``detect_format_change`` and ``allow_raise`` remain accepted for callers from older
@@ -60,13 +65,43 @@ class FLVBridge:
         self.buffer=bytearray();self.header_done=False;self.converted_tags=0
         self.detect_format_change=bool(detect_format_change)  # compatibility/observability only
         self.allow_raise=allow_raise
-        self.video_cfg=None;self.video_codec=None;self.format_changes=0;self.codec_changes=0
+        self.user=str(user or 'system');self.event_logger=event_logger or log_event
+        self.video_cfg=None;self.video_codec=None;self.video_size=None
+        self.format_changes=0;self.codec_changes=0
         self.skip_file_header=False
         self.skip_script=False
         self.skip_same_cfg=False
         self.last_out_ts=None
         self.ts_bias=0
         self.ts_bias_pending=False
+        self._video_samples=deque();self._bitrate_kbps=None;self._bitrate_check_ts=None
+        self._last_keyframe_ts=None;self._keyframe_interval=None;self._keyframe_event_ts=None
+    def _event(self,event,details):
+        try:self.event_logger(self.user,event,details)
+        except Exception:pass
+    def _observe_frame(self,ts,size,keyframe):
+        samples=self._video_samples;samples.append((ts,size))
+        while samples and ts-samples[0][0]>4000:samples.popleft()
+        if len(samples)>=2 and samples[-1][0]-samples[0][0]>=3000 and (
+                self._bitrate_check_ts is None or ts-self._bitrate_check_ts>=2000):
+            span=max(1,samples[-1][0]-samples[0][0]);kbps=round(sum(x[1] for x in samples)*8/span,1)
+            old=self._bitrate_kbps;self._bitrate_kbps=kbps;self._bitrate_check_ts=ts
+            ratio=abs(kbps-old)/max(old,1) if old is not None else 0
+            threshold=max(0.1,float(os.getenv('STREAM_BITRATE_CHANGE_RATIO','0.35')))
+            minimum=max(1.0,float(os.getenv('STREAM_BITRATE_CHANGE_KBPS','50')))
+            if old is not None and ratio>=threshold and abs(kbps-old)>=minimum:
+                self._event('BITRATE_CHANGE',{'old_kbps':old,'new_kbps':kbps,
+                    'change_percent':round((kbps-old)*100/max(old,1),1),'source_dts_ms':ts})
+        if keyframe:
+            if self._last_keyframe_ts is not None and ts>self._last_keyframe_ts:
+                interval=ts-self._last_keyframe_ts;old=self._keyframe_interval
+                if old is not None and abs(interval-old)>=max(500,old*0.30) and (
+                        self._keyframe_event_ts is None or ts-self._keyframe_event_ts>=5000):
+                    self._event('KEYFRAME_CHANGE',{'old_interval_ms':round(old,1),
+                        'new_interval_ms':interval,'source_dts_ms':ts})
+                    self._keyframe_event_ts=ts
+                self._keyframe_interval=interval if old is None else old*0.7+interval*0.3
+            self._last_keyframe_ts=ts
     def reconnect(self):
         """Prepare to splice a new TikTok FLV into the same FFmpeg stdin.
 
@@ -79,6 +114,8 @@ class FLVBridge:
         self.skip_script=True
         self.skip_same_cfg=True
         self.ts_bias_pending=self.last_out_ts is not None
+        self._event('RECONNECT',{'scope':'tiktok_http','outbound_rtmp_reused':True,
+                                 'last_output_dts_ms':self.last_out_ts})
     def feed(self,chunk):
         self.buffer.extend(chunk);output=bytearray()
         if not self.header_done:
@@ -96,7 +133,7 @@ class FLVBridge:
             if len(self.buffer)-offset<total:break
             body=bytes(self.buffer[offset+11:offset+11+size]);tag_type=header[0]
             if tag_type & 0x20:raise ValueError('Filtered/encrypted FLV tag is unsupported')
-            kind = tag_type & 31
+            kind = tag_type & 31;raw_ts=_tag_ts(header)
             if kind==18 and self.skip_script:
                 offset+=total
                 continue
@@ -114,13 +151,31 @@ class FLVBridge:
             if kind==9:
                 signature=_video_signature(body)
                 if signature is not None:
-                    previous=self.video_cfg
+                    previous=self.video_cfg;old_size=self.video_size
                     changed=previous is not None and signature!=previous
+                    codec=signature[:4].decode('ascii','replace')
+                    new_size=video_dimensions(codec,signature[4:])
                     if changed:
                         self.format_changes+=1
-                        if signature[:4]!=previous[:4]:self.codec_changes+=1
-                    self.video_cfg=signature
-                    self.video_codec=signature[:4].decode('ascii','replace')
+                        old_codec=previous[:4].decode('ascii','replace')
+                        codec_changed=codec!=old_codec
+                        if codec_changed:
+                            self.codec_changes+=1;self._video_samples.clear()
+                            self._bitrate_kbps=None;self._last_keyframe_ts=None;self._keyframe_interval=None
+                        details={'codec':codec,'previous_codec':old_codec,'source_dts_ms':raw_ts,
+                                 'old_config_bytes':len(previous)-4,'new_config_bytes':len(signature)-4,
+                                 'outbound_rtmp_reused':True}
+                        self._event('DECODER_CONFIG_CHANGE',details)
+                        if codec_changed:self._event('CODEC_CHANGE',details)
+                        if old_size and new_size and old_size!=new_size:
+                            self._event('RESOLUTION_CHANGE',{'codec':codec,
+                                'old_width':old_size[0],'old_height':old_size[1],
+                                'new_width':new_size[0],'new_height':new_size[1],
+                                'source_dts_ms':raw_ts,'outbound_rtmp_reused':True})
+                    elif previous is None:
+                        self._event('STREAM_CONFIG',{'codec':codec,'width':new_size[0] if new_size else None,
+                            'height':new_size[1] if new_size else None,'config_bytes':len(signature)-4})
+                    self.video_cfg=signature;self.video_codec=codec;self.video_size=new_size
                     # Only the first identical record after reconnect is redundant. A changed
                     # record is emitted now, before any dependent coded frame in this input.
                     if self.skip_same_cfg:
@@ -128,7 +183,6 @@ class FLVBridge:
                         if previous is not None and not changed:
                             offset+=total
                             continue
-            raw_ts=_tag_ts(header)
             if self.ts_bias_pending:
                 self.ts_bias=(self.last_out_ts or 0)+40-raw_ts
                 self.ts_bias_pending=False
@@ -137,6 +191,11 @@ class FLVBridge:
             # Equal DTS across audio/video (or config/frame) is valid and must not be rewritten.
             # Only reconnect() adds one constant bias; last_out_ts tracks the emitted maximum.
             self.last_out_ts=out_ts if self.last_out_ts is None else max(self.last_out_ts,out_ts)
+            if kind==9 and body:
+                enhanced=bool(body[0]&0x80);packet=(body[0]&15) if enhanced else (body[1] if len(body)>1 else -1)
+                codec_id=body[0]&15
+                coded=(enhanced and packet in (1,3)) or (not enhanced and codec_id==7 and packet==1)
+                if coded:self._observe_frame(out_ts,len(body),((body[0]>>4)&7)==1)
             header[1:4]=len(body).to_bytes(3,'big')
             output.extend(header);output.extend(body);output.extend((11+len(body)).to_bytes(4,'big'))
             offset+=total

ingest.py

@@ -6,7 +6,7 @@ import os, threading, time
 from datetime import datetime
 from zoneinfo import ZoneInfo
 
-APP_VERSION = 'relay-v2.19-config-refresh'
+APP_VERSION = 'relay-v2.20-stream-monitor'
 
 
 def env_rtmp_destination():

relay.py

@@ -4,6 +4,7 @@ from collections import deque
 from urllib.parse import urlparse
 from curl_cffi import requests
 from flv_bridge import FLVBridge
+from stream_monitor import log_event, note_rtmp_open
 
 
 def source_http_headers(url, user=None, headers_for=None):
@@ -865,7 +866,7 @@ class Sender:
         hls = (candidate or {}).get('media') == 'hls'
         # Replacement SPS/PPS or VPS/SPS/PPS records are forwarded through this bridge and the
         # same FFmpeg/RTMPS publisher. They are not a reason to select x264 or restart output.
-        bridge = FLVBridge()
+        bridge = FLVBridge(user=user)
         response = None
         p = None
         error_reader = None
@@ -879,7 +880,8 @@ class Sender:
                 'src_metered': False,
                 'force_stop': False, 'timeouts': 0, 'orphan': False, 'stable_notified': False, 'max_seconds': max_seconds,
                 'stall_reason': '', 'spawned': False, 'spawn_error': '',
-                'writing': False, 'output_failure': False, 'ff_last': time.monotonic()}
+                'writing': False, 'output_failure': False, 'connection_event': False,
+                'ff_last': time.monotonic()}
         self.set(status='الاتصال بيوتيوب', source=candidate['label'], output_seconds=0,
                  last_output_at=None, stalled=False, last_source_at=sess['last'], source_bytes=0)
         with self.lock:
@@ -901,6 +903,9 @@ class Sender:
                                      stderr=subprocess.PIPE, bufsize=0)
             sess['spawned'] = True
             sess['spawned_at'] = time.monotonic()
+            if destination.startswith(('rtmp://','rtmps://')):
+                note_rtmp_open(user,{'mode':self.active_mode or self.send_mode(),
+                                     'codec':'h265' if self.hevc_payload(candidate) else 'h264'})
             with self.lock:
                 self.process = p
             self.log(user, '[RELAY] بدء FFmpeg ← %s (%s؛ mode=%s)' % (candidate['label'],
@@ -1064,11 +1069,15 @@ class Sender:
             if p is None:
                 sess['spawn_error'] = 'تعذر بدء FFmpeg/المصدر (%s)' % type(e).__name__
                 self.log(user, '[RELAY] فشل بدء الإرسال: %s' % sess['spawn_error'])
+            log_event(user,'CONNECTION_ERROR',{'scope':self.last_failure_side or 'startup',
+                'error_type':type(e).__name__,'stage':'open'});sess['connection_event']=True
             raise
         except SourceStalled as e:
             sess['stall_reason'] = str(e)
             if not self.last_failure_side:
                 self.last_failure_side = 'source'
+            log_event(user,'CONNECTION_ERROR',{'scope':self.last_failure_side,
+                'error_type':type(e).__name__,'reason':str(e)[:300]});sess['connection_event']=True
             tag = 'خرج الإرسال' if self.last_failure_side == 'output' else 'المصدر'
             self.log(user, '[WATCHDOG] توقف %s: %s' % (tag, e))
             raise
@@ -1086,6 +1095,9 @@ class Sender:
                 self.set(status='', stalled=False, note='لم يبدأ FFmpeg أصلًا')
                 raise RelayStartError(sess.get('spawn_error') or 'تعذر بدء جلسة الإرسال')
             _why = sess.get('stall_reason') or sess.get('error') or 'إيقاف عادي'
+            if self.last_failure_side=='output' and not sess.get('connection_event'):
+                log_event(user,'CONNECTION_ERROR',{'scope':'youtube_rtmp','reason':_why[:300],
+                    'ffmpeg_exit_code':p.poll()});sess['connection_event']=True
             self._teardown(user, p, response, sess, stop, _why)
             if error_reader:
                 error_reader.join(timeout=1)

stream_monitor.py

@@ -0,0 +1,220 @@
+"""Persistent, bounded diagnostics for byte-copy live streams.
+
+Events are operational metadata only: no media, URLs, stream keys, cookies, or
+encoded payloads are written. The JSON array lives under DATA_DIR/logs, appends
+one compact object per event, and is periodically compacted with atomic replace.
+"""
+import json
+import os
+import re
+import threading
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+class _Bits:
+    def __init__(self,data):
+        self.data=data;self.bit=0
+    def read(self,n):
+        if n<0 or self.bit+n>len(self.data)*8:raise ValueError('truncated bitstream')
+        value=0
+        for _ in range(n):
+            value=(value<<1)|((self.data[self.bit>>3]>>(7-(self.bit&7)))&1);self.bit+=1
+        return value
+    def ue(self):
+        zeros=0
+        while self.read(1)==0:
+            zeros+=1
+            if zeros>31:raise ValueError('invalid Exp-Golomb value')
+        return (1<<zeros)-1+(self.read(zeros) if zeros else 0)
+    def se(self):
+        value=self.ue()
+        return (value+1)//2 if value&1 else -(value//2)
+
+
+def _rbsp(nal):
+    out=bytearray();zeros=0
+    for byte in nal:
+        if zeros>=2 and byte==3:
+            zeros=0;continue
+        out.append(byte)
+        zeros=zeros+1 if byte==0 else 0
+    return bytes(out)
+
+
+def _skip_h264_scaling(bits,size):
+    last=next_scale=8
+    for j in range(size):
+        if next_scale:
+            next_scale=(last+bits.se()+256)%256
+        last=next_scale or last
+
+
+def _h264_sps_size(nal):
+    data=_rbsp(nal[1:]);b=_Bits(data)
+    profile=b.read(8);b.read(8);b.read(8);b.ue()
+    chroma=1;separate=0
+    if profile in (44,83,86,100,110,118,122,128,138,139,134,135,144,244):
+        chroma=b.ue()
+        if chroma==3:separate=b.read(1)
+        b.ue();b.ue();b.read(1)
+        if b.read(1):
+            for i in range(8 if chroma!=3 else 12):
+                if b.read(1):_skip_h264_scaling(b,16 if i<6 else 64)
+    b.ue();poc=b.ue()
+    if poc==0:b.ue()
+    elif poc==1:
+        b.read(1);b.se();b.se()
+        for _ in range(b.ue()):b.se()
+    b.ue();b.read(1)
+    width_mbs=b.ue()+1;height_maps=b.ue()+1
+    frame_only=b.read(1)
+    if not frame_only:b.read(1)
+    b.read(1)
+    crop=(0,0,0,0)
+    if b.read(1):crop=tuple(b.ue() for _ in range(4))
+    chroma_array=0 if separate else chroma
+    sub_w=(1,2,2,1)[chroma_array];sub_h=(1,2,1,1)[chroma_array]
+    unit_x=1 if chroma_array==0 else sub_w
+    unit_y=(2-frame_only)*(1 if chroma_array==0 else sub_h)
+    width=width_mbs*16-(crop[0]+crop[1])*unit_x
+    height=(2-frame_only)*height_maps*16-(crop[2]+crop[3])*unit_y
+    return (width,height) if width>0 and height>0 else None
+
+
+def _hevc_profile_tier_level(bits,max_sub_layers):
+    bits.read(96)
+    profiles=[];levels=[]
+    for _ in range(max_sub_layers):profiles.append(bits.read(1));levels.append(bits.read(1))
+    if max_sub_layers:
+        for _ in range(max_sub_layers,8):bits.read(2)
+    for profile,level in zip(profiles,levels):
+        if profile:bits.read(88)
+        if level:bits.read(8)
+
+
+def _hevc_sps_size(nal):
+    data=_rbsp(nal[2:]);b=_Bits(data)
+    b.read(4);max_sub=b.read(3);b.read(1)
+    _hevc_profile_tier_level(b,max_sub)
+    b.ue();chroma=b.ue()
+    separate=b.read(1) if chroma==3 else 0
+    width=b.ue();height=b.ue();crop=(0,0,0,0)
+    if b.read(1):crop=tuple(b.ue() for _ in range(4))
+    chroma_array=0 if separate else chroma
+    sub_w=(1,2,2,1)[chroma_array];sub_h=(1,2,1,1)[chroma_array]
+    width-=(crop[0]+crop[1])*sub_w;height-=(crop[2]+crop[3])*sub_h
+    return (width,height) if width>0 and height>0 else None
+
+
+def video_dimensions(codec,config):
+    """Read coded width/height from avcC or hvcC; malformed records return None."""
+    try:
+        if codec=='avc1':
+            if len(config)<7:return None
+            count=config[5]&31;offset=6
+            for _ in range(count):
+                size=int.from_bytes(config[offset:offset+2],'big');offset+=2
+                nal=config[offset:offset+size];offset+=size
+                if nal and nal[0]&31==7:return _h264_sps_size(nal)
+            return None
+        if codec=='hvc1':
+            if len(config)<23:return None
+            offset=23
+            for _ in range(config[22]):
+                nal_type=config[offset]&63;count=int.from_bytes(config[offset+1:offset+3],'big');offset+=3
+                for _ in range(count):
+                    size=int.from_bytes(config[offset:offset+2],'big');offset+=2
+                    nal=config[offset:offset+size];offset+=size
+                    if nal_type==33:return _hevc_sps_size(nal)
+    except (IndexError,ValueError):
+        return None
+    return None
+
+
+class StreamMonitor:
+    def __init__(self,path=None):
+        root=os.getenv('DATA_DIR') or ('/data/tiktok-relay' if Path('/data').is_dir() else './storage')
+        base=Path(root)
+        self.path=Path(path or os.getenv('STREAM_EVENTS_PATH') or base/'logs/stream_events.json')
+        try:self.maximum=max(100,min(int(os.getenv('STREAM_EVENT_MAX','10000')),100000))
+        except (TypeError,ValueError):self.maximum=10000
+        self.lock=threading.RLock();self.events=[];self.sequence=0;self.rtmp_opens={}
+        self._load()
+    def _load(self):
+        try:
+            data=json.loads(self.path.read_text())
+            if isinstance(data,list):
+                self.events=[x for x in data[-self.maximum:] if isinstance(x,dict)]
+                self.sequence=max((int(x.get('sequence') or 0) for x in self.events),default=0)
+        except (OSError,ValueError,TypeError):
+            pass
+    @staticmethod
+    def _safe(value,depth=0):
+        if depth>4:return '…'
+        if isinstance(value,dict):
+            return {str(k)[:80]:StreamMonitor._safe(v,depth+1) for k,v in value.items()
+                    if not any(x in str(k).lower() for x in ('url','token','cookie','key','auth'))}
+        if isinstance(value,(list,tuple)):return [StreamMonitor._safe(x,depth+1) for x in value[:50]]
+        if isinstance(value,str):
+            value=re.sub(r'(?i)\b(?:rtmps?|https?)://\S+','«رابط-مخفي»',value)
+            value=re.sub(r'(?i)(sessionid|token|stream[_ -]?key|authorization)\s*[=:]\s*[^\s;&]+',
+                         r'\1=«مخفي»',value)
+            return value[:500]
+        if isinstance(value,(int,float,bool)) or value is None:return value
+        return StreamMonitor._safe(str(value),depth+1)
+    def _rewrite(self):
+        self.path.parent.mkdir(parents=True,exist_ok=True)
+        tmp=self.path.with_name(self.path.name+'.tmp')
+        tmp.write_text(json.dumps(self.events,ensure_ascii=False,separators=(',',':')))
+        try:os.chmod(tmp,0o600)
+        except OSError:pass
+        os.replace(tmp,self.path)
+    def _append(self,item):
+        encoded=json.dumps(item,ensure_ascii=False,separators=(',',':')).encode()
+        try:
+            with self.path.open('r+b') as fh:
+                fh.seek(-1,os.SEEK_END)
+                if fh.read(1)!=b']':raise ValueError('invalid JSON tail')
+                fh.seek(-1,os.SEEK_END)
+                fh.write((b',' if len(self.events)>1 else b'')+encoded+b']');fh.truncate()
+            try:os.chmod(self.path,0o600)
+            except OSError:pass
+        except (OSError,ValueError):
+            self._rewrite()
+    def log(self,broadcaster,event,details=None):
+        try:
+            with self.lock:
+                self.sequence+=1
+                item={'timestamp':datetime.now(timezone.utc).isoformat(timespec='milliseconds').replace('+00:00','Z'),
+                      'sequence':self.sequence,'broadcaster':str(broadcaster or 'system')[:100],
+                      'event':str(event or 'UNKNOWN')[:80],'details':self._safe(details or {})}
+                self.events.append(item)
+                if len(self.events)>self.maximum:
+                    self.events=self.events[-max(100,int(self.maximum*0.9)):];self._rewrite()
+                elif not self.path.exists():self._rewrite()
+                else:self._append(item)
+                return item
+        except (OSError,TypeError,ValueError):
+            return None
+    def read(self,limit=200,broadcaster=None):
+        try:limit=max(1,min(int(limit),2000))
+        except (TypeError,ValueError):limit=200
+        with self.lock:
+            rows=self.events
+            if broadcaster:rows=[x for x in rows if x.get('broadcaster')==broadcaster]
+            return [dict(x) for x in rows[-limit:]]
+    def note_rtmp_open(self,broadcaster,details=None):
+        user=str(broadcaster or 'system')
+        with self.lock:
+            count=self.rtmp_opens.get(user,0)+1;self.rtmp_opens[user]=count
+        if count>1:
+            merged={'scope':'youtube_rtmp','session_number':count};merged.update(details or {})
+            return self.log(user,'RECONNECT',merged)
+        return None
+
+
+monitor=StreamMonitor()
+def log_event(broadcaster,event,details=None):return monitor.log(broadcaster,event,details)
+def read_events(limit=200,broadcaster=None):return monitor.read(limit,broadcaster)
+def note_rtmp_open(broadcaster,details=None):return monitor.note_rtmp_open(broadcaster,details)

test_flv_bridge.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 """Deterministic decoder-record/reconnect tests; no network or transcoding."""
+import json
 import os
 import pathlib
 import shutil
@@ -10,6 +11,7 @@ import tempfile
 ROOT=pathlib.Path(sys.argv[1] if len(sys.argv)>1 else pathlib.Path(__file__).parent).resolve()
 sys.path.insert(0,str(ROOT))
 from flv_bridge import FLVBridge  # noqa: E402
+from stream_monitor import StreamMonitor, video_dimensions  # noqa: E402
 
 passed=failed=0
 def check(name,ok,detail=''):
@@ -69,7 +71,9 @@ VIDEO_H=b'\x00\x00\x00\x04\x26\x01\xaf\x09'
 
 # A changed AVCDecoderConfigurationRecord must precede its dependent frame and must not raise,
 # even when an old caller still enables detect_format_change.
-b=FLVBridge(detect_format_change=True,allow_raise=lambda:True)
+events=[]
+b=FLVBridge(detect_format_change=True,allow_raise=lambda:True,user='alice',
+            event_logger=lambda user,event,details:events.append((user,event,details)))
 avc_input=flv(tag(9,0,seq('avc1',AVC_A)),tag(9,0,coded('avc1',VIDEO_A,b'\xff\xff\xfe')),
               tag(9,40,seq('avc1',AVC_B)),tag(9,40,coded('avc1',VIDEO_A)))
 avc_out=chunked(b,avc_input);avc_tags=parse(avc_out)
@@ -81,6 +85,9 @@ check('AVC coded bytes and signed CTS pass unchanged',
       [x[2] for x in avc_tags if x[0]==9 and x[2][1:2]==b'\x01']==
       [coded('avc1',VIDEO_A,b'\xff\xff\xfe'),coded('avc1',VIDEO_A)])
 check('continuous-source DTS is preserved exactly',[x[1] for x in avc_tags]==[0,0,40,40])
+check('AVC config and resolution changes are recorded with broadcaster identity',
+      [x[1] for x in events]==['STREAM_CONFIG','DECODER_CONFIG_CHANGE','RESOLUTION_CHANGE'] and
+      events[-1][0]=='alice' and events[-1][2]['new_width']==640)
 
 # A complete avcC may itself contain accumulated arrays. Preserve the record as a unit rather
 # than trying to append old records with potentially conflicting parameter-set IDs.
@@ -110,11 +117,15 @@ eh=parse(FLVBridge().feed(flv(tag(9,0,eh_seq),tag(9,1,eh_frame))))
 check('already-enhanced hvc1 bodies pass unchanged',[x[2] for x in eh]==[eh_seq,eh_frame])
 
 # Reattach suppresses only an identical startup record and preserves timestamp continuity.
-br=FLVBridge();first=br.feed(flv(tag(9,100,seq('avc1',AVC_A)),tag(9,140,coded('avc1',VIDEO_A))))
+reconnect_events=[]
+br=FLVBridge(user='alice',event_logger=lambda user,event,details:reconnect_events.append((event,details)))
+first=br.feed(flv(tag(9,100,seq('avc1',AVC_A)),tag(9,140,coded('avc1',VIDEO_A))))
 last=br.last_out_ts;br.reconnect()
 second=br.feed(flv(tag(18,0,b'metadata'),tag(9,0,seq('avc1',AVC_A)),tag(9,40,coded('avc1',VIDEO_A))))
 joined=parse(first+second)
 check('reattach drops second FLV header and startup metadata',len(second)>0 and not second.startswith(b'FLV') and all(x[0]!=18 for x in joined))
+check('source reattach is recorded without claiming RTMP closure',
+      reconnect_events[-1][0]=='RECONNECT' and reconnect_events[-1][1]['outbound_rtmp_reused'] is True)
 check('reattach drops only identical startup decoder record',sum(1 for x in joined if x[0]==9 and x[2][1:2]==b'\x00')==1)
 check('reattach shifts DTS once while preserving valid equal timestamps',
       br.last_out_ts>last and all(a[1]<=z[1] for a,z in zip(joined,joined[1:])))
@@ -129,12 +140,15 @@ check('changed reattach remains on one FLV byte stream',not part2.startswith(b'F
 # The byte bridge can represent a FourCC transition without raising or closing. FFmpeg 7 creates
 # a new late AVStream for such a transition, so production source selection still pins one codec;
 # same-codec decoder-record replacement above is the fully verified path.
-bs=FLVBridge(detect_format_change=True)
+switch_events=[]
+bs=FLVBridge(detect_format_change=True,user='alice',
+             event_logger=lambda user,event,details:switch_events.append(event))
 switch=bs.feed(flv(tag(9,0,seq('avc1',AVC_A)),tag(9,1,coded('avc1',VIDEO_A)),
                    tag(9,40,seq('hvc1',HEVC_A,legacy=False)),tag(9,41,coded('hvc1',VIDEO_H,legacy=False))))
 switch_tags=parse(switch)
 check('AVC to HEVC transition does not raise or truncate bridge output',len(switch_tags)==4)
 check('codec transition is separately observable',bs.format_changes==1 and bs.codec_changes==1 and bs.video_codec=='hvc1')
+check('codec transition writes CODEC_CHANGE without stopping the bridge','CODEC_CHANGE' in switch_events)
 
 try:
     FLVBridge().feed(flv(tag(9,0,b'\x90hvc1')))
@@ -143,6 +157,32 @@ except ValueError:
     truncated=True
 check('truncated Enhanced SequenceStart fails in a controlled way',truncated)
 
+# Metric transitions are inferred from tag bytes/timestamps, never by decoding the picture.
+metric_events=[];records=[tag(9,0,seq('avc1',AVC_A))]
+for ts in range(0,10000,40):
+    frame=1 if ts in (0,2000,4000,5000,6000,7000,8000,9000) else 2
+    payload=(b'x'*100) if ts<5000 else (b'y'*1000)
+    records.append(tag(9,ts,coded('avc1',payload,frame=frame)))
+FLVBridge(user='alice',event_logger=lambda user,event,details:metric_events.append(event)).feed(flv(*records))
+check('byte-level monitor detects a sudden bitrate change','BITRATE_CHANGE' in metric_events)
+check('byte-level monitor detects a keyframe-interval change','KEYFRAME_CHANGE' in metric_events)
+check('AVC and HEVC SPS dimensions are parsed without decoding',
+      video_dimensions('avc1',AVC_A)==(320,240) and video_dimensions('avc1',AVC_B)==(640,360) and
+      video_dimensions('hvc1',HEVC_A)==(320,240) and video_dimensions('hvc1',HEVC_B)==(640,360))
+
+with tempfile.TemporaryDirectory() as mon_dir:
+    mon_path=pathlib.Path(mon_dir)/'logs'/'stream_events.json';mon=StreamMonitor(mon_path)
+    mon.log('alice','CONNECTION_ERROR',{'scope':'youtube_rtmp','url':'secret-url',
+            'stream_key':'secret-key','reason':'failed https://secret-host/live/SECRET'})
+    mon.note_rtmp_open('alice');mon.note_rtmp_open('alice',{'mode':'copy'})
+    saved=json.loads(mon_path.read_text())
+    check('stream event file is valid JSON with the required schema',
+          len(saved)==2 and all(set(('timestamp','broadcaster','event','details'))<=set(x) for x in saved))
+    check('stream monitor never persists URLs, keys or tokens',
+          all(x not in mon_path.read_text() for x in ('secret-url','secret-key','secret-host','SECRET')))
+    check('stream event file is owner-only',mon_path.stat().st_mode&0o777==0o600)
+    check('second outbound publisher open is recorded as RECONNECT',saved[-1]['event']=='RECONNECT')
+
 # FFmpeg 7 is the real outbound middle layer. If available, prove its FLV demux/mux pair writes
 # replacement AVC and HEVC configuration records instead of fixing extradata at startup.
 ff=os.environ.get('FFMPEG_PATH') or shutil.which('ffmpeg')

test_regressions.py

@@ -129,11 +129,14 @@ with TestClient(appmod.app) as client:
     unauth=client.get('/api/state')
     wrong=client.get('/api/state',auth=('admin','wrong'))
     right=client.get('/api/state',auth=('admin',os.environ['PASSWORD']))
+    stream_events=client.get('/api/stream-events',auth=('admin',os.environ['PASSWORD']))
 check('embedded root shows login instead of raw 401', root.status_code==200 and 'login-panel' in root.text)
 check('public health contains no broadcaster/account details', set(health)<= {'status','application','monitor','version'}, str(health))
 check('operational API rejects anonymous access', unauth.status_code==401)
 check('operational API rejects a wrong password', wrong.status_code==401)
 check('operational API accepts the configured password', right.status_code==200)
+check('authenticated stream-event API exposes the bounded JSON log',
+      stream_events.status_code==200 and stream_events.json().get('file')=='logs/stream_events.json')
 
 appmod.set_token('alice','sessionid='+'x'*32)
 mode=stat.S_IMODE(appmod.TOKENS.stat().st_mode)

test_relay.py

@@ -392,7 +392,9 @@ check('Dockerfile defaults to byte-copy AVC/Enhanced-HEVC without forced x264',
       'YOUTUBE_LIGHT_MAX=0' in df and
       'YOUTUBE_AUTO_TRANSCODE=0' in df and
       'YOUTUBE_FORCE_TRANSCODE_USERS=' not in df and
-      'YOUTUBE_ENHANCED_RTMP=auto' in df)
+      'YOUTUBE_ENHANCED_RTMP=auto' in df and
+      'stream_monitor.py' in df and 'YOUTUBE_HEALTH_INTERVAL=20' in df and
+      'STREAM_EVENT_MAX=10000' in df)
 mgr_src=(ROOT/'youtube_manager.py').read_text()
 check('run_manager calls restart_dead_senders', 'restart_dead_senders' in mgr_src.split('def run_manager')[1])
 pump=(ROOT/'relay.py').read_text()

youtube_manager.py

@@ -8,6 +8,7 @@ from zoneinfo import ZoneInfo
 from urllib.parse import urlparse
 from curl_cffi import requests
 from relay import RelayStartError, Sender, SourceStalled, source_http_headers
+from stream_monitor import log_event
 import inspect
 
 from youtube_api import GoogleAPI, Journal, APIError
@@ -1758,7 +1759,7 @@ class ManagedYouTube:
         (videoResolutionUnsupported / videoCodec) and why playback stalls (gopSizeLong /
         openGop). Surfacing it replaces guesswork with ground truth. Polled on a slower
         cadence than the life-cycle check so it stays inside the quota budget."""
-        interval=max(45,int(os.getenv('YOUTUBE_HEALTH_INTERVAL','60')))
+        interval=max(15,int(os.getenv('YOUTUBE_HEALTH_INTERVAL','20')))
         if time.time()-float(ev.get('last_health') or 0)<interval:return
         stream_id=(self.journal.read()['streams'].get(user) or {}).get('id')
         if not stream_id:return
@@ -1774,6 +1775,8 @@ class ManagedYouTube:
         text=' | '.join(issues)
         prev=ev.get('last_issues')
         status=str(health.get('status') or '')
+        log_event(user,'YOUTUBE_HEALTH',{'status':status or 'unknown','issues':issues,
+                                         'issue_count':len(issues)})
         self.event_update(key,youtube_health=status,youtube_issues=text,last_issues=text)
         if text and text!=prev:
             self.log(user,'يوتيوب يشخّص البث: '+text)