Wait for Steam shortcut overviews before cold launches (#758)

tools/linuxfs/overlay/usr/local/bin/droiddeck-agent

@@ -477,6 +477,55 @@ def steam_selection(agent_dir, timeout=5):
     return {"appId": int(value)}
 
 
+def steam_shortcut_app_id(game_id):
+    """Return the 32-bit app id in Steam's shortcut CGameID form, or None for other ids.
+
+    DroidDeck's shortcut IDs are `(appid << 32) | 0x02000000`. The URL still needs that full
+    CGameID for RunGame; the app-details API separately needs the unsigned 32-bit app id.
+    """
+    if not isinstance(game_id, str) or not game_id.isascii() or not game_id.isdigit():
+        return None
+    try:
+        game_id_value = int(game_id)
+    except ValueError:
+        return None
+    if str(game_id_value) != game_id or not 0 < game_id_value <= 0xFFFFFFFFFFFFFFFF:
+        return None
+    if game_id_value <= 0xFFFFFFFF or game_id_value & 0xFFFFFFFF != 0x02000000:
+        return None
+    app_id = game_id_value >> 32
+    return app_id if 0 < app_id <= 0xFFFFFFFF else None
+
+
+def wait_for_steam_shortcut_overview(agent_dir, game_id, timeout=45):
+    """Wait until Steam can map a shortcut CGameID to its 32-bit app overview."""
+    app_id = steam_shortcut_app_id(game_id)
+    if app_id is None:
+        return {"supported": False, "ready": False}
+
+    timeout = request_timeout({"timeout": timeout}, 45)
+    deadline = time.monotonic() + timeout
+    expression = """(() => {
+        const store = globalThis.appStore;
+        if (typeof store?.GetAppOverviewByGameID !== 'function') return null;
+        const overview = store.GetAppOverviewByGameID(%s);
+        return overview == null ? null : String(overview.appid);
+    })()""" % json.dumps(game_id)
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            break
+        try:
+            result = cdp_eval(agent_dir, expression, await_promise=False,
+                              timeout=min(5, max(0.25, remaining)))
+            if result.get("value") == str(app_id):
+                return {"supported": True, "ready": True, "appId": app_id}
+        except AgentError as error:
+            if error.code not in ("NO_DEVTOOLS", "DEVTOOLS_UNREACHABLE", "NO_TARGET", "TIMEOUT"):
+                raise
+        time.sleep(min(0.25, max(0, deadline - time.monotonic())))
+    raise AgentError("STEAM_OVERVIEW_TIMEOUT", "Steam did not load shortcut app overview %d" % app_id)
+
 
 def game_pids(app_id):
     """Our processes started for this Steam app (Steam sets SteamAppId / SteamGameId in their
@@ -740,6 +789,21 @@ def main(argv):
     if not base:
         print("BL_LAUNCH_DIR is not set", file=sys.stderr)
         return 2
+    if argv and argv[0] == "--wait-steam-shortcut-overview":
+        if len(argv) not in (2, 3):
+            print("usage: droiddeck-agent --wait-steam-shortcut-overview <cgameid> [timeout-seconds]", file=sys.stderr)
+            return 2
+        timeout = argv[2] if len(argv) == 3 else 45
+        try:
+            prepared = wait_for_steam_shortcut_overview(Path(base) / "agent", argv[1], timeout)
+        except (AgentError, OSError, ValueError) as error:
+            print("droiddeck-agent: shortcut overview was not ready: %s" % error, file=sys.stderr)
+            return 1
+        if not prepared["supported"]:
+            print("droiddeck-agent: id is not a DroidDeck shortcut CGameID")
+            return 3
+        print("droiddeck-agent: Steam overview is ready for shortcut app %d" % prepared["appId"])
+        return 0
     serve(Path(base) / "agent", focus="--no-focus" not in argv)
     return 0
 

tools/linuxfs/overlay/usr/local/bin/droiddeck-session

@@ -571,13 +571,28 @@ case $mode in
       if [ -n "$_bl_held" ]; then
         set -- "${_bl_rest[@]}"
         (
+          _bl_launch_ready=
           _bl_end=$((SECONDS + 120))
           until [ "$(xprop -root GAMESCOPE_FOCUSED_APP 2>/dev/null | sed -n 's/.*= *//p')" = 769 ] || [ "$SECONDS" -ge "$_bl_end" ]; do
             sleep 1
           done
-          sleep 4
-          echo "== steam: interface up; launching the requested game $_bl_held"
-          printf '%s\n' "$_bl_held" > "$BL_LAUNCH_DIR/steam-game.tmp" && mv -f "$BL_LAUNCH_DIR/steam-game.tmp" "$BL_LAUNCH_DIR/steam-game"
+          if /usr/local/bin/droiddeck-agent --wait-steam-shortcut-overview "$_bl_held" 45; then
+            echo "== steam: shortcut overview ready; launching the requested game $_bl_held"
+            _bl_launch_ready=1
+          else
+            _bl_overview_status=$?
+            if [ "$_bl_overview_status" -eq 3 ]; then
+              echo "== steam: not a shortcut CGameID; keeping the existing launch path"
+              sleep 4
+              _bl_launch_ready=1
+            else
+              echo "== steam: expected shortcut overview did not become ready; holding the launch URL"
+            fi
+          fi
+          if [ "$_bl_launch_ready" = 1 ]; then
+            echo "== steam: interface up; launching the requested game $_bl_held"
+            printf '%s\n' "$_bl_held" > "$BL_LAUNCH_DIR/steam-game.tmp" && mv -f "$BL_LAUNCH_DIR/steam-game.tmp" "$BL_LAUNCH_DIR/steam-game"
+          fi
         ) &
       fi
     fi

tools/tests/test_droiddeck_agent.py

@@ -255,6 +255,42 @@ class AgentTest(unittest.TestCase):
             answer = self.ask({"kind": "steam-selection"})
         self.assertEqual("INVALID_SELECTION", answer["error"]["code"])
 
+    def test_shortcut_cgameid_waits_for_exact_steam_overview_mapping(self):
+        game_id = str((3551182975 << 32) | 0x02000000)
+        self.assertEqual(3551182975, agent.steam_shortcut_app_id(game_id))
+        self.assertIsNone(agent.steam_shortcut_app_id("413150"))
+        self.assertIsNone(agent.steam_shortcut_app_id(str((3551182975 << 32) | 0x03000000)))
+        clock = [0.0]
+        with patch.object(agent, "cdp_eval", side_effect=[{"value": None}, {"value": "3551182975"}]) as evaluate, \
+             patch.object(agent.time, "monotonic", side_effect=lambda: clock[0]), \
+             patch.object(agent.time, "sleep", side_effect=lambda delay: clock.__setitem__(0, clock[0] + delay)):
+            result = agent.wait_for_steam_shortcut_overview(self.dir, game_id, timeout=1)
+        self.assertEqual({"supported": True, "ready": True, "appId": 3551182975}, result)
+        expression = evaluate.call_args_list[0].args[1]
+        self.assertIn("GetAppOverviewByGameID", expression)
+        self.assertIn(json.dumps(game_id), expression)
+        self.assertNotIn("RequestAppDetails", expression)
+        self.assertFalse(evaluate.call_args_list[0].kwargs["await_promise"])
+
+    def test_wrong_overview_app_id_keeps_shortcut_url_held_until_timeout(self):
+        game_id = str((3537741571 << 32) | 0x02000000)
+        clock = [0.0]
+        with patch.object(agent, "cdp_eval", return_value={"value": "646570"}) as evaluate, \
+             patch.object(agent.time, "monotonic", side_effect=lambda: clock[0]), \
+             patch.object(agent.time, "sleep", side_effect=lambda delay: clock.__setitem__(0, clock[0] + delay)):
+            with self.assertRaises(agent.AgentError) as raised:
+                agent.wait_for_steam_shortcut_overview(self.dir, game_id, timeout=0.5)
+        self.assertEqual("STEAM_OVERVIEW_TIMEOUT", raised.exception.code)
+        self.assertGreater(evaluate.call_count, 0)
+        self.assertTrue(all("RequestAppDetails" not in call.args[1] for call in evaluate.call_args_list))
+
+    def test_normal_steam_ids_and_unknown_cgameids_skip_overview_wait(self):
+        for game_id in ("413150", str((123456789 << 32) | 0x03000000), "18446744073709551616"):
+            with self.subTest(game_id=game_id), patch.object(agent, "cdp_eval") as evaluate:
+                result = agent.wait_for_steam_shortcut_overview(self.dir, game_id)
+            self.assertFalse(result["supported"])
+            evaluate.assert_not_called()
+
     def test_cdp_exception(self):
         server = FakeDevTools({"result": {"result": {"type": "object"}, "exceptionDetails": {
             "text": "Uncaught", "exception": {"description": "ReferenceError: SteamClient is not defined"}}}})

tools/tests/test_steam_cold_launch.py

@@ -0,0 +1,90 @@
+"""Cold shortcut launches wait for Steam's CGameID-to-app overview mapping."""
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import textwrap
+import unittest
+
+SESSION = Path(__file__).resolve().parents[1] / "linuxfs/overlay/usr/local/bin/droiddeck-session"
+SOURCE = SESSION.read_text()
+END = SOURCE.index(
+    '    if [ -s "$HOME/.bl-autolaunch" ]; then',
+    SOURCE.index("          _bl_launch_ready="),
+)
+START = SOURCE.rfind(
+    '    if [ -n "${BL_LAUNCH_DIR:-}" ] && [ -z "${BL_DESKTOP:-}" ]; then',
+    0,
+    END,
+)
+assert START >= 0
+LAUNCH_BLOCK = textwrap.dedent(SOURCE[START:END])
+
+
+class SteamColdLaunchTest(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.launch_dir = self.root / "session"
+        self.launch_dir.mkdir()
+        self.bin = self.root / "bin"
+        self.bin.mkdir()
+        self.agent_args = self.root / "agent-args"
+        self.sleep_args = self.root / "sleep-args"
+
+        xprop = self.bin / "xprop"
+        xprop.write_text("#!/bin/sh\necho 'GAMESCOPE_FOCUSED_APP(CARDINAL) = 769'\n")
+        xprop.chmod(0o755)
+        agent = self.bin / "droiddeck-agent"
+        agent.write_text("#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$AGENT_ARGS\"\nexit \"$AGENT_STATUS\"\n")
+        agent.chmod(0o755)
+        sleep = self.bin / "sleep"
+        sleep.write_text("#!/bin/sh\nprintf '%s\\n' \"$@\" >> \"$SLEEP_ARGS\"\n")
+        sleep.chmod(0o755)
+        self.block = LAUNCH_BLOCK.replace("/usr/local/bin/droiddeck-agent", str(agent))
+
+    def run_launch(self, target, helper_status):
+        env = dict(os.environ, HOME=str(self.root), BL_LAUNCH_DIR=str(self.launch_dir),
+                   PATH=str(self.bin) + os.pathsep + os.environ["PATH"],
+                   AGENT_ARGS=str(self.agent_args), SLEEP_ARGS=str(self.sleep_args),
+                   AGENT_STATUS=str(helper_status))
+        script = self.block + "\nprintf 'ARG:%s\\n' \"$@\"\nwait\n"
+        url = target if target.startswith("steam://") else "steam://rungameid/" + target
+        result = subprocess.run(["bash", "-c", script, "cold-launch", url],
+                                env=env, text=True, capture_output=True, timeout=10)
+        self.assertEqual(0, result.returncode, result.stderr)
+        return result
+
+    def test_shortcut_overview_gate_releases_original_64_bit_cgameid_without_fixed_sleep(self):
+        game_id = str((3551182975 << 32) | 0x02000000)
+        result = self.run_launch(game_id, 0)
+        self.assertEqual("steam-game", (self.launch_dir / "steam-game").name)
+        self.assertEqual(game_id, (self.launch_dir / "steam-game").read_text().strip())
+        self.assertEqual(["--wait-steam-shortcut-overview", game_id, "45"], self.agent_args.read_text().splitlines())
+        self.assertFalse(self.sleep_args.exists(), result.stdout)
+
+    def test_native_app_id_keeps_legacy_bounded_launch_path(self):
+        game_id = "413150"
+        self.run_launch(game_id, 3)
+        self.assertEqual(game_id, (self.launch_dir / "steam-game").read_text().strip())
+        self.assertEqual(["--wait-steam-shortcut-overview", game_id, "45"], self.agent_args.read_text().splitlines())
+        self.assertEqual("4", self.sleep_args.read_text().strip())
+
+    def test_shortcut_without_the_expected_overview_does_not_forward_cgameid(self):
+        game_id = str((3551182975 << 32) | 0x02000000)
+        result = self.run_launch(game_id, 1)
+        self.assertFalse((self.launch_dir / "steam-game").exists(), result.stdout)
+        self.assertEqual(["--wait-steam-shortcut-overview", game_id, "45"], self.agent_args.read_text().splitlines())
+        self.assertFalse(self.sleep_args.exists())
+
+    def test_non_game_steam_urls_remain_on_the_original_startup_arguments(self):
+        url = "steam://open/library"
+        result = self.run_launch(url, 0)
+        self.assertIn("ARG:" + url, result.stdout)
+        self.assertFalse((self.launch_dir / "steam-game").exists())
+        self.assertFalse(self.agent_args.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()