refactor(streaming): let the backend pick the resume picker's states (#5210)

* refactor(streaming): let the backend pick the resume picker's states

The player decided which of a ROM's states the resume picker offers from
a core rule the config shipped, and so repeated the rules the claim
checks a pick against.

The player now asks `GET /api/streaming/states/{rom_id}` which states
load natively and which the broker can import, so the rules live only in
the backend. A native state whose filename carries no slot number is
offered only when the broker can import it, since it can't resume as is.
The player asks again only when the ROM's states, the emulator or the
import kinds change, keeps its last answer if a refresh fails, and holds
Play until the first answer for the ROM arrives, so a quick click can't
skip the state about to be preselected.

`StreamingContainerSchema.state_core` and `StateCoreSchema` are removed,
and the config no longer asks each broker for its core.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

* fix(streaming): address Greptile review on the state picks

Refetch the picks when a state is renamed, since the filename decides whether it resumes as it is, and fold the new docstrings to one sentence.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

---------

Co-authored-by: Zach Clendenen <[email protected]>
Co-authored-by: Claude Fable 5.1 <[email protected]>

backend/endpoints/responses/streaming.py

@@ -20,17 +20,6 @@ class SlotCapabilitiesSchema(BaseModel):
     has_manual_disc_swap: bool
 
 
-class StateCoreSchema(BaseModel):
-    """Which RetroArch states the container's core can load."""
-
-    expected: str
-    # Whether a state with no recorded core, the default's, matches too.
-    default_matches: bool
-    # The emulators, lowercased, another client's state can be filed under and
-    # still load in this core: its RetroArch folder and that folder's aliases.
-    folders: list[str]
-
-
 class StreamingContainerSchema(BaseModel):
     """One platform the fleet can stream, as the play screen needs it."""
 
@@ -45,9 +34,15 @@ class StreamingContainerSchema(BaseModel):
     # Which foreign-emulator picks the broker declares it can import, empty when it
     # declares none or cannot be asked.
     import_kinds: list[Literal["save", "state"]]
-    # The picker's core filter, the same one hydration applies. None filters
-    # nothing: not RetroArch, or the broker can't say which core runs.
-    state_core: StateCoreSchema | None = None
+
+
+class StreamStatePicksSchema(BaseModel):
+    """Which of a ROM's states the stream's resume picker offers, by id."""
+
+    # Resume as they are; the newest is preselected.
+    native: list[int]
+    # Native ones plus the foreign ones the broker imports.
+    pickable: list[int]
 
 
 class StreamingConfigSchema(BaseModel):

backend/endpoints/streaming.py

@@ -50,9 +50,9 @@ from endpoints.responses.streaming import (
     SaveStateResponse,
     SessionStatusSchema,
     SlotCapabilitiesSchema,
-    StateCoreSchema,
     StreamingConfigSchema,
     StreamingContainerSchema,
+    StreamStatePicksSchema,
     SwapDiscResponse,
     VolumeResponse,
 )
@@ -61,6 +61,7 @@ from handler.auth.dependencies import assert_rom_visible
 from handler.database import (
     db_container_adoption_handler,
     db_rom_handler,
+    db_state_handler,
     db_user_handler,
 )
 from handler.redis_handler import async_cache
@@ -300,7 +301,7 @@ def _joinable_container_label(
     return entries[0].container_label or entries[0].label
 
 
-# Two checks per platform; past this many, a config waits a second timeout.
+# One check per platform; past this many, a config waits a second timeout.
 _CONFIG_CHECK_THREADS = 64
 # Shared by every request, so overlapping page loads hold no more threads than one.
 _config_pool = ThreadPoolExecutor(
@@ -350,28 +351,9 @@ async def get_config(request: Request) -> StreamingConfigSchema:
         if access.platform_is_visible(request, c.platform)
     ]
     # All at once, so a silent broker costs one timeout.
-    specs, cores = await asyncio.gather(
-        asyncio.gather(
-            *(
-                _shared_check(
-                    _check_key("spec", c),
-                    webstation.import_spec,
-                    c,
-                    c.emulator,
-                    c.platform,
-                )
-                for c in visible
-            )
-        ),
-        asyncio.gather(
-            *(
-                _shared_check(_check_key("core", c), states.state_core_for, c)
-                for c in visible
-            )
-        ),
-    )
+    specs = await asyncio.gather(*(_import_spec(c) for c in visible))
     safe_containers: list[StreamingContainerSchema] = []
-    for c, spec, core in zip(visible, specs, cores, strict=True):
+    for c, spec in zip(visible, specs, strict=True):
         safe_containers.append(
             StreamingContainerSchema(
                 platform=c.platform,
@@ -394,16 +376,6 @@ async def get_config(request: Request) -> StreamingConfigSchema:
                 supports_live_states=c.supports_live_states,
                 # So the picker only offers a foreign pick the claim will take.
                 import_kinds=spec.pickable_kinds() if spec is not None else [],
-                # So the picker hides states another RetroArch core wrote.
-                state_core=(
-                    StateCoreSchema(
-                        expected=core.expected,
-                        default_matches=core.default_matches,
-                        folders=core.folders,
-                    )
-                    if core is not None
-                    else None
-                ),
             )
         )
 
@@ -414,6 +386,56 @@ async def get_config(request: Request) -> StreamingConfigSchema:
     )
 
 
+def _import_spec(c: ResolvedContainer) -> Awaitable[webstation.ImportSpec | None]:
+    return _shared_check(
+        _check_key("spec", c), webstation.import_spec, c, c.emulator, c.platform
+    )
+
+
+def _streamable_rom(
+    request: Request, rom_id: int
+) -> tuple[Rom, list[ResolvedContainer]]:
+    """The ROM and the containers serving its platform, else a 404 (the same one
+    for a hidden ROM, so its id can't launch it)."""
+    rom = db_rom_handler.get_rom(rom_id)
+    if rom is None:
+        raise HTTPException(status_code=404, detail="ROM not found")
+    assert_rom_visible(request, rom, not_found_detail="ROM not found")
+    platform = rom.platform_slug
+    candidates = containers_for_platform(platform)
+    if not candidates:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No streaming container configured for platform '{platform}'",
+        )
+    return rom, candidates
+
+
+@protected_route(router.get, "/states/{rom_id}", [Scope.ROMS_READ, Scope.ASSETS_READ])
+async def get_state_picks(request: Request, rom_id: int) -> StreamStatePicksSchema:
+    """The ROM's states the resume picker offers, checked as the claim checks a
+    pick; 404 when the ROM is hidden or missing, or no container serves it."""
+    rom, candidates = _streamable_rom(request, rom_id)
+    # The claim validates the pick against the pool's first container too.
+    c = candidates[0]
+    spec, core, shared = await asyncio.gather(
+        _import_spec(c),
+        _shared_check(_check_key("core", c), states.state_core_for, c),
+        asyncio.to_thread(
+            db_state_handler.get_rom_shared_states,
+            rom_id=rom.id,
+            user_id=request.user.id,
+        ),
+    )
+    picks = states.state_picks(
+        shared,
+        c.emulator,
+        core,
+        imports_states=spec is not None and spec.resume_slot() is not None,
+    )
+    return StreamStatePicksSchema(**picks._asdict())
+
+
 async def _win_container(
     request: Request,
     candidates: list[ResolvedContainer],
@@ -841,22 +863,8 @@ async def claim_session(
     """
     access.assert_can_start_streaming(request)
 
-    rom = db_rom_handler.get_rom(req.rom_id)
-    if rom is None:
-        raise HTTPException(status_code=404, detail="ROM not found")
-
-    # A hidden ROM/platform must not be launchable via its id: enforce the same
-    # visibility policy as the ROM detail/content endpoints before any broker
-    # launch. Raises a 404 that masks the hidden ROM's existence.
-    assert_rom_visible(request, rom, not_found_detail="ROM not found")
-
+    rom, candidates = _streamable_rom(request, req.rom_id)
     platform = rom.platform_slug
-    candidates = containers_for_platform(platform)
-    if not candidates:
-        raise HTTPException(
-            status_code=404,
-            detail=f"No streaming container configured for platform '{platform}'",
-        )
 
     # Pool members are interchangeable (`ResolvedContainer.interchangeable_with`),
     # so the pre-claim validation below holds for whichever one the walk wins.

backend/handler/streaming/states.py

@@ -23,6 +23,7 @@ import os
 import re
 import zipfile
 import zlib
+from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any, NamedTuple
@@ -423,6 +424,33 @@ def is_native_state(state: State, emulator: str, core: StateCore | None) -> bool
     return _loads_in(state, emulator, core)
 
 
+class StatePicks(NamedTuple):
+    """The state ids the resume picker offers, in the order given."""
+
+    native: list[int]
+    pickable: list[int]
+
+
+def state_picks(
+    candidates: Iterable[State],
+    emulator: str,
+    core: StateCore | None,
+    imports_states: bool,
+) -> StatePicks:
+    """Split the states the resume claim takes: native ones with a slot resume as
+    they are; with `imports_states` the rest import, except another core's capture."""
+    native: list[int] = []
+    pickable: list[int] = []
+    for state in candidates:
+        loads = is_native_state(state, emulator, core)
+        if loads and slot_from_state_filename(emulator, state.file_name) is not None:
+            native.append(state.id)
+            pickable.append(state.id)
+        elif imports_states and (loads or not _filed_by_stream(state, emulator)):
+            pickable.append(state.id)
+    return StatePicks(native, pickable)
+
+
 def user_states_for_emulator(
     user_id: int, rom_id: int, emulator: str, core: StateCore | None = None
 ) -> list[State]:

backend/tests/endpoints/test_streaming.py

@@ -412,26 +412,170 @@ def test_get_config_ships_capabilities_for_a_retroarch_platform(client, access_t
     assert caps["autosave_slot"] == 10
 
 
-def test_config_ships_the_state_core_rule(client, access_token):
-    with _streaming(_webstation(platforms={"snes": "retroarch:bsnes"})):
+def _picks(client, token, rom: Rom, platform: str, spec=None, core=None):
+    container = _webstation(platforms={rom.platform_slug: platform})
+    with _streaming(container):
         with (
             patch(
-                "handler.streaming.webstation.require_import_spec", return_value=None
+                "handler.streaming.webstation.require_import_spec", return_value=spec
             ),
-            patch("handler.streaming.webstation.default_core", return_value="snes9x"),
+            patch("handler.streaming.states.state_core_for", return_value=core),
         ):
-            body = client.get(
-                "/api/streaming/config", headers=_auth(access_token)
-            ).json()
-    (snes,) = [c for c in body["containers"] if c["platform"] == "snes"]
-    assert snes["state_core"] == {
-        "expected": "bsnes",
-        "default_matches": False,
-        "folders": ["bsnes"],
+            return client.get(f"/api/streaming/states/{rom.id}", headers=_auth(token))
+
+
+_STATE_IMPORT = webstation.ImportSpec(frozenset({"state"}), "push", 9)
+
+
+@pytest.mark.parametrize(
+    ("spec", "pickable"),
+    [(None, {"own", "synced"}), (_STATE_IMPORT, {"own", "synced", "foreign"})],
+)
+def test_state_picks_follow_the_resume_rules(
+    client, access_token, rom: Rom, admin_user: User, spec, pickable
+):
+    """The booted core's states resume natively, wherever they are filed; a
+    foreign one only through an import, and another core's capture never."""
+    made = {
+        "own": _core_state(rom, admin_user, "Game.state", "bsnes"),
+        "other_core": _core_state(rom, admin_user, "Game.state1", "snes9x"),
+        "synced": _add_state(rom, admin_user, "Game.state2", "bsnes"),
+        "foreign": _add_state(rom, admin_user, "Game.ss0", "snes9x"),
     }
+    names = {state.id: name for name, state in made.items()}
+
+    r = _picks(
+        client,
+        access_token,
+        rom,
+        "retroarch:bsnes",
+        spec=spec,
+        core=states.StateCore("bsnes", False),
+    )
+
+    assert r.status_code == 200
+    assert {names[i] for i in r.json()["native"]} == {"own", "synced"}
+    assert {names[i] for i in r.json()["pickable"]} == pickable
 
 
-def test_config_asks_a_broker_its_spec_and_core_at_once(client, access_token):
+_BSNES = states.StateCore("bsnes", False)
+_PICK_STATES = [
+    # (id, emulator, core)
+    (5, "retroarch", "bsnes"),
+    (6, "retroarch", "snes9x"),
+    (7, "retroarch", None),
+    (8, "duckstation", None),
+    (9, "bsnes", None),
+    (10, "bsnes", "bsnes"),
+    (11, "snes9x", None),
+    (12, "Snes9x 2010", None),
+    (13, "Snes9x 2010", "snes9x2010"),
+    (14, "snes9x2010", "snes9x2010"),
+    (15, "snes9x2010", None),
+    (16, "Snes9x 2010", "snes9x"),
+    (17, "snes9x", "snes9x"),
+]
+
+
+@pytest.mark.parametrize(
+    ("core", "imports_states", "native", "pickable"),
+    [
+        # Another core's capture never; Cloud Sync's under the core's folder too.
+        (_BSNES, False, [5, 9, 10], [5, 9, 10]),
+        # A state with no core recorded is the default's.
+        (
+            states.StateCore("snes9x", True),
+            False,
+            [6, 7, 11, 16, 17],
+            [6, 7, 11, 16, 17],
+        ),
+        (
+            states.StateCore("snes9x2010", False, folder="Snes9x 2010"),
+            False,
+            [12, 13, 14],
+            [12, 13, 14],
+        ),
+        # Unknown core: only the stream's own, wherever the core filed them.
+        (None, False, [5, 6, 7, 10, 13, 14, 16, 17], [5, 6, 7, 10, 13, 14, 16, 17]),
+        # An import takes any other client's state, never another core's capture.
+        (_BSNES, True, [5, 9, 10], [5, 8, 9, 10, 11, 12, 15]),
+    ],
+)
+def test_state_picks_split_native_from_importable(
+    core, imports_states, native, pickable
+):
+    made = [
+        State(id=i, emulator=e, core=c, file_name="Game.state")
+        for i, e, c in _PICK_STATES
+    ]
+
+    picks = states.state_picks(made, "retroarch", core, imports_states)
+
+    assert picks == states.StatePicks(native, pickable)
+
+
+@pytest.mark.parametrize(("imports_states", "pickable"), [(False, []), (True, [1, 2])])
+def test_state_picks_offer_a_state_with_no_slot_only_through_an_import(
+    imports_states, pickable
+):
+    """The claim cannot resume `.state100` as it is, so only an import takes it."""
+    made = [
+        State(id=1, emulator="bsnes", file_name="Game.state100"),
+        State(id=2, emulator="retroarch", core="bsnes", file_name="Game.state100"),
+    ]
+
+    picks = states.state_picks(made, "retroarch", _BSNES, imports_states)
+
+    assert picks == states.StatePicks([], pickable)
+
+
+def test_state_picks_without_a_known_core_take_the_streams_own(
+    client, access_token, rom: Rom, admin_user: User
+):
+    own = _core_state(rom, admin_user, "Game.state", "bsnes")
+    _add_state(rom, admin_user, "Game.state2", "bsnes")
+
+    r = _picks(client, access_token, rom, "retroarch")
+
+    assert r.json() == {"native": [own.id], "pickable": [own.id]}
+
+
+def test_state_picks_list_states_newest_first(
+    client, access_token, rom: Rom, admin_user: User
+):
+    """In the order the rom's state list carries them, so the first native one
+    is the newest the picker preselects."""
+    older = _add_state_at(rom, admin_user, "Game.01.p2s", 1)
+    newer = _add_state_at(rom, admin_user, "Game.02.p2s", 2)
+    shared = [
+        s.id
+        for s in db_state_handler.get_rom_shared_states(
+            rom_id=rom.id, user_id=admin_user.id
+        )
+    ]
+
+    r = _picks(client, access_token, rom, "pcsx2")
+
+    assert r.json()["native"] == shared
+    assert set(shared) == {older.id, newer.id}
+
+
+def test_state_picks_hide_a_hidden_rom(
+    client, viewer_access_token, viewer_user: User, rom: Rom
+):
+    _hide(PermEntity.ROMS, rom.id, viewer_user.id)
+    assert _picks(client, viewer_access_token, rom, "pcsx2").status_code == 404
+
+
+def test_state_picks_404_without_a_container(client, access_token, rom: Rom):
+    with _streaming(_webstation(platforms={"snes": "retroarch"})):
+        r = client.get(f"/api/streaming/states/{rom.id}", headers=_auth(access_token))
+    assert r.status_code == 404
+
+
+def test_state_picks_ask_the_broker_its_spec_and_core_at_once(
+    client, access_token, rom: Rom
+):
     """So an unreachable broker costs the play screen one timeout, not two."""
     both = threading.Barrier(2, timeout=5)
 
@@ -439,7 +583,7 @@ def test_config_asks_a_broker_its_spec_and_core_at_once(client, access_token):
         both.wait()
         return value
 
-    with _streaming(_webstation(platforms={"snes": "retroarch"})):
+    with _streaming(_webstation(platforms={rom.platform_slug: "retroarch"})):
         with (
             patch("handler.streaming.webstation.require_import_spec", side_effect=meet),
             patch(
@@ -447,30 +591,40 @@ def test_config_asks_a_broker_its_spec_and_core_at_once(client, access_token):
                 side_effect=lambda c: meet(value="snes9x"),
             ),
         ):
-            r = client.get("/api/streaming/config", headers=_auth(access_token))
+            r = client.get(
+                f"/api/streaming/states/{rom.id}", headers=_auth(access_token)
+            )
     assert r.status_code == 200
 
 
+def test_config_leaves_the_state_rules_to_the_picks(client, access_token):
+    """The config asks the broker only its import spec; the core is the picks' call."""
+    with _streaming(_webstation(platforms={"snes": "retroarch"})):
+        with (
+            patch(
+                "handler.streaming.webstation.require_import_spec", return_value=None
+            ),
+            patch("handler.streaming.webstation.default_core") as default_core,
+        ):
+            body = client.get(
+                "/api/streaming/config", headers=_auth(access_token)
+            ).json()
+    default_core.assert_not_called()
+    assert "state_core" not in body["containers"][0]
+
+
 def test_config_asks_every_platform_at_once(client, access_token):
     """More checks than the default thread pool holds (at most 32), so a silent
     broker still costs the play screen one timeout rather than one per wave."""
-    platforms = (
-        "nes snes n64 gb gbc gba genesis atari2600 pokemon-mini psp psx sms "
-        "gamegear tg16 ngp wonderswan lynx"
-    ).split()
-    everyone = threading.Barrier(2 * len(platforms), timeout=5)
+    platforms = [f"platform-{n}" for n in range(34)]
+    everyone = threading.Barrier(len(platforms), timeout=5)
 
-    def meet(*_args, value=None):
+    def meet(*_args):
         everyone.wait()
-        return value
 
     with _streaming(_webstation(platforms=dict.fromkeys(platforms, "retroarch"))):
-        with (
-            patch("handler.streaming.webstation.require_import_spec", side_effect=meet),
-            patch(
-                "handler.streaming.webstation.default_core",
-                side_effect=lambda c: meet(value="snes9x"),
-            ),
+        with patch(
+            "handler.streaming.webstation.require_import_spec", side_effect=meet
         ):
             r = client.get("/api/streaming/config", headers=_auth(access_token))
     assert r.status_code == 200
@@ -486,14 +640,13 @@ def test_overlapping_configs_share_each_broker_check(client, access_token):
 
     def join(*args):
         joins.append(args[0])
-        if len(joins) == 4:
+        if len(joins) == 2:
             both_joined.set()
         return shared_check(*args)
 
-    def stall(*_args, value=None):
-        asked.append(value or "spec")
+    def stall(*_args):
+        asked.append("spec")
         assert both_joined.wait(timeout=5)
-        return value
 
     with _streaming(_webstation(platforms={"snes": "retroarch"})):
         with (
@@ -501,10 +654,6 @@ def test_overlapping_configs_share_each_broker_check(client, access_token):
             patch(
                 "handler.streaming.webstation.require_import_spec", side_effect=stall
             ),
-            patch(
-                "handler.streaming.webstation.default_core",
-                side_effect=lambda c: stall(value="snes9x"),
-            ),
             ThreadPoolExecutor(2) as pages,
         ):
             loads = [
@@ -515,18 +664,7 @@ def test_overlapping_configs_share_each_broker_check(client, access_token):
             ]
             responses = [load.result(timeout=10) for load in loads]
     assert [r.status_code for r in responses] == [200, 200]
-    assert sorted(asked) == ["snes9x", "spec"]
-
-
-def test_config_has_no_state_core_outside_retroarch(client, access_token):
-    with _streaming(_webstation()):
-        with patch(
-            "handler.streaming.webstation.require_import_spec", return_value=None
-        ):
-            body = client.get(
-                "/api/streaming/config", headers=_auth(access_token)
-            ).json()
-    assert all(c["state_core"] is None for c in body["containers"])
+    assert asked == ["spec"]
 
 
 def test_get_config_labels_each_platform_by_its_emulator(client, access_token):

frontend/src/__generated__/index.ts

@@ -279,12 +279,12 @@ export type { SmartCollectionSchema } from './models/SmartCollectionSchema';
 export type { SoundtrackTrackMetaSchema } from './models/SoundtrackTrackMetaSchema';
 export type { SSAgeRating } from './models/SSAgeRating';
 export type { SSPhysicalDisc } from './models/SSPhysicalDisc';
-export type { StateCoreSchema } from './models/StateCoreSchema';
 export type { StateSchema } from './models/StateSchema';
 export type { StatsReturn } from './models/StatsReturn';
 export type { SteamPlatforms } from './models/SteamPlatforms';
 export type { StreamingConfigSchema } from './models/StreamingConfigSchema';
 export type { StreamingContainerSchema } from './models/StreamingContainerSchema';
+export type { StreamStatePicksSchema } from './models/StreamStatePicksSchema';
 export type { SwapDiscRequest } from './models/SwapDiscRequest';
 export type { SwapDiscResponse } from './models/SwapDiscResponse';
 export type { SyncCompletePayload } from './models/SyncCompletePayload';

frontend/src/__generated__/models/StateCoreSchema.ts

@@ -1,13 +0,0 @@
-/* generated using openapi-typescript-codegen -- do not edit */
-/* istanbul ignore file */
-/* tslint:disable */
-/* eslint-disable */
-/**
- * Which RetroArch states the container's core can load.
- */
-export type StateCoreSchema = {
-    expected: string;
-    default_matches: boolean;
-    folders: Array<string>;
-};
-

frontend/src/__generated__/models/StreamStatePicksSchema.ts

@@ -0,0 +1,12 @@
+/* generated using openapi-typescript-codegen -- do not edit */
+/* istanbul ignore file */
+/* tslint:disable */
+/* eslint-disable */
+/**
+ * Which of a ROM's states the stream's resume picker offers, by id.
+ */
+export type StreamStatePicksSchema = {
+    native: Array<number>;
+    pickable: Array<number>;
+};
+

frontend/src/__generated__/models/StreamingContainerSchema.ts

@@ -3,7 +3,6 @@
 /* tslint:disable */
 /* eslint-disable */
 import type { SlotCapabilitiesSchema } from './SlotCapabilitiesSchema';
-import type { StateCoreSchema } from './StateCoreSchema';
 /**
  * One platform the fleet can stream, as the play screen needs it.
  */
@@ -17,6 +16,5 @@ export type StreamingContainerSchema = {
     supports_save_picker: boolean;
     supports_live_states: boolean;
     import_kinds: Array<'save' | 'state'>;
-    state_core?: (StateCoreSchema | null);
 };
 

frontend/src/services/api/streaming.ts

@@ -24,6 +24,7 @@ import type {
   SlotCapabilitiesSchema,
   StreamingConfigSchema,
   StreamingContainerSchema,
+  StreamStatePicksSchema,
   SwapDiscResponse,
   VolumeResponse,
 } from "@/__generated__";
@@ -39,6 +40,7 @@ import api, { keepaliveHeaders } from "@/services/api";
 export type PlatformCapabilities = SlotCapabilitiesSchema;
 export type StreamingContainer = StreamingContainerSchema;
 export type StreamingConfig = StreamingConfigSchema;
+export type StreamStatePicks = StreamStatePicksSchema;
 export type LaunchingSession = LaunchingSessionSchema;
 
 // Socket payloads reach no route, so the backend publishes them into the
@@ -97,6 +99,12 @@ async function fetchConfig() {
   });
 }
 
+/** Which of the ROM's states the resume picker offers, decided by the same
+ *  rules the claim checks a pick against. */
+async function fetchStatePicks(romId: number) {
+  return api.get<StreamStatePicks>(`/streaming/states/${romId}`);
+}
+
 /** Answers 202 as soon as the container is reserved. The room URL arrives on
  *  the socket as `streaming:launch-ready`, so this needs no ceiling of its own
  *  beyond the client default. */
@@ -315,6 +323,7 @@ function releaseSessionKeepalive(
 
 export default {
   fetchConfig,
+  fetchStatePicks,
   claimSession,
   listJoinableSessions,
   joinSession,

frontend/src/v2/views/Player/Stream.test.ts

@@ -2,7 +2,11 @@ import { RBtn, RSlider } from "@v2/lib";
 import { flushPromises, mount, type VueWrapper } from "@vue/test-utils";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { defineComponent, nextTick, type Slots, type VNodeChild } from "vue";
-import type { SaveSchema, StateCoreSchema, StateSchema } from "@/__generated__";
+import type {
+  SaveSchema,
+  StateSchema,
+  StreamStatePicksSchema,
+} from "@/__generated__";
 import type { DetailedRom } from "@/stores/roms";
 import { saveFixture, stateFixture } from "@/utils/assets.fixtures";
 import { detailedRomFixture } from "@/utils/rom.fixtures";
@@ -14,6 +18,7 @@ import Stream from "./Stream.vue";
 const mocks = vi.hoisted(() => ({
   claimSession: vi.fn(),
   fetchConfig: vi.fn(() => Promise.resolve()),
+  fetchStatePicks: vi.fn(),
   getRom: vi.fn(),
   fetchSessionStatus: vi.fn(),
   heartbeatSession: vi.fn(),
@@ -57,6 +62,7 @@ vi.mock("@/services/api/rom", () => ({
 
 vi.mock("@/services/api/streaming", () => ({
   default: {
+    fetchStatePicks: mocks.fetchStatePicks,
     loadState: mocks.loadState,
     saveState: mocks.saveState,
     setVolume: mocks.setVolume,
@@ -248,7 +254,8 @@ async function launch(opts: {
   states?: StateSchema[];
   liveStates?: boolean;
   imports?: ("save" | "state")[];
-  stateCore?: StateCoreSchema | null;
+  // The backend's answer; every state is native unless a test says otherwise.
+  picks?: StreamStatePicksSchema | Error | Promise<StreamStatePicksSchema>;
 }): Promise<VueWrapper> {
   mocks.container = {
     name: "WEBSTATION-DEV",
@@ -257,13 +264,19 @@ async function launch(opts: {
     supports_save_picker: opts.picker,
     supports_live_states: opts.liveStates ?? true,
     import_kinds: opts.imports ?? [],
-    state_core: opts.stateCore ?? null,
     supports_memory_cards: false,
     supports_multiplayer: false,
   };
   mocks.getRom.mockResolvedValue({
     data: romWith(opts.saves ?? ARCHIVES, opts.states),
   });
+  const ids = (opts.states ?? []).map((state) => state.id);
+  const picks = opts.picks ?? { native: ids, pickable: ids };
+  mocks.fetchStatePicks.mockImplementation(() =>
+    picks instanceof Error
+      ? Promise.reject(picks)
+      : Promise.resolve(picks).then((data) => ({ data })),
+  );
   const wrapper = mount(Stream, {
     shallow: true,
     global: {
@@ -597,6 +610,8 @@ describe("Stream save picker", () => {
   });
 });
 
+type UserState = DetailedRom["all_user_states"][number];
+
 type StreamVm = {
   onPlay: () => Promise<void>;
   performStop: () => Promise<void>;
@@ -611,6 +626,10 @@ type StreamVm = {
   containerHost: string;
   errorMessage: string;
   errorHint: string;
+  playReady: boolean;
+  rom: DetailedRom;
+  pickableStates: UserState[];
+  selectedState: UserState | null;
 };
 
 function vmOf(wrapper: VueWrapper): StreamVm {
@@ -1368,103 +1387,127 @@ describe("Stream state picker", () => {
   });
 
   const STATES = [
-    stateFixture({ id: 5, emulator: "retroarch" }),
+    stateFixture({ id: 7, emulator: "retroarch" }),
     stateFixture({ id: 6, emulator: "duckstation" }),
+    stateFixture({ id: 5, emulator: "retroarch" }),
   ];
 
   function pickableStateIds(wrapper: VueWrapper): number[] {
-    return (
-      wrapper.vm as unknown as { pickableStates: StateSchema[] }
-    ).pickableStates.map((s) => s.id);
+    return vmOf(wrapper).pickableStates.map((s) => s.id);
   }
 
-  it("offers only this emulator's states where the broker declares no state import", async () => {
-    const wrapper = await launch({ picker: false, states: STATES });
+  function selectedStateId(wrapper: VueWrapper): number | undefined {
+    return vmOf(wrapper).selectedState?.id;
+  }
 
-    expect(pickableStateIds(wrapper)).toEqual([5]);
+  it("asks the backend which of the rom's states to offer", async () => {
+    await launch({ picker: false, states: STATES });
+
+    expect(mocks.fetchStatePicks).toHaveBeenCalledWith(3);
   });
 
-  it("offers another emulator's states where the broker imports them", async () => {
+  it("offers the picked states in the rom's newest-first order", async () => {
     const wrapper = await launch({
       picker: false,
       states: STATES,
-      imports: ["state"],
+      picks: { native: [5], pickable: [5, 6] },
     });
 
-    expect(pickableStateIds(wrapper)).toEqual([5, 6]);
+    expect(pickableStateIds(wrapper)).toEqual([6, 5]);
   });
 
-  const CORE_STATES = [
-    stateFixture({ id: 5, emulator: "retroarch", core: "bsnes" }),
-    stateFixture({ id: 6, emulator: "retroarch", core: "snes9x" }),
-    stateFixture({ id: 7, emulator: "retroarch", core: null }),
-    stateFixture({ id: 8, emulator: "duckstation", core: null }),
-    // Other clients' states, filed under a core's RetroArch folder.
-    stateFixture({ id: 9, emulator: "bsnes", core: null }),
-    stateFixture({ id: 10, emulator: "Snes9x 2010", core: null }),
-    // A stream's capture, filed under the core's folder.
-    stateFixture({ id: 11, emulator: "bsnes", core: "bsnes" }),
-    stateFixture({ id: 12, emulator: "snes9x", core: "snes9x" }),
-  ];
-  const BSNES: StateCoreSchema = {
-    expected: "bsnes",
-    default_matches: false,
-    folders: ["bsnes"],
-  };
-
-  it("hides states another RetroArch core wrote", async () => {
+  it("preselects the newest native state, not a newer import", async () => {
     const wrapper = await launch({
       picker: false,
-      states: CORE_STATES,
-      stateCore: BSNES,
+      states: STATES,
+      picks: { native: [5], pickable: [6, 5] },
     });
 
-    expect(pickableStateIds(wrapper)).toEqual([5, 9, 11]);
+    expect(selectedStateId(wrapper)).toBe(5);
   });
 
-  it("treats a state with no core as the default's", async () => {
+  it("offers nothing when the backend can't say", async () => {
     const wrapper = await launch({
       picker: false,
-      states: CORE_STATES,
-      stateCore: {
-        expected: "snes9x",
-        default_matches: true,
-        folders: ["snes9x"],
-      },
+      states: STATES,
+      picks: new Error("offline"),
     });
 
-    expect(pickableStateIds(wrapper)).toEqual([6, 7, 12]);
+    expect(pickableStateIds(wrapper)).toEqual([]);
+    expect(selectedStateId(wrapper)).toBeUndefined();
   });
 
-  it("loads another client's state filed under an alias of the core's folder", async () => {
+  it("keeps the pick when a refresh of the picks fails", async () => {
     const wrapper = await launch({
       picker: false,
-      states: CORE_STATES,
-      stateCore: {
-        expected: "snes9x2010",
-        default_matches: false,
-        folders: ["snes9x 2010", "snes9x2010"],
-      },
+      states: STATES,
+      picks: { native: [7, 5], pickable: [7, 6, 5] },
     });
+    const vm = vmOf(wrapper);
+    vm.selectedState = vm.rom.all_user_states[2]!;
+    mocks.fetchStatePicks.mockRejectedValue(new Error("offline"));
 
-    expect(pickableStateIds(wrapper)).toEqual([10]);
+    vm.rom = {
+      ...vm.rom,
+      all_user_states: vm.rom.all_user_states.filter((s) => s.id !== 6),
+    };
+    await flushPromises();
+
+    expect(mocks.fetchStatePicks).toHaveBeenCalledTimes(2);
+    expect(pickableStateIds(wrapper)).toEqual([7, 5]);
+    expect(selectedStateId(wrapper)).toBe(5);
   });
 
-  it("offers only the stream's own states when the core is unknown", async () => {
-    const wrapper = await launch({ picker: false, states: CORE_STATES });
-
-    expect(pickableStateIds(wrapper)).toEqual([5, 6, 7, 11, 12]);
-  });
-
-  it("keeps foreign states for the import path but still hides other cores", async () => {
+  it("holds Play until the picks answer", async () => {
+    let answer!: (picks: StreamStatePicksSchema) => void;
     const wrapper = await launch({
       picker: false,
-      states: CORE_STATES,
-      imports: ["state"],
-      stateCore: BSNES,
+      states: STATES,
+      picks: new Promise((resolve) => (answer = resolve)),
+    });
+    expect(vmOf(wrapper).playReady).toBe(false);
+
+    answer({ native: [5], pickable: [5] });
+    await flushPromises();
+
+    expect(vmOf(wrapper).playReady).toBe(true);
+    expect(selectedStateId(wrapper)).toBe(5);
+  });
+
+  it("frees Play when the picks fail", async () => {
+    const wrapper = await launch({
+      picker: false,
+      states: STATES,
+      picks: new Error("offline"),
     });
 
-    expect(pickableStateIds(wrapper)).toEqual([5, 8, 9, 10, 11]);
+    expect(vmOf(wrapper).playReady).toBe(true);
+    expect(selectedStateId(wrapper)).toBeUndefined();
+  });
+
+  it("asks again only when the answer could change", async () => {
+    const wrapper = await launch({ picker: false, states: STATES });
+    const vm = vmOf(wrapper);
+
+    vm.rom = { ...vm.rom, all_user_states: [...vm.rom.all_user_states] };
+    await flushPromises();
+
+    expect(mocks.fetchStatePicks).toHaveBeenCalledOnce();
+  });
+
+  it("asks again when a state is renamed", async () => {
+    const wrapper = await launch({ picker: false, states: STATES });
+    const vm = vmOf(wrapper);
+
+    vm.rom = {
+      ...vm.rom,
+      all_user_states: vm.rom.all_user_states.map((s) =>
+        s.id === 5 ? { ...s, file_name: "game.state100" } : s,
+      ),
+    };
+    await flushPromises();
+
+    expect(mocks.fetchStatePicks).toHaveBeenCalledTimes(2);
   });
 });
 

frontend/src/v2/views/Player/Stream.vue

@@ -51,6 +51,7 @@ import streamingApi, {
   type LaunchReady,
   type MemoryCardImport,
   type MemoryCardImportDetail,
+  type StreamStatePicks,
   type StreamingContainer,
 } from "@/services/api/streaming";
 import storeAuth from "@/stores/auth";
@@ -380,44 +381,51 @@ const selectedSave = computed<SaveSchema | null>(
     newestSave.value,
 );
 
-// A stream files a state under its emulator or, for a RetroArch core, under
-// the core's folder with the core recorded, which only a stream records.
-function filedByStream(state: UserStateSchema): boolean {
-  const emulator = emulatorKey(container.value?.emulator);
-  if (!emulator) return false;
-  return (
-    emulatorKey(state.emulator) === emulator ||
-    (emulator === "retroarch" && state.core != null)
-  );
+// The backend picks the offered states by the claim's rules: native ones resume
+// as they are, the rest only through an import the broker declares.
+const statePicks = ref<StreamStatePicks | null>(null);
+// The rom whose picks have answered, which Play waits for.
+const statePicksRomId = ref<number | null>(null);
+let statePicksRequest = 0;
+// Keyed on what changes the answer, a rename included, so a config or rom
+// refresh that changes nothing asks nothing.
+const statePicksKey = computed(() => {
+  const current = container.value;
+  if (!rom.value || !current) return null;
+  const ids = (rom.value.all_user_states ?? [])
+    .map((s) => `${s.id}:${s.file_name}`)
+    .join(",");
+  const kinds = current.import_kinds.join(",");
+  return `${rom.value.id}|${current.emulator}|${kinds}|${ids}`;
+});
+watch(
+  statePicksKey,
+  async (key) => {
+    const romId = rom.value?.id;
+    if (key === null || romId === undefined || isJoining) return;
+    const request = ++statePicksRequest;
+    let picks: StreamStatePicks | null = null;
+    try {
+      ({ data: picks } = await streamingApi.fetchStatePicks(romId));
+    } catch {
+      // A failed refresh keeps the last answer, so the user's pick survives;
+      // with none yet, nothing is offered and Play boots fresh.
+    }
+    if (!alive.value || request !== statePicksRequest) return;
+    if (picks) statePicks.value = picks;
+    statePicksRomId.value = romId;
+  },
+  { immediate: true },
+);
+
+// In the order all_user_states carries them, newest first.
+function statesById(ids: number[] | undefined): UserStateSchema[] {
+  const wanted = new Set(ids);
+  return (rom.value?.all_user_states ?? []).filter((s) => wanted.has(s.id));
 }
 
-// A capture loads in its core (none recorded: the default's); another client's
-// state loads when it is filed in the core's folder.
-function loadsNatively(state: UserStateSchema): boolean {
-  const core = container.value?.state_core;
-  if (!core) return filedByStream(state);
-  if (filedByStream(state))
-    return (
-      state.core === core.expected || (!state.core && core.default_matches)
-    );
-  return core.folders.includes(emulatorKey(state.emulator));
-}
-
-const nativeStreamStates = computed<UserStateSchema[]>(() => {
-  if (!rom.value || !container.value) return [];
-  return (rom.value.all_user_states ?? []).filter(loadsNatively);
-});
-
-// Other clients' states too where the broker imports one, through the import
-// path; another core's capture never loads, which the import refuses as well.
-const pickableStates = computed<UserStateSchema[]>(() => {
-  if (!rom.value) return [];
-  if (!container.value?.import_kinds.includes("state"))
-    return nativeStreamStates.value;
-  return (rom.value.all_user_states ?? []).filter(
-    (s) => loadsNatively(s) || !filedByStream(s),
-  );
-});
+const nativeStreamStates = computed(() => statesById(statePicks.value?.native));
+const pickableStates = computed(() => statesById(statePicks.value?.pickable));
 
 // Every capture is kept, so a heavy save-stater ends up with a history the
 // horizontal strip buries. Grid and list trade thumbnail size for how many
@@ -539,9 +547,16 @@ const emulatorLabel = computed(
   () => container.value?.label ?? platformLabel.value,
 );
 
-// Held with the resume picker, so Play never sends a pick the player can't see.
+// Held with the resume picker and its picks, so Play never sends a pick the
+// player can't see or skips the state about to be preselected.
 const playReady = computed(
-  () => !!rom.value && configFresh.value && playerState.value !== "loading",
+  () =>
+    !!rom.value &&
+    configFresh.value &&
+    playerState.value !== "loading" &&
+    (isJoining ||
+      statePicksKey.value === null ||
+      statePicksRomId.value === rom.value.id),
 );
 usePlayFocus(".r-v2-stream__play", playReady, gameRunning);