feat(streaming): hydrate and resume only states the booted core can load

A RetroArch container's core filter comes from the core activate reports
booting, or before launch from the configured core and the broker's
default (asked over its cores route and cached for a minute). Hydration,
an explicit resume and the config payload all apply it, so a state
written by one core never lands where another core's quick-load reads it.
A resume the booted core cannot load launches fresh, and the resume
endpoint refuses such a state up front with a 400.

backend/endpoints/responses/streaming.py

@@ -20,6 +20,14 @@ 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
+
+
 class StreamingContainerSchema(BaseModel):
     """One platform the fleet can stream, as the play screen needs it."""
 
@@ -34,6 +42,9 @@ 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 StreamingConfigSchema(BaseModel):

backend/endpoints/streaming.py

@@ -42,6 +42,7 @@ from endpoints.responses.streaming import (
     SaveStateResponse,
     SessionStatusSchema,
     SlotCapabilitiesSchema,
+    StateCoreSchema,
     StreamingConfigSchema,
     StreamingContainerSchema,
     SwapDiscResponse,
@@ -304,14 +305,17 @@ async def get_config(request: Request) -> StreamingConfigSchema:
         if access.platform_is_visible(request, c.platform)
     ]
     # Concurrently, so one unreachable broker costs one timeout, not one each.
-    specs = await asyncio.gather(
-        *(
-            asyncio.to_thread(webstation.import_spec, c, c.emulator, c.platform)
-            for c in visible
-        )
+    specs, cores = await asyncio.gather(
+        asyncio.gather(
+            *(
+                asyncio.to_thread(webstation.import_spec, c, c.emulator, c.platform)
+                for c in visible
+            )
+        ),
+        asyncio.gather(*(asyncio.to_thread(states.state_core_for, c) for c in visible)),
     )
     safe_containers: list[StreamingContainerSchema] = []
-    for c, spec in zip(visible, specs, strict=True):
+    for c, spec, core in zip(visible, specs, cores, strict=True):
         safe_containers.append(
             StreamingContainerSchema(
                 platform=c.platform,
@@ -334,6 +338,14 @@ 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
+                    )
+                    if core is not None
+                    else None
+                ),
             )
         )
 

backend/handler/streaming/launch.py

@@ -147,6 +147,11 @@ async def run_launch(
         if container.core and isinstance(launch_result, dict)
         else None
     )
+    state_core = await asyncio.to_thread(
+        states.state_core_for,
+        container,
+        launch_result if isinstance(launch_result, dict) else None,
+    )
     await stamp_launched(
         session_key, session, host=host, core=container.core, core_tier=core_tier
     )
@@ -164,7 +169,9 @@ async def run_launch(
             # The pick is the newest capture, which the save archive carries.
             resume_pushed = resume_on_activate
         else:
-            resume_pushed = await states.push_resume_state(container, resume_state)
+            resume_pushed = await states.push_resume_state(
+                container, resume_state, state_core
+            )
 
     await push_to_user(
         session.get("user_id"),
@@ -188,7 +195,7 @@ async def run_launch(
             rom.id,
             container,
             resume_pushed=resume_pushed,
-            core_tier=core_tier,
+            state_core=state_core,
         )
     )
 

backend/handler/streaming/protocol.py

@@ -161,8 +161,11 @@ class WebstationProtocol(BrokerProtocol):
         path = urlparse(host.strip()).path.rstrip("/")
         return not path or path == self.subfolder
 
+    def api_route(self, path: str) -> str:
+        return f"{self.subfolder}/api{path}"
+
     def session_route(self, path: str) -> str:
-        return f"{self.subfolder}/api/session{path}"
+        return self.api_route(f"/session{path}")
 
     def transfer_route(self, path: str) -> str:
         return self.session_route(path)

backend/handler/streaming/states.py

@@ -23,8 +23,9 @@ import os
 import re
 import zipfile
 import zlib
+from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import NamedTuple
+from typing import Any, NamedTuple
 from urllib.parse import quote
 
 from fastapi import HTTPException
@@ -138,6 +139,13 @@ def resolve_resume_state(
     emulator = container.emulator
     native = (state.emulator or "").lower() == emulator
     if native:
+        # Another core's file would land where this one's quick-load reads it.
+        core = state_core_for(container)
+        if core is not None and not core.matches(state.core):
+            raise HTTPException(
+                status_code=400,
+                detail="State was made by a different RetroArch core",
+            )
         slot = slot_from_state_filename(emulator, state.file_name)
         if slot is not None:
             return state, slot, False
@@ -287,12 +295,54 @@ async def store_state_screenshot(
     )
 
 
-def user_states_for_emulator(user_id: int, rom_id: int, emulator: str) -> list[State]:
-    """The user's states for this ROM and emulator, newest first."""
+@dataclass(frozen=True)
+class StateCore:
+    """Which stored states a RetroArch container's core can load."""
+
+    expected: str
+    # Whether `expected` is the platform's default core, which wrote every
+    # state stored before cores were recorded.
+    default_matches: bool
+
+    def matches(self, core: str | None) -> bool:
+        return core == self.expected or (core is None and self.default_matches)
+
+
+def state_core_for(
+    container: ResolvedContainer, launched: dict[str, Any] | None = None
+) -> StateCore | None:
+    """The core filter for this container's states, None to filter nothing.
+
+    `launched` is the activate reply, which names the running core once the
+    game is up. Before that the broker is asked for its default, so this
+    blocks and belongs off the event loop.
+    """
+    if not container.is_webstation or container.emulator.lower() != "retroarch":
+        return None
+    reply = launched if isinstance(launched, dict) else {}
+    booted = reply.get("core")
+    if isinstance(booted, str) and booted:
+        return StateCore(booted, reply.get("core_tier") == "default")
+    default = webstation.default_core(container)
+    expected = container.core or default
+    if expected is None:
+        return None
+    return StateCore(expected, default is not None and expected == default)
+
+
+def user_states_for_emulator(
+    user_id: int, rom_id: int, emulator: str, core: StateCore | None = None
+) -> list[State]:
+    """The user's states for this ROM and emulator, newest first.
+
+    `core` narrows them to what a RetroArch core can load. Prune and dedup
+    leave it out: they work on the whole history, whichever core wrote it.
+    """
     states = [
         s
         for s in db_state_handler.get_states(user_id=user_id, rom_ids=[rom_id])
         if (s.emulator or "").lower() == emulator
+        and (core is None or core.matches(s.core))
     ]
     # Ties on id, because updated_at only has second resolution: two captures
     # in the same second would otherwise order arbitrarily, and only the first
@@ -484,6 +534,13 @@ async def push_resume_state(
     Best-effort: a failure means the session just starts fresh, which the claim
     response reports through `resume`.
     """
+    if state_core is not None and not state_core.matches(resume_state.core):
+        log.warning(
+            "resume state is %s's, core %s booted, launching fresh",
+            resume_state.core,
+            state_core.expected,
+        )
+        return False
     try:
         content = await fs_asset_handler.read_file(
             f"{resume_state.file_path}/{resume_state.file_name}"
@@ -496,6 +553,8 @@ async def push_resume_state(
         container,
         container_state_filename(resume_state.file_name),
         content,
+        # So the broker refuses it too when RomM could not tell which core runs.
+        resume_state.core or (state_core.expected if state_core is not None else None),
     )
     if not pushed:
         log.warning("resume state not pushed, launching fresh")
@@ -507,7 +566,7 @@ async def hydrate_states_to_broker(
     rom_id: int,
     container: ResolvedContainer,
     resume_pushed: bool = False,
-    core_tier: str | None = None,
+    state_core: StateCore | None = None,
 ) -> int:
     """Background task: push the newest stored state for this ROM down to the
     freshly claimed container. Emulators read state files lazily, so pushing
@@ -524,13 +583,9 @@ async def hydrate_states_to_broker(
     """
     if resume_pushed or container.resumes_from_archive:
         return 0
-    # Stored states don't record their core, so each is presumed the default's.
-    if container.core is not None and core_tier != "default":
-        log.info(
-            "not hydrating states into core %s (%s), they were made by the default",
-            container.core,
-            core_tier,
-        )
+    # A configured core with no filter means the launch never named what runs.
+    if container.core is not None and state_core is None:
+        log.info("not hydrating states into core %s, it is unconfirmed", container.core)
         return 0
 
     user = db_user_handler.get_user(user_id)
@@ -539,7 +594,7 @@ async def hydrate_states_to_broker(
         return 0
     emulator = container.emulator
 
-    states = user_states_for_emulator(user_id, rom_id, emulator)
+    states = user_states_for_emulator(user_id, rom_id, emulator, state_core)
     if not states:
         return 0
 
@@ -556,6 +611,7 @@ async def hydrate_states_to_broker(
         container,
         container_state_filename(newest.file_name),
         content,
+        state_core.expected if state_core is not None else None,
     )
     if ok:
         log.info("hydrated newest state to container, rom=%s", rom.name)

backend/handler/streaming/webstation.py

@@ -34,7 +34,7 @@ from fastapi import HTTPException
 from config import STREAMING_LAUNCH_TIMEOUT, STREAMING_SAVE_TIMEOUT
 from handler.streaming import broker
 from handler.streaming.config import ResolvedContainer
-from handler.streaming.protocol import ACK_TIMEOUT
+from handler.streaming.protocol import ACK_TIMEOUT, WebstationProtocol
 from logger.logger import log
 from models.user import User
 
@@ -154,6 +154,49 @@ def import_spec(
     return spec
 
 
+# Short-lived, 404s included: a broker upgrade adds the route and a restart can
+# change the default, and neither should need a RomM restart to be seen.
+_DEFAULT_CORE_TTL = 60.0
+_default_core_cache: dict[tuple[str, str], tuple[float, str | None]] = {}
+
+
+def default_core(container: ResolvedContainer) -> str | None:
+    """The core this broker boots for the platform when config names none.
+
+    None when it can't say: not RetroArch, a broker too old, or unreachable.
+    """
+    protocol = container.protocol
+    if (
+        not isinstance(protocol, WebstationProtocol)
+        or container.emulator.lower() != "retroarch"
+    ):
+        return None
+    cache_key = (container.key, container.platform)
+    cached = _default_core_cache.get(cache_key)
+    if cached is not None and cached[0] > time.monotonic():
+        return cached[1]
+    path = protocol.api_route(
+        f"/retroarch/cores?platform={quote(container.platform, safe='')}"
+    )
+    core: str | None = None
+    try:
+        resp = broker.request(container, path, method="GET", timeout=ACK_TIMEOUT)
+    except urllib.error.HTTPError as exc:
+        code = exc.code
+        exc.close()
+        if code != 404:
+            log.warning("retroarch cores check failed with HTTP %d", code)
+            return None
+    except (urllib.error.URLError, OSError, http.client.HTTPException, ValueError):
+        log.warning("retroarch cores check unreachable, not filtering states by core")
+        return None
+    else:
+        value = resp.get("default") if isinstance(resp, dict) else None
+        core = value if isinstance(value, str) and value else None
+    _default_core_cache[cache_key] = (time.monotonic() + _DEFAULT_CORE_TTL, core)
+    return core
+
+
 def activate(
     container: ResolvedContainer,
     *,

backend/tests/endpoints/test_streaming.py

@@ -3,6 +3,7 @@ import io
 import json
 import logging
 import re
+import threading
 import time
 import zipfile
 from collections.abc import Iterator
@@ -130,6 +131,13 @@ def clear_import_spec_cache():
         yield
 
 
+@pytest.fixture(autouse=True)
+def clear_default_core_cache():
+    """Isolate webstation's default-core cache per test."""
+    with patch.dict(webstation._default_core_cache, clear=True):
+        yield
+
+
 @pytest.fixture(autouse=True)
 def exit_pull_queue():
     with exit_pulls_spawned_inline() as queue:
@@ -390,6 +398,48 @@ 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"})):
+        with (
+            patch("handler.streaming.webstation.import_spec", return_value=None),
+            patch("handler.streaming.webstation.default_core", return_value="snes9x"),
+        ):
+            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}
+
+
+def test_config_asks_a_broker_its_spec_and_core_at_once(client, access_token):
+    """So an unreachable broker costs the play screen one timeout, not two."""
+    both = threading.Barrier(2, timeout=5)
+
+    def meet(*_args, value=None):
+        both.wait()
+        return value
+
+    with _streaming(_webstation(platforms={"snes": "retroarch"})):
+        with (
+            patch("handler.streaming.webstation.import_spec", side_effect=meet),
+            patch(
+                "handler.streaming.webstation.default_core",
+                side_effect=lambda c: meet(value="snes9x"),
+            ),
+        ):
+            r = client.get("/api/streaming/config", headers=_auth(access_token))
+    assert r.status_code == 200
+
+
+def test_config_has_no_state_core_outside_retroarch(client, access_token):
+    with _streaming(_webstation()):
+        with patch("handler.streaming.webstation.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"])
+
+
 def test_get_config_labels_each_platform_by_its_emulator(client, access_token):
     """A shared webstation container names the button after the emulator that
     serves each platform, RetroArch after its core, unless the platform block
@@ -4846,11 +4896,86 @@ def test_hydrate_skips_states_missing_on_disk(rom: Rom, admin_user: User):
     push.assert_not_called()
 
 
-def _hydrate_ra(rom: Rom, user: User, value, core_tier):
-    """Hydrate a RetroArch container serving `rom` as `value`, returning the
-    count and the push mock."""
-    db_state_handler.add_state(_state_for(rom, user, "Game.state", "retroarch"))
-    container = _resolved(_webstation(platforms={rom.platform_slug: value}))
+def _hydrate_core(rom: Rom, user: User, value, state_core):
+    """Hydrate a RetroArch container serving the ROM as `value`, returning the
+    push mock."""
+    with (
+        patch(
+            "handler.filesystem.fs_asset_handler.read_file",
+            new=AsyncMock(return_value=b"state-bytes"),
+        ),
+        patch("handler.streaming.states.push_state_file", return_value=True) as push,
+    ):
+        asyncio.run(
+            states.hydrate_states_to_broker(
+                user.id, rom.id, _snes(value), state_core=state_core
+            )
+        )
+    return push
+
+
+def test_hydrate_pushes_an_unrecorded_state_into_the_default_core(
+    rom: Rom, admin_user: User
+):
+    """Every state stored before cores were recorded is NULL and was written by
+    the default, so an upgrade must keep hydrating it."""
+    _core_state(rom, admin_user, "Game.state", None)
+    push = _hydrate_core(rom, admin_user, "retroarch", states.StateCore("snes9x", True))
+    push.assert_called_once()
+    assert push.call_args.args[1:] == ("Game.state", b"state-bytes", "snes9x")
+
+
+def test_hydrate_skips_a_newer_state_from_another_core(rom: Rom, admin_user: User):
+    """The newest matching state, not the newest overall."""
+    _core_state(rom, admin_user, "Old.state", "bsnes")
+    _core_state(rom, admin_user, "New.state", None)
+    push = _hydrate_core(
+        rom, admin_user, "retroarch:bsnes", states.StateCore("bsnes", False)
+    )
+    assert push.call_args.args[1:] == ("Old.state", b"state-bytes", "bsnes")
+
+
+def test_hydrate_pushes_nothing_when_no_state_matches(rom: Rom, admin_user: User):
+    _core_state(rom, admin_user, "Game.state", None)
+    push = _hydrate_core(
+        rom, admin_user, "retroarch:bsnes", states.StateCore("bsnes", False)
+    )
+    push.assert_not_called()
+
+
+def test_hydrate_filters_nothing_when_the_core_is_unknown(rom: Rom, admin_user: User):
+    """No configured core and a broker that can't name its default is how every
+    RetroArch launch ran before cores were recorded."""
+    _core_state(rom, admin_user, "Game.state", "whatever")
+    push = _hydrate_core(rom, admin_user, "retroarch", None)
+    assert push.call_args.args[1:] == ("Game.state", b"state-bytes", None)
+
+
+def test_hydrate_pushes_nothing_into_a_configured_core_it_cannot_match(
+    rom: Rom, admin_user: User
+):
+    """No filter with a core configured means the launch never said what runs,
+    so nothing goes rather than something that may be the wrong core's."""
+    _core_state(rom, admin_user, "Game.state", "bsnes")
+    push = _hydrate_core(rom, admin_user, "retroarch:bsnes", None)
+    push.assert_not_called()
+
+
+def test_the_launch_hands_hydration_the_core_activate_booted(
+    client, access_token, rom: Rom
+):
+    with patch(
+        "handler.streaming.states.hydrate_states_to_broker",
+        new=AsyncMock(return_value=0),
+    ) as hydrate:
+        _ra_claim(client, access_token, rom, "retroarch:bsnes", _BSNES_REPLY)
+    assert hydrate.call_args.kwargs["state_core"] == states.StateCore("bsnes", False)
+
+
+def _push_resume(rom: Rom, user: User, recorded, state_core):
+    """Push a resume state recorded as written by `recorded`, returning
+    (pushed, push mock)."""
+    state = _core_state(rom, user, "Game.state", recorded)
     with (
         patch(
             "handler.filesystem.fs_asset_handler.read_file",
@@ -4859,43 +4984,73 @@ def _hydrate_ra(rom: Rom, user: User, value, core_tier):
         patch("handler.streaming.states.push_state_file", return_value=True) as push,
     ):
         pushed = asyncio.run(
-            states.hydrate_states_to_broker(
-                user.id, rom.id, container, core_tier=core_tier
-            )
+            states.push_resume_state(_snes("retroarch:bsnes"), state, state_core)
         )
     return pushed, push
 
 
-@pytest.mark.parametrize("tier", ["vetted", "untested", "blocked", None])
-def test_hydrate_pushes_nothing_into_a_non_default_core(
-    rom: Rom, admin_user: User, tier
+@pytest.mark.parametrize(
+    ("recorded", "state_core", "label"),
+    [
+        ("bsnes", states.StateCore("bsnes", False), "bsnes"),
+        (None, states.StateCore("snes9x", True), "snes9x"),
+        # The broker's check is the backstop when RomM can't tell the core.
+        ("bsnes", None, "bsnes"),
+        (None, None, None),
+    ],
+)
+def test_a_resume_push_names_the_core_that_wrote_the_state(
+    rom: Rom, admin_user: User, recorded, state_core, label
 ):
-    """Stored states don't record their core yet, so every one is presumed the
-    default's; pushed into another core it lands where quick-load reads it."""
-    pushed, push = _hydrate_ra(rom, admin_user, "retroarch:bsnes", tier)
-    assert pushed == 0
+    pushed, push = _push_resume(rom, admin_user, recorded, state_core)
+    assert pushed is True
+    assert push.call_args.args[3] == label
+
+
+@pytest.mark.parametrize("recorded", ["snes9x", None])
+def test_a_resume_push_skips_a_state_the_booted_core_cannot_load(
+    rom: Rom, admin_user: User, recorded
+):
+    """The claim's check can disagree with the core activate booted, and the
+    session then starts fresh rather than quick-loading the wrong core's file."""
+    pushed, push = _push_resume(
+        rom, admin_user, recorded, states.StateCore("bsnes", False)
+    )
+    assert pushed is False
     push.assert_not_called()
 
 
-def test_hydrate_still_pushes_when_the_configured_core_is_the_default(
+def test_resume_refuses_a_state_another_core_wrote(rom: Rom, admin_user: User):
+    """A picker opened before the config change still lists it, and the claim
+    is the last place to stop it."""
+    state = _core_state(rom, admin_user, "Game.state", "snes9x")
+    with patch("handler.streaming.webstation.default_core", return_value="snes9x"):
+        with pytest.raises(HTTPException) as err:
+            states.resolve_resume_state(
+                admin_user.id, rom, _snes("retroarch:bsnes"), state.id
+            )
+    assert err.value.status_code == 400
+    assert "different RetroArch core" in err.value.detail
+
+
+def test_resume_takes_a_state_its_core_wrote(rom: Rom, admin_user: User):
+    state = _core_state(rom, admin_user, "Game.state3", "bsnes")
+    with patch("handler.streaming.webstation.default_core", return_value="snes9x"):
+        picked, slot, foreign = states.resolve_resume_state(
+            admin_user.id, rom, _snes("retroarch:bsnes"), state.id
+        )
+    assert (picked.id, slot, foreign) == (state.id, 3, False)
+
+
+def test_resume_takes_an_unrecorded_state_on_the_default_core(
     rom: Rom, admin_user: User
 ):
-    pushed, _ = _hydrate_ra(rom, admin_user, "retroarch:snes9x", "default")
-    assert pushed == 1
-
-
-def test_hydrate_is_unchanged_without_a_core(rom: Rom, admin_user: User):
-    pushed, _ = _hydrate_ra(rom, admin_user, "retroarch", None)
-    assert pushed == 1
-
-
-def test_the_launch_hands_the_core_tier_to_hydration(client, access_token, rom: Rom):
-    with patch(
-        "handler.streaming.states.hydrate_states_to_broker",
-        new=AsyncMock(return_value=0),
-    ) as hydrate:
-        _ra_claim(client, access_token, rom, "retroarch:bsnes", _BSNES_REPLY)
-    assert hydrate.call_args.kwargs["core_tier"] == "untested"
+    state = _core_state(rom, admin_user, "Game.state", None)
+    with patch("handler.streaming.webstation.default_core", return_value="snes9x"):
+        _, slot, _ = states.resolve_resume_state(
+            admin_user.id, rom, _snes("retroarch"), state.id
+        )
+    assert slot == 0
 
 
 def _add_state_at(rom: Rom, user: User, file_name: str, day: int) -> State:
@@ -7649,6 +7804,110 @@ def test_a_pulled_state_is_stored_with_its_core(rom: Rom, admin_user: User):
     assert [s.core for s in stored] == ["bsnes"]
 
 
+_CORES_REPLY = {"platform": "snes", "default": "snes9x", "cores": []}
+
+
+def test_default_core_asks_the_broker_cores_route():
+    with patch(
+        "handler.streaming.broker.request", return_value=_CORES_REPLY
+    ) as request:
+        assert webstation.default_core(_snes("retroarch")) == "snes9x"
+    assert request.call_args.args[1] == "/streaming/api/retroarch/cores?platform=snes"
+
+
+def test_default_core_answers_from_cache_within_the_ttl():
+    with patch(
+        "handler.streaming.broker.request", return_value=_CORES_REPLY
+    ) as request:
+        webstation.default_core(_snes("retroarch"))
+        webstation.default_core(_snes("retroarch"))
+    assert request.call_count == 1
+
+
+def test_default_core_asks_again_once_a_404_expires():
+    """An upgraded broker gains the route, so an old one's 404 must not outlive
+    the TTL."""
+    with (
+        patch("handler.streaming.webstation._DEFAULT_CORE_TTL", 0.0),
+        patch(
+            "handler.streaming.broker.request", side_effect=_http_error(404)
+        ) as request,
+    ):
+        assert webstation.default_core(_snes("retroarch")) is None
+        webstation.default_core(_snes("retroarch"))
+    assert request.call_count == 2
+
+
+def test_default_core_is_not_asked_outside_retroarch():
+    with patch("handler.streaming.broker.request") as request:
+        assert webstation.default_core(_resolved(_webstation())) is None
+    request.assert_not_called()
+
+
+def test_state_core_matches_its_own_core_and_unrecorded_default_states():
+    assert states.StateCore("bsnes", False).matches("bsnes")
+    assert not states.StateCore("bsnes", False).matches(None)
+    assert not states.StateCore("bsnes", False).matches("snes9x")
+    assert states.StateCore("snes9x", True).matches(None)
+
+
+def test_state_core_takes_the_core_activate_booted():
+    """Once the game is up, the reply says what runs; no second request."""
+    with patch("handler.streaming.webstation.default_core") as default_core:
+        booted = states.state_core_for(
+            _snes("retroarch:bsnes"), {"core": "bsnes", "core_tier": "untested"}
+        )
+        default = states.state_core_for(
+            _snes("retroarch"), {"core": "snes9x", "core_tier": "default"}
+        )
+    assert booted == states.StateCore("bsnes", default_matches=False)
+    assert default == states.StateCore("snes9x", default_matches=True)
+    default_core.assert_not_called()
+
+
+def test_state_core_asks_for_the_default_before_launch():
+    with patch("handler.streaming.webstation.default_core", return_value="snes9x"):
+        configured = states.state_core_for(_snes("retroarch:bsnes"))
+        default = states.state_core_for(_snes("retroarch"))
+    assert configured == states.StateCore("bsnes", default_matches=False)
+    assert default == states.StateCore("snes9x", default_matches=True)
+
+
+def test_state_core_is_unknown_when_the_broker_cannot_say():
+    """An older broker has no cores route; filtering nothing is what RomM did
+    before it knew about cores."""
+    with patch("handler.streaming.webstation.default_core", return_value=None):
+        assert states.state_core_for(_snes("retroarch")) is None
+
+
+def test_a_configured_core_still_filters_when_the_default_is_unknown():
+    """A NULL row is the default's, and the default is not known to be this
+    core, so it stays hidden rather than risk a quick-load of the wrong file."""
+    with patch("handler.streaming.webstation.default_core", return_value=None):
+        core = states.state_core_for(_snes("retroarch:bsnes"))
+    assert core == states.StateCore("bsnes", default_matches=False)
+
+
+def test_no_state_core_outside_retroarch():
+    assert states.state_core_for(_resolved(_webstation())) is None
+
+
+def test_user_states_narrow_to_a_core(rom: Rom, admin_user: User):
+    _core_state(rom, admin_user, "A.state", None)
+    _core_state(rom, admin_user, "B.state", "bsnes")
+    _core_state(rom, admin_user, "C.state", "snes9x")
+
+    def names(core):
+        found = states.user_states_for_emulator(
+            admin_user.id, rom.id, "retroarch", core
+        )
+        return sorted(s.file_name for s in found)
+
+    assert names(None) == ["A.state", "B.state", "C.state"]
+    assert names(states.StateCore("bsnes", False)) == ["B.state"]
+    assert names(states.StateCore("snes9x", True)) == ["A.state", "C.state"]
+
+
 def test_pull_state_to_library_runs_for_a_webstation_container(
     rom: Rom, admin_user: User
 ):