feat(tiktok): add tiktok.trending verb for the Explore feed

The Explore feed (/api/explore/item_list) is a global trending-video feed
served to anonymous sessions, and it returns the same itemStruct shape as the
other listings — so the verb reuses parse_video, the listing flow, the
TikTokVideoItem output, and the per-video billing meter wholesale. Adds a
browser-capture marker + fetch_trending, a synthetic-target orchestrator entry,
and the tiktok.trending capability, surfaced on the chat subagent.

surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/tiktok/tools/index.py

@@ -1,4 +1,4 @@
-"""``tiktok`` sub-agent tools: the TikTok scrape, comments, and user-search verbs."""
+"""``tiktok`` sub-agent tools: scrape, comments, user-search, and trending verbs."""
 
 from __future__ import annotations
 
@@ -10,13 +10,14 @@ from app.agents.chat.multi_agent_chat.shared.permissions import Ruleset
 from app.capabilities.core.access.agent import build_capability_tools
 from app.capabilities.tiktok.comments.definition import TIKTOK_COMMENTS
 from app.capabilities.tiktok.scrape.definition import TIKTOK_SCRAPE
+from app.capabilities.tiktok.trending.definition import TIKTOK_TRENDING
 from app.capabilities.tiktok.user_search.definition import TIKTOK_USER_SEARCH
 
 NAME = "tiktok"
 
 RULESET = Ruleset(origin=NAME, rules=[])
 
-_CI_VERBS = [TIKTOK_SCRAPE, TIKTOK_COMMENTS, TIKTOK_USER_SEARCH]
+_CI_VERBS = [TIKTOK_SCRAPE, TIKTOK_COMMENTS, TIKTOK_USER_SEARCH, TIKTOK_TRENDING]
 
 
 def load_tools(

surfsense_backend/app/capabilities/tiktok/__init__.py

@@ -4,4 +4,5 @@ from __future__ import annotations
 
 from app.capabilities.tiktok.comments import definition as _comments  # noqa: F401
 from app.capabilities.tiktok.scrape import definition as _scrape  # noqa: F401
+from app.capabilities.tiktok.trending import definition as _trending  # noqa: F401
 from app.capabilities.tiktok.user_search import definition as _user_search  # noqa: F401

surfsense_backend/app/capabilities/tiktok/trending/__init__.py

@@ -0,0 +1,3 @@
+"""``tiktok.trending``: pull the current trending videos from the Explore feed."""
+
+from __future__ import annotations

surfsense_backend/app/capabilities/tiktok/trending/definition.py

@@ -0,0 +1,23 @@
+"""``tiktok.trending`` capability registration (billed per video on the shared
+``TIKTOK_MICROS_PER_VIDEO`` meter)."""
+
+from __future__ import annotations
+
+from app.capabilities.core import BillingUnit, Capability, register_capability
+from app.capabilities.tiktok.trending.executor import build_trending_executor
+from app.capabilities.tiktok.trending.schemas import TrendingInput, TrendingOutput
+
+TIKTOK_TRENDING = Capability(
+    name="tiktok.trending",
+    description=(
+        "Get the current trending TikTok videos from the Explore feed. No input "
+        "needed beyond how many to return."
+    ),
+    input_schema=TrendingInput,
+    output_schema=TrendingOutput,
+    executor=build_trending_executor(),
+    billing_unit=BillingUnit.TIKTOK_VIDEO,
+    docs_url="/docs/connectors/native/tiktok",
+)
+
+register_capability(TIKTOK_TRENDING)

surfsense_backend/app/capabilities/tiktok/trending/executor.py

@@ -0,0 +1,32 @@
+"""``tiktok.trending`` executor: Explore feed -> scraper -> TikTok video items."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+from app.capabilities.core import Executor
+from app.capabilities.core.progress import emit_progress
+from app.capabilities.tiktok.trending.schemas import TrendingInput, TrendingOutput
+from app.proprietary.platforms.tiktok import scrape_tiktok_trending
+
+TrendingFn = Callable[..., Awaitable[list[dict]]]
+
+
+def build_trending_executor(trending_fn: TrendingFn | None = None) -> Executor:
+    """Bind the executor to a trending fn (defaults to the proprietary actor)."""
+    trending_fn = trending_fn or scrape_tiktok_trending
+
+    async def execute(payload: TrendingInput) -> TrendingOutput:
+        emit_progress(
+            "starting",
+            "Fetching TikTok trending videos",
+            total=payload.max_items,
+            unit="item",
+        )
+        items = await trending_fn(count=payload.max_items)
+        emit_progress(
+            "done", f"Fetched {len(items)} video(s)", current=len(items), unit="item"
+        )
+        return TrendingOutput(items=items)
+
+    return execute

surfsense_backend/app/capabilities/tiktok/trending/schemas.py

@@ -0,0 +1,41 @@
+"""``tiktok.trending`` I/O contracts.
+
+The Explore feed (``/api/explore/item_list``) is a single global feed of trending
+videos, served to anonymous sessions. No source input is needed — just how many
+to return. Each result reuses :class:`TikTokVideoItem`, so trending videos bill on
+the same per-video meter as ``tiktok.scrape``.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from app.capabilities.tiktok.scrape.schemas import MAX_TIKTOK_ITEMS
+from app.proprietary.platforms.tiktok import TikTokVideoItem
+
+
+class TrendingInput(BaseModel):
+    max_items: int = Field(
+        default=20,
+        ge=1,
+        le=MAX_TIKTOK_ITEMS,
+        description="Max trending videos to return from the Explore feed.",
+    )
+
+    @property
+    def estimated_units(self) -> int:
+        """Worst-case billable videos for the pre-flight gate (le=100 ceiling)."""
+        return self.max_items
+
+
+class TrendingOutput(BaseModel):
+    items: list[TikTokVideoItem] = Field(
+        default_factory=list,
+        description="One item per trending video returned, in feed order.",
+    )
+
+    @property
+    def billable_units(self) -> int:
+        """One returned video = one billable unit; an ErrorItem (``errorCode`` set,
+        for an empty/withheld feed) is surfaced but never charged."""
+        return sum(1 for item in self.items if not getattr(item, "errorCode", None))

surfsense_backend/app/proprietary/platforms/tiktok/__init__.py

@@ -10,6 +10,7 @@ from .orchestrator import (
     iter_tiktok,
     scrape_tiktok,
     scrape_tiktok_comments,
+    scrape_tiktok_trending,
     search_tiktok_users,
 )
 from .schemas import (
@@ -29,5 +30,6 @@ __all__ = [
     "iter_tiktok",
     "scrape_tiktok",
     "scrape_tiktok_comments",
+    "scrape_tiktok_trending",
     "search_tiktok_users",
 ]

surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py

@@ -12,18 +12,19 @@ from collections.abc import AsyncIterator
 from typing import Any
 from urllib.parse import quote
 
+from .extraction.timestamps import now_iso
 from .flows import FetchCommentsFn, FetchFn, FetchListingFn, FetchUsersFn
 from .flows.comments import iter_comments
 from .flows.listing import iter_listing
 from .flows.profile import iter_profile
 from .flows.user_search import iter_user_search
 from .flows.video import iter_video
-from .extraction.timestamps import now_iso
 from .schemas import ErrorItem, TikTokScrapeInput
 from .session import (
     fetch_comments,
     fetch_html,
     fetch_item_list,
+    fetch_trending,
     fetch_user_search,
 )
 from .targets import resolve_target
@@ -32,6 +33,7 @@ from .targets.types import TikTokTarget
 _PROFILE_URL = "https://www.tiktok.com/@{name}"
 _HASHTAG_URL = "https://www.tiktok.com/tag/{tag}"
 _SEARCH_URL = "https://www.tiktok.com/search?q={query}"
+_EXPLORE_URL = "https://www.tiktok.com/explore"
 
 
 def _resolve_targets(input_model: TikTokScrapeInput) -> list[TikTokTarget]:
@@ -132,6 +134,26 @@ async def search_tiktok_users(
     return results
 
 
+async def scrape_tiktok_trending(
+    *,
+    count: int,
+    fetch_trending_fn: FetchListingFn = fetch_trending,
+) -> list[dict[str, Any]]:
+    """Collect up to ``count`` trending videos from the Explore feed.
+
+    A single global feed, so it reuses the listing flow (parse/dedupe/cap/empty-
+    ErrorItem) over a synthetic target — no user input to resolve.
+    """
+    from app.capabilities.core.progress import emit_progress
+
+    target = TikTokTarget(kind="trending", value="explore", url=_EXPLORE_URL)
+    results: list[dict[str, Any]] = []
+    async for item in iter_listing(target, cap=count, fetch_listing=fetch_trending_fn):
+        results.append(item)
+        emit_progress("scraping", current=len(results), total=count, unit="item")
+    return results
+
+
 async def scrape_tiktok_comments(
     video_urls: list[str],
     *,

surfsense_backend/app/proprietary/platforms/tiktok/session/__init__.py

@@ -4,7 +4,12 @@ from __future__ import annotations
 
 from .client import fetch_html
 from .errors import TikTokAccessBlockedError
-from .listing import fetch_comments, fetch_item_list, fetch_user_search
+from .listing import (
+    fetch_comments,
+    fetch_item_list,
+    fetch_trending,
+    fetch_user_search,
+)
 from .proxy import bind_proxy_holder, open_proxy_holder, proxy_session
 
 __all__ = [
@@ -13,6 +18,7 @@ __all__ = [
     "fetch_comments",
     "fetch_html",
     "fetch_item_list",
+    "fetch_trending",
     "fetch_user_search",
     "open_proxy_holder",
     "proxy_session",

surfsense_backend/app/proprietary/platforms/tiktok/session/listing.py

@@ -51,6 +51,8 @@ _ITEM_LIST_MARKERS = (
 )
 # The user-search XHR carries account records (user_list), not itemStructs.
 _USER_SEARCH_MARKERS = ("/api/search/user",)
+# The Explore feed's trending videos arrive as ordinary itemStructs.
+_EXPLORE_MARKERS = ("/api/explore/item_list",)
 # The comment feed fires only after the comments panel is opened.
 _COMMENT_MARKERS = ("/api/comment/list",)
 _COMMENT_ICON_SELECTORS = (
@@ -255,3 +257,15 @@ async def fetch_comments(page_url: str, target_count: int) -> list[dict[str, Any
         comments_from_response,
         _scroll_comments,
     )
+
+
+async def fetch_trending(page_url: str, target_count: int) -> list[dict[str, Any]]:
+    """Return up to ``target_count`` trending itemStructs from the Explore feed."""
+    return await asyncio.to_thread(
+        _fetch_sync,
+        page_url,
+        target_count,
+        _EXPLORE_MARKERS,
+        items_from_response,
+        _scroll_page,
+    )

surfsense_backend/app/proprietary/platforms/tiktok/targets/types.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 from dataclasses import dataclass
 from typing import Literal
 
-TargetKind = Literal["video", "profile", "hashtag", "search"]
+TargetKind = Literal["video", "profile", "hashtag", "search", "trending"]
 SearchSection = Literal["video", "user"]
 
 

surfsense_backend/tests/unit/capabilities/tiktok/test_registry.py

@@ -11,6 +11,7 @@ from app.capabilities.core import BillingUnit
 from app.capabilities.core.store import get_capability
 from app.capabilities.tiktok.comments.schemas import CommentsInput, CommentsOutput
 from app.capabilities.tiktok.scrape.schemas import ScrapeInput, ScrapeOutput
+from app.capabilities.tiktok.trending.schemas import TrendingInput, TrendingOutput
 from app.capabilities.tiktok.user_search.schemas import (
     UserSearchInput,
     UserSearchOutput,
@@ -44,3 +45,13 @@ def test_tiktok_comments_is_registered_and_billed_per_comment():
     assert cap.input_schema is CommentsInput
     assert cap.output_schema is CommentsOutput
     assert cap.billing_unit is BillingUnit.TIKTOK_COMMENT
+
+
+def test_tiktok_trending_is_registered_and_billed_per_video():
+    cap = get_capability("tiktok.trending")
+
+    assert cap.name == "tiktok.trending"
+    assert cap.input_schema is TrendingInput
+    assert cap.output_schema is TrendingOutput
+    # Trending returns videos, so it shares the per-video meter.
+    assert cap.billing_unit is BillingUnit.TIKTOK_VIDEO

surfsense_backend/tests/unit/capabilities/tiktok/trending/test_executor.py

@@ -0,0 +1,38 @@
+"""``tiktok.trending`` executor: verb input → scraper count → typed video items.
+
+Boundary mocked: the proprietary trending actor (injected fake). NOT mocked: the
+verb's own count forwarding and the dict→TikTokVideoItem wrapping.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.capabilities.tiktok.trending.executor import build_trending_executor
+from app.capabilities.tiktok.trending.schemas import TrendingInput, TrendingOutput
+
+pytestmark = pytest.mark.unit
+
+
+class _FakeTrending:
+    """Records the count it was called with; returns canned items."""
+
+    def __init__(self, items: list[dict]):
+        self._items = items
+        self.calls: list[int] = []
+
+    async def __call__(self, *, count: int) -> list[dict]:
+        self.calls.append(count)
+        return self._items
+
+
+async def test_forwards_count_and_wraps_items():
+    trending = _FakeTrending([{"id": "1", "text": "viral"}])
+    execute = build_trending_executor(trending_fn=trending)
+
+    out = await execute(TrendingInput(max_items=30))
+
+    assert isinstance(out, TrendingOutput)
+    assert len(out.items) == 1
+    assert out.items[0].id == "1"
+    assert trending.calls == [30]

surfsense_backend/tests/unit/capabilities/tiktok/trending/test_schemas.py

@@ -0,0 +1,33 @@
+"""``tiktok.trending`` input guards and billing: bounded count, ErrorItems free."""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from app.capabilities.tiktok.scrape.schemas import MAX_TIKTOK_ITEMS
+from app.capabilities.tiktok.trending.schemas import TrendingInput, TrendingOutput
+
+pytestmark = pytest.mark.unit
+
+
+def test_defaults_and_bounds():
+    payload = TrendingInput()
+    assert payload.max_items == 20
+    assert payload.estimated_units == 20
+    with pytest.raises(ValidationError):
+        TrendingInput(max_items=0)
+    with pytest.raises(ValidationError):
+        TrendingInput(max_items=MAX_TIKTOK_ITEMS + 1)
+
+
+def test_error_items_are_not_billed():
+    # Real videos count; an ErrorItem (empty/withheld feed) is surfaced free.
+    out = TrendingOutput(
+        items=[
+            {"id": "1", "webVideoUrl": "https://tiktok.com/@a/video/1"},
+            {"errorCode": "no_items", "input": "explore", "error": "empty"},
+        ]
+    )
+    assert len(out.items) == 2
+    assert out.billable_units == 1

surfsense_backend/tests/unit/platforms/tiktok/test_trending.py

@@ -0,0 +1,35 @@
+"""Trending orchestration over a fake fetch (no network).
+
+Drives ``scrape_tiktok_trending``: the Explore feed -> captured itemStructs ->
+video items, reusing the listing flow (parse/dedupe/cap/empty-ErrorItem).
+"""
+
+from __future__ import annotations
+
+from app.proprietary.platforms.tiktok import scrape_tiktok_trending
+
+
+async def test_trending_parses_dedupes_and_caps():
+    async def fake_fetch(url: str, _cap: int) -> list[dict]:
+        assert url == "https://www.tiktok.com/explore"
+        return [
+            {"id": "1", "author": {"uniqueId": "a"}},
+            {"id": "1", "author": {"uniqueId": "a"}},
+            {"id": "2", "author": {"uniqueId": "b"}},
+        ]
+
+    items = await scrape_tiktok_trending(count=2, fetch_trending_fn=fake_fetch)
+
+    assert [i["id"] for i in items] == ["1", "2"]
+    assert items[0]["webVideoUrl"] == "https://www.tiktok.com/@a/video/1"
+    assert items[0]["scrapedAt"] is not None
+
+
+async def test_trending_empty_feed_emits_error_item():
+    async def fake_fetch(_url: str, _cap: int) -> list[dict]:
+        return []
+
+    items = await scrape_tiktok_trending(count=5, fetch_trending_fn=fake_fetch)
+
+    assert len(items) == 1
+    assert items[0]["errorCode"] == "no_items"