@@ -9,6 +9,30 @@ completes consent.
 See the [shared connector procedures](../../docs/connectors/README.md) for OAuth credential
 rotation, CLI operations, and the general process for changing scopes.
 
+## Discover scheduled YouTube broadcasts
+
+`videos` lists recent uploads, which does not establish whether a scheduled
+stream exists. Use the existing authorized YouTube connection and the shared
+discovery API/console resource instead:
+
+```bash
+dapier connections show youtube datatalks
+dapier connections discover youtube:datatalks broadcasts
+dapier connections discover youtube:datatalks broadcasts --param status=all
+dapier connections discover youtube:datatalks broadcasts --param page_token=NEXT_TOKEN
+```
+
+The default is `status=upcoming`; `active`, `completed` and `all` are also
+supported. Verify `channel_id`, title and `scheduled_start` against the owning
+event before changing anything. Results include the existing description and
+privacy/lifecycle state. If the final row has `next_page_token`, continue with
+that token and the same status before concluding a target is absent.
+
+This resource uses Google's [liveBroadcasts.list](https://developers.google.com/youtube/data/live/reference/rest/v3/liveBroadcasts/list)
+and works with `youtube.readonly`. It creates or changes no broadcast. A read
+grant does not authorize metadata or thumbnail writes; do not broaden an OAuth
+grant to bypass a missing write permission.
+
 ## Current configuration
 
 - Google Cloud project: **`dtcdev-click`**.
 
@@ -299,6 +299,23 @@ def _run_videos(connection, params, *, transport=None):
 ))
 
 
+def _run_broadcasts(connection, params, *, transport=None):
+    return provider.discover(connection, "broadcasts", params, transport=transport)
+
+
+register_discovery(Discovery(
+    name="broadcasts",
+    connector="youtube",
+    label="Live broadcasts",
+    description="Owned live broadcasts (default: upcoming), including scheduled streams",
+    params=({"key": "status", "label": "Status (upcoming, active, completed, all)",
+             "type": "text", "required": False},
+            {"key": "page_token", "label": "Next page token", "type": "text",
+             "required": False}),
+    run=_run_broadcasts,
+))
+
+
 def _run_test(connection, *, transport=None):
     return provider.test_connection(connection, transport=transport)
 
 
@@ -434,6 +434,45 @@ def _youtube_videos(connection, token, params, limit, *, transport=None):
                                    limit, transport=transport)
 
 
+def _youtube_broadcasts(connection, token, params, limit, *, transport=None):
+    """Read scheduled streams separately from uploaded videos.
+
+    The status filter scopes liveBroadcasts.list to the authenticated channel;
+    it must not be combined with the mutually exclusive mine filter.
+    """
+    status = params.get("status", "upcoming")
+    if status not in {"upcoming", "active", "completed", "all"}:
+        raise DiscoveryError("broadcasts status must be upcoming, active, completed or all")
+    query = {"part": "snippet,status", "broadcastStatus": status,
+             "broadcastType": "all", "maxResults": min(limit, 50)}
+    if params.get("page_token"):
+        query["pageToken"] = params["page_token"]
+    url = f"{YOUTUBE_API_URL}/liveBroadcasts?" + urllib.parse.urlencode(query)
+    data = _request("GET", url, token, None, transport=transport)
+    items = []
+    for entry in data.get("items") or []:
+        if not isinstance(entry, dict) or not entry.get("id"):
+            continue
+        snippet = entry.get("snippet") or {}
+        state = entry.get("status") or {}
+        items.append({
+            "id": entry["id"], "name": snippet.get("title") or entry["id"],
+            "channel_id": snippet.get("channelId"),
+            "scheduled_start": snippet.get("scheduledStartTime"),
+            "scheduled_end": snippet.get("scheduledEndTime"),
+            "lifecycle_status": state.get("lifeCycleStatus"),
+            "privacy_status": state.get("privacyStatus"),
+            "description": snippet.get("description", ""),
+            "url": f"https://www.youtube.com/watch?v={entry['id']}",
+        })
+    if data.get("nextPageToken"):
+        if not items:
+            raise DiscoveryError("YouTube returned an empty broadcast page with a continuation token",
+                                 status=502)
+        items[-1]["next_page_token"] = data["nextPageToken"]
+    return items
+
+
 # --- Dropbox ---
 
 
@@ -758,6 +797,10 @@ def _zoom_recordings(connection, token, params, limit, *, transport=None):
                  (Param("playlist_id", True, "Playlist ID from the playlists list"),)),
         Resource("videos", "Channel videos", "The channel's recent uploads, newest first",
                  _youtube_videos),
+        Resource("broadcasts", "Live broadcasts", "Owned live broadcasts (default: upcoming)",
+                 _youtube_broadcasts,
+                 (Param("status", False, "upcoming (default), active, completed or all"),
+                  Param("page_token", False, "next_page_token from the last result row"))),
     ],
     "dropbox": [
         Resource("folder", "Folder entries",
 
@@ -45,7 +45,7 @@ def live_token(monkeypatch):
 def test_youtube_catalog_lists_channel_playlists_playlist_items_and_videos():
     resources = discovery.resources_for("youtube")
     assert [resource.name for resource in resources] == \
-        ["channel", "playlists", "playlist_items", "videos"]
+        ["channel", "playlists", "playlist_items", "videos", "broadcasts"]
     playlist_items = resources[2]
     assert [(param.name, param.required) for param in playlist_items.params] == \
         [("playlist_id", True)]
@@ -166,7 +166,56 @@ def fake_discover(connection, resource, params, **kwargs):
     assert sorted(seen) == sorted([("yt-main", "channel", {"a": "b"}),
                                    ("yt-main", "playlists", {"a": "b"}),
                                    ("yt-main", "playlist_items", {"a": "b"}),
-                                   ("yt-main", "videos", {"a": "b"})])
+                                   ("yt-main", "videos", {"a": "b"}),
+                                   ("yt-main", "broadcasts", {"a": "b"})])
+
+
+def test_broadcasts_include_upcoming_details_and_continuation():
+    from urllib.parse import parse_qs, urlparse
+
+    transport = FakeTransport(("liveBroadcasts?", 200, {
+        "items": [{"id": "scheduled-1", "snippet": {
+            "title": "A future Q&A", "channelId": "UCfixture",
+            "description": "Existing speaker details",
+            "scheduledStartTime": "2027-03-02T16:00:00Z",
+        }, "status": {"lifeCycleStatus": "ready", "privacyStatus": "unlisted"}}],
+        "nextPageToken": "page-two",
+    }))
+    items = discovery.discover(CONNECTION, "broadcasts", {}, transport=transport)
+    assert items[0]["id"] == "scheduled-1"
+    assert items[0]["channel_id"] == "UCfixture"
+    assert items[0]["scheduled_start"] == "2027-03-02T16:00:00Z"
+    assert items[0]["description"] == "Existing speaker details"
+    assert items[0]["privacy_status"] == "unlisted"
+    assert items[0]["next_page_token"] == "page-two"
+    query = parse_qs(urlparse(transport.calls[0]["url"]).query)
+    assert query["broadcastStatus"] == ["upcoming"]
+    assert query["broadcastType"] == ["all"]
+    assert "mine" not in query
+
+
+def test_broadcasts_follow_explicit_page_and_preserve_empty_result():
+    transport = FakeTransport(("pageToken=page-two", 200, {"items": []}))
+    assert discovery.discover(CONNECTION, "broadcasts", {
+        "status": "all", "page_token": "page-two", "limit": 100,
+    }, transport=transport) == []
+    assert "maxResults=50" in transport.calls[0]["url"]
+
+
+def test_broadcasts_reject_invalid_status():
+    transport = FakeTransport()
+    with pytest.raises(discovery.DiscoveryError, match="status must"):
+        discovery.discover(CONNECTION, "broadcasts", {"status": "draft"},
+                           transport=transport)
+    assert not transport.calls
+
+
+def test_empty_incomplete_broadcast_page_is_not_reported_as_absent():
+    transport = FakeTransport(("liveBroadcasts?", 200, {
+        "items": [], "nextPageToken": "more",
+    }))
+    with pytest.raises(discovery.DiscoveryError, match="empty broadcast page"):
+        discovery.discover(CONNECTION, "broadcasts", {}, transport=transport)
 
 
 def test_youtube_connection_test_reports_the_channel_identity(monkeypatch):