添加 Android 调试场景工具

.gitignore

@@ -18,3 +18,6 @@ node_modules/
 .kotlin/
 /.vscode/settings.json
 /.codex/skills
+/.debug-runs/
+__pycache__/
+*.py[cod]

app/src/debug/AndroidManifest.xml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="utf-8"?>
+<manifest xmlns:android="http://schemas.android.com/apk/res/android">
+    <application>
+        <activity
+            android:name="io.legado.app.debug.DebugScenarioActivity"
+            android:exported="true"
+            android:theme="@style/AppTheme.Transparent" />
+    </application>
+</manifest>

app/src/debug/assets/debug-fixtures/long-chapter-v1.json

@@ -0,0 +1,27 @@
+{
+  "schemaVersion": 1,
+  "id": "long-chapter-v1",
+  "resetPolicy": "replace",
+  "book": {
+    "bookUrl": "legado-debug://fixture/long-chapter-v1",
+    "name": "调试夹具：长章节",
+    "author": "Legado Debug",
+    "pageMode": "simulation",
+    "startChapter": 0,
+    "startPosition": 0
+  },
+  "chapters": [
+    {
+      "title": "第一章 稳定翻页",
+      "readyMarker": "LEGADO_DEBUG_LONG_CHAPTER_V1",
+      "paragraph": "这是用于阅读页翻页调试的固定长章节。每一段都带有稳定编号，用于确认页面内容确实发生变化。天地玄黄，宇宙洪荒；日月盈昃，辰宿列张。",
+      "repeat": 180
+    },
+    {
+      "title": "第二章 边界验证",
+      "readyMarker": "LEGADO_DEBUG_LONG_CHAPTER_V1_CHAPTER_2",
+      "paragraph": "这是第二个固定章节，用于连续翻页越过章节边界后的行为验证。内容完全离线，不依赖书源、网络或设备历史状态。",
+      "repeat": 80
+    }
+  ]
+}

app/src/debug/java/io/legado/app/debug/DebugScenarioActivity.kt

@@ -0,0 +1,153 @@
+package io.legado.app.debug
+
+import android.content.Intent
+import android.os.Bundle
+import android.util.Log
+import androidx.appcompat.app.AppCompatActivity
+import androidx.lifecycle.lifecycleScope
+import androidx.room.withTransaction
+import io.legado.app.constant.BookType
+import io.legado.app.constant.PageAnim
+import io.legado.app.data.appDb
+import io.legado.app.data.entities.Book
+import io.legado.app.data.entities.BookChapter
+import io.legado.app.data.entities.BookSource
+import io.legado.app.help.book.BookHelp
+import io.legado.app.ui.main.MainIntent
+import io.legado.app.utils.GSON
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+
+class DebugScenarioActivity : AppCompatActivity() {
+
+    override fun onCreate(savedInstanceState: Bundle?) {
+        super.onCreate(savedInstanceState)
+        val fixtureId = intent.getStringExtra(EXTRA_FIXTURE)
+        val sessionId = intent.getStringExtra(EXTRA_SESSION).orEmpty()
+        if (fixtureId == null || !FIXTURE_ID.matches(fixtureId)) {
+            fail(sessionId, "invalid fixture ID")
+            return
+        }
+
+        lifecycleScope.launch {
+            runCatching {
+                withContext(Dispatchers.IO) { installFixture(fixtureId) }
+            }.onSuccess { bookUrl ->
+                Log.i(LOG_TAG, "[session=$sessionId] FIXTURE_READY fixture=$fixtureId")
+                startActivity(
+                    MainIntent.createReadBookIntent(this@DebugScenarioActivity, bookUrl)
+                        .addFlags(Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_CLEAR_TASK)
+                )
+            }.onFailure { error ->
+                fail(sessionId, "fixture import failed", error)
+            }
+        }
+    }
+
+    private suspend fun installFixture(fixtureId: String): String {
+        val fixture = assets.open("debug-fixtures/$fixtureId.json").bufferedReader().use {
+            GSON.fromJson(it, DebugFixture::class.java)
+        }
+        require(fixture.schemaVersion == 1) { "unsupported fixture schema" }
+        require(fixture.id == fixtureId) { "fixture ID mismatch" }
+        require(fixture.resetPolicy == "replace") { "unsupported reset policy" }
+        require(fixture.chapters.isNotEmpty()) { "fixture must contain chapters" }
+        require(fixture.book.startChapter in fixture.chapters.indices) { "invalid start chapter" }
+
+        val pageAnim = when (fixture.book.pageMode) {
+            "simulation" -> PageAnim.simulationPageAnim
+            else -> error("unsupported page mode: ${fixture.book.pageMode}")
+        }
+        val chapters = fixture.chapters.mapIndexed { index, chapter ->
+            require(chapter.repeat in 1..1000) { "invalid repeat for chapter $index" }
+            BookChapter(
+                url = "${fixture.book.bookUrl}/chapter/$index",
+                title = chapter.title,
+                baseUrl = fixture.book.bookUrl,
+                bookUrl = fixture.book.bookUrl,
+                index = index,
+            )
+        }
+        val book = Book(
+            bookUrl = fixture.book.bookUrl,
+            tocUrl = "${fixture.book.bookUrl}/toc",
+            origin = DEBUG_ORIGIN,
+            originName = "Legado Debug Fixture",
+            name = fixture.book.name,
+            author = fixture.book.author,
+            type = BookType.text,
+            totalChapterNum = chapters.size,
+            durChapterTitle = chapters[fixture.book.startChapter].title,
+            durChapterIndex = fixture.book.startChapter,
+            durChapterPos = fixture.book.startPosition,
+            durChapterTime = System.currentTimeMillis(),
+            canUpdate = false,
+            readConfig = Book.ReadConfig(pageAnim = pageAnim),
+        )
+
+        appDb.withTransaction {
+            appDb.bookSourceDao.insert(
+                BookSource(
+                    bookSourceUrl = DEBUG_ORIGIN,
+                    bookSourceName = "Legado Debug Fixture",
+                    enabled = false,
+                    enabledExplore = false,
+                )
+            )
+            appDb.bookDao.getBook(book.bookUrl)?.let { appDb.bookDao.delete(it) }
+            appDb.bookDao.insert(book)
+            appDb.bookChapterDao.insert(*chapters.toTypedArray())
+        }
+        chapters.forEachIndexed { index, chapter ->
+            val definition = fixture.chapters[index]
+            val content = buildString {
+                appendLine(definition.readyMarker)
+                repeat(definition.repeat) { paragraphIndex ->
+                    append(paragraphIndex + 1)
+                    append(". ")
+                    appendLine(definition.paragraph)
+                }
+            }
+            BookHelp.saveText(book, chapter, content)
+        }
+        return book.bookUrl
+    }
+
+    private fun fail(sessionId: String, message: String, error: Throwable? = null) {
+        Log.e(LOG_TAG, "[session=$sessionId] FIXTURE_ERROR $message", error)
+        finishAffinity()
+    }
+
+    private data class DebugFixture(
+        val schemaVersion: Int,
+        val id: String,
+        val resetPolicy: String,
+        val book: DebugBook,
+        val chapters: List<DebugChapter>,
+    )
+
+    private data class DebugBook(
+        val bookUrl: String,
+        val name: String,
+        val author: String,
+        val pageMode: String,
+        val startChapter: Int,
+        val startPosition: Int,
+    )
+
+    private data class DebugChapter(
+        val title: String,
+        val readyMarker: String,
+        val paragraph: String,
+        val repeat: Int,
+    )
+
+    private companion object {
+        const val EXTRA_FIXTURE = "fixture"
+        const val EXTRA_SESSION = "session"
+        const val LOG_TAG = "LegadoDebug"
+        const val DEBUG_ORIGIN = "legado-debug://fixture-source"
+        val FIXTURE_ID = Regex("[a-z0-9][a-z0-9-]*")
+    }
+}

tools/android/README.md

@@ -0,0 +1,47 @@
+# Android 调试场景
+
+项目级 Android 调试入口。它负责构建、安装、恢复 fixture、执行领域动作，并将结果写入
+`.debug-runs/<session-id>/summary.json`。
+
+当前只提供首个端到端场景：
+
+```bash
+tools/android/debug.sh run reader/page-turn \
+  --device emulator-5554 \
+  --fixture long-chapter-v1 \
+  --action-repeat 20
+```
+
+当只有一台 ready 状态的设备时可以省略 `--device`。`--action-repeat` 表示在同一次应用启动中
+重复领域动作；它不等同于完整重置后多次运行场景。
+
+退出码：
+
+| 退出码 | 含义 |
+| --- | --- |
+| 0 | 通过 |
+| 1 | 行为断言失败 |
+| 2 | 应用崩溃或 ANR |
+| 3 | 构建或安装失败 |
+| 4 | 环境或设备失败 |
+| 5 | 场景定义错误 |
+
+fixture 位于 `app/src/debug/assets/debug-fixtures/`，由只存在于 debug 包中的
+`DebugScenarioActivity` 导入。场景只使用 `open_reader` 和 `turn_page` 两个领域动作；这里刻意
+不提供坐标级通用 DSL。
+
+`summary.json` 是主要机器输入。`failure.suspectedFrame` 仅在堆栈中成功解析到应用代码帧时出现，
+并明确携带 `confidence`；`crash.txt` 始终是原始证据。
+
+运行器不会在每次场景后停止 Gradle daemon，以免破坏增量构建。开发机需要回收构建进程时，
+在本轮工作结束后执行：
+
+```bash
+./gradlew --stop
+```
+
+运行 Python 单元测试：
+
+```bash
+python3 -m unittest discover -s tools/android/tests -v
+```

tools/android/debug.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+exec python3 "$SCRIPT_DIR/runner.py" "$@"

tools/android/runner.py

@@ -0,0 +1,587 @@
+#!/usr/bin/env python3
+"""Deterministic Android debug scenario runner for Legado."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import hashlib
+import json
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import time
+from typing import Any, Sequence
+
+
+EXIT_ASSERTION = 1
+EXIT_CRASH = 2
+EXIT_BUILD = 3
+EXIT_ENVIRONMENT = 4
+EXIT_SCENARIO = 5
+
+PACKAGE = "io.legato.kazusa.debug"
+DEBUG_ACTIVITY = "io.legado.app.debug.DebugScenarioActivity"
+MAIN_ACTIVITY = "io.legado.app.ui.main.MainActivity"
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+PROJECT_ROOT = SCRIPT_DIR.parent.parent
+SCENARIO_ROOT = SCRIPT_DIR / "scenarios"
+RUN_ROOT = PROJECT_ROOT / ".debug-runs"
+FIXTURE_ROOT = PROJECT_ROOT / "app" / "src" / "debug" / "assets" / "debug-fixtures"
+
+
+class ScenarioError(ValueError):
+    pass
+
+
+class CommandError(RuntimeError):
+    def __init__(self, command: Sequence[str], result: subprocess.CompletedProcess[str]):
+        self.command = list(command)
+        self.result = result
+        super().__init__(f"command failed ({result.returncode}): {' '.join(command)}")
+
+
+def load_scenario(name: str) -> dict[str, Any]:
+    if not re.fullmatch(r"[a-z0-9][a-z0-9/_-]*", name):
+        raise ScenarioError(f"invalid scenario name: {name}")
+    path = (SCENARIO_ROOT / f"{name}.json").resolve()
+    if SCENARIO_ROOT.resolve() not in path.parents or not path.is_file():
+        raise ScenarioError(f"scenario not found: {name}")
+    try:
+        scenario = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        raise ScenarioError(f"cannot read scenario {name}: {error}") from error
+    validate_scenario(scenario, name)
+    return scenario
+
+
+def validate_scenario(scenario: Any, requested_name: str | None = None) -> None:
+    if not isinstance(scenario, dict):
+        raise ScenarioError("scenario root must be an object")
+    if scenario.get("schemaVersion") != 1:
+        raise ScenarioError("schemaVersion must be 1")
+    name = scenario.get("name")
+    if not isinstance(name, str) or not name:
+        raise ScenarioError("name must be a non-empty string")
+    if requested_name is not None and name != requested_name:
+        raise ScenarioError(f"scenario name mismatch: expected {requested_name}, got {name}")
+    fixture = scenario.get("fixture")
+    if not isinstance(fixture, str) or not re.fullmatch(r"[a-z0-9][a-z0-9-]*", fixture):
+        raise ScenarioError("fixture must be a lowercase ID")
+    marker = scenario.get("readyMarker")
+    if not isinstance(marker, str) or not marker.strip():
+        raise ScenarioError("readyMarker must be a non-empty string")
+    timeout = scenario.get("readyTimeoutSeconds", 30)
+    if not isinstance(timeout, int) or timeout < 1 or timeout > 300:
+        raise ScenarioError("readyTimeoutSeconds must be between 1 and 300")
+
+    actions = scenario.get("actions")
+    if not isinstance(actions, list) or not actions:
+        raise ScenarioError("actions must be a non-empty array")
+    for index, action in enumerate(actions):
+        if not isinstance(action, dict) or action.get("type") not in {"open_reader", "turn_page"}:
+            raise ScenarioError(f"actions[{index}] has an unsupported type")
+        if action["type"] == "turn_page":
+            if action.get("direction") not in {"next", "previous"}:
+                raise ScenarioError(f"actions[{index}].direction must be next or previous")
+            count = action.get("count", 1)
+            if not isinstance(count, int) or count < 1 or count > 100:
+                raise ScenarioError(f"actions[{index}].count must be between 1 and 100")
+
+    assertions = scenario.get("assertions")
+    supported = {
+        "reader_ready",
+        "reader_activity_visible",
+        "page_content_changed",
+        "no_fatal_exception",
+    }
+    if not isinstance(assertions, list) or not assertions:
+        raise ScenarioError("assertions must be a non-empty array")
+    unknown = set(assertions) - supported
+    if unknown:
+        raise ScenarioError(f"unsupported assertions: {', '.join(sorted(unknown))}")
+
+
+def validate_fixture(fixture_id: str) -> dict[str, Any]:
+    if not re.fullmatch(r"[a-z0-9][a-z0-9-]*", fixture_id):
+        raise ScenarioError("fixture must be a lowercase ID")
+    path = FIXTURE_ROOT / f"{fixture_id}.json"
+    if not path.is_file():
+        raise ScenarioError(f"fixture not found: {fixture_id}")
+    try:
+        fixture = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        raise ScenarioError(f"cannot read fixture {fixture_id}: {error}") from error
+    validate_fixture_data(fixture, fixture_id)
+    return fixture
+
+
+def validate_fixture_data(fixture: Any, fixture_id: str) -> None:
+    if not isinstance(fixture, dict):
+        raise ScenarioError(f"fixture root must be an object: {fixture_id}")
+    if fixture.get("schemaVersion") != 1 or fixture.get("id") != fixture_id:
+        raise ScenarioError(f"invalid fixture metadata: {fixture_id}")
+    if fixture.get("resetPolicy") != "replace":
+        raise ScenarioError(f"unsupported fixture resetPolicy: {fixture_id}")
+    book = fixture.get("book")
+    if not isinstance(book, dict):
+        raise ScenarioError(f"fixture book must be an object: {fixture_id}")
+    required_strings = ("bookUrl", "name", "author")
+    if any(not isinstance(book.get(key), str) or not book[key] for key in required_strings):
+        raise ScenarioError(f"fixture book metadata is incomplete: {fixture_id}")
+    if book.get("pageMode") != "simulation":
+        raise ScenarioError(f"unsupported fixture pageMode: {fixture_id}")
+    chapters = fixture.get("chapters")
+    if not isinstance(chapters, list) or not chapters:
+        raise ScenarioError(f"fixture must contain chapters: {fixture_id}")
+    start_chapter = book.get("startChapter")
+    start_position = book.get("startPosition")
+    if not isinstance(start_chapter, int) or start_chapter not in range(len(chapters)):
+        raise ScenarioError(f"fixture startChapter is invalid: {fixture_id}")
+    if not isinstance(start_position, int) or start_position < 0:
+        raise ScenarioError(f"fixture startPosition is invalid: {fixture_id}")
+    for index, chapter in enumerate(chapters):
+        if not isinstance(chapter, dict):
+            raise ScenarioError(f"fixture chapters[{index}] must be an object")
+        for key in ("title", "readyMarker", "paragraph"):
+            if not isinstance(chapter.get(key), str) or not chapter[key]:
+                raise ScenarioError(f"fixture chapters[{index}].{key} is required")
+        repeat = chapter.get("repeat")
+        if not isinstance(repeat, int) or repeat < 1 or repeat > 1000:
+            raise ScenarioError(f"fixture chapters[{index}].repeat must be between 1 and 1000")
+
+
+def run_command(
+    command: Sequence[str],
+    *,
+    timeout: int = 120,
+    check: bool = True,
+    cwd: Path = PROJECT_ROOT,
+) -> subprocess.CompletedProcess[str]:
+    result = subprocess.run(
+        list(command),
+        cwd=cwd,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        timeout=timeout,
+    )
+    if check and result.returncode != 0:
+        raise CommandError(command, result)
+    return result
+
+
+def adb(device: str, *arguments: str, timeout: int = 60, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return run_command(["adb", "-s", device, *arguments], timeout=timeout, check=check)
+
+
+def select_device(requested: str | None) -> str:
+    if shutil.which("adb") is None:
+        raise RuntimeError("adb is not available on PATH")
+    result = run_command(["adb", "devices"], timeout=15)
+    devices = []
+    for line in result.stdout.splitlines()[1:]:
+        fields = line.split()
+        if len(fields) >= 2 and fields[1] == "device":
+            devices.append(fields[0])
+    if requested:
+        if requested not in devices:
+            raise RuntimeError(f"device is not ready: {requested}")
+        return requested
+    if len(devices) == 1:
+        return devices[0]
+    if not devices:
+        raise RuntimeError("no ready Android device found")
+    raise RuntimeError("multiple devices found; pass --device")
+
+
+def create_session(scenario_name: str, fixture: str, device: str | None) -> tuple[str, Path, dict[str, Any]]:
+    now = dt.datetime.now().astimezone()
+    session_id = now.strftime("%Y%m%d-%H%M%S-%f")[:-3]
+    run_dir = RUN_ROOT / session_id
+    run_dir.mkdir(parents=True, exist_ok=False)
+    metadata = {
+        "schemaVersion": 1,
+        "sessionId": session_id,
+        "scenario": scenario_name,
+        "fixture": fixture,
+        "device": device,
+        "startedAt": now.isoformat(timespec="milliseconds"),
+        "package": PACKAGE,
+    }
+    write_json(run_dir / "metadata.json", metadata)
+    return session_id, run_dir, metadata
+
+
+def write_json(path: Path, value: Any) -> None:
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def find_debug_apk() -> Path:
+    output_root = PROJECT_ROOT / "app" / "build" / "outputs" / "apk"
+    candidates = [
+        path
+        for path in output_root.rglob("*.apk")
+        if "androidTest" not in path.parts and "debug" in path.parts
+    ]
+    if not candidates:
+        raise RuntimeError("debug APK was not produced")
+    universal = [path for path in candidates if "universal" in path.name]
+    return max(universal or candidates, key=lambda path: path.stat().st_mtime)
+
+
+def build_and_install(device: str, run_dir: Path) -> tuple[dict[str, Any], Path]:
+    started = time.monotonic()
+    command = ["./gradlew", ":app:assembleAppDebug", "-PenableAbiSplits=false"]
+    result = run_command(command, timeout=900, check=False)
+    (run_dir / "build.log").write_text(result.stdout, encoding="utf-8")
+    build = {
+        "status": "passed" if result.returncode == 0 else "failed",
+        "durationMs": round((time.monotonic() - started) * 1000),
+        "artifact": "build.log",
+    }
+    if result.returncode != 0:
+        raise CommandError(command, result)
+    apk = find_debug_apk()
+    install = adb(device, "install", "-r", "-t", str(apk), timeout=180, check=False)
+    (run_dir / "install.log").write_text(install.stdout, encoding="utf-8")
+    if install.returncode != 0:
+        raise CommandError(["adb", "install", str(apk)], install)
+    build["apk"] = str(apk.relative_to(PROJECT_ROOT))
+    return build, apk
+
+
+def launch_fixture(device: str, fixture: str, session_id: str) -> None:
+    adb(device, "shell", "am", "force-stop", PACKAGE)
+    adb(device, "logcat", "-c")
+    component = f"{PACKAGE}/{DEBUG_ACTIVITY}"
+    result = adb(
+        device,
+        "shell",
+        "am",
+        "start",
+        "-W",
+        "-n",
+        component,
+        "--es",
+        "fixture",
+        fixture,
+        "--es",
+        "session",
+        session_id,
+        timeout=60,
+        check=False,
+    )
+    if result.returncode != 0 or "Error:" in result.stdout:
+        raise CommandError(["adb", "shell", "am", "start", component], result)
+
+
+def ui_dump(device: str) -> str:
+    return adb(device, "exec-out", "uiautomator", "dump", "/dev/tty", timeout=20, check=False).stdout
+
+
+def current_focus(device: str) -> str:
+    return adb(device, "shell", "dumpsys", "window", "windows", timeout=20, check=False).stdout
+
+
+def wait_until_reader_ready(device: str, marker: str, timeout_seconds: int) -> tuple[bool, str]:
+    deadline = time.monotonic() + timeout_seconds
+    latest = ""
+    while time.monotonic() < deadline:
+        latest = ui_dump(device)
+        if marker in latest and PACKAGE in current_focus(device):
+            return True, latest
+        time.sleep(0.5)
+    return False, latest
+
+
+def screen_size(device: str) -> tuple[int, int]:
+    output = adb(device, "shell", "wm", "size").stdout
+    matches = re.findall(r"(\d+)x(\d+)", output)
+    if not matches:
+        raise RuntimeError(f"cannot determine screen size: {output.strip()}")
+    width, height = matches[-1]
+    return int(width), int(height)
+
+
+def capture_screenshot(device: str, path: Path) -> None:
+    with path.open("wb") as output:
+        result = subprocess.run(
+            ["adb", "-s", device, "exec-out", "screencap", "-p"],
+            cwd=PROJECT_ROOT,
+            stdout=output,
+            stderr=subprocess.PIPE,
+            timeout=30,
+        )
+    if result.returncode != 0:
+        path.unlink(missing_ok=True)
+        raise RuntimeError(result.stderr.decode(errors="replace"))
+
+
+def perform_actions(device: str, scenario: dict[str, Any], action_repeat: int) -> int:
+    width, height = screen_size(device)
+    performed = 0
+    for action in scenario["actions"]:
+        if action["type"] == "open_reader":
+            continue
+        direction = action["direction"]
+        count = action.get("count", 1) * action_repeat
+        x = round(width * (0.84 if direction == "next" else 0.16))
+        y = round(height * 0.5)
+        for _ in range(count):
+            adb(device, "shell", "input", "tap", str(x), str(y), timeout=10)
+            performed += 1
+            time.sleep(0.4)
+    return performed
+
+
+def collect_logcat(device: str, run_dir: Path) -> str:
+    logcat = adb(device, "logcat", "-d", "-v", "threadtime", timeout=45, check=False).stdout
+    (run_dir / "logcat.txt").write_text(logcat, encoding="utf-8")
+    crash = extract_crash(logcat, PACKAGE)
+    (run_dir / "crash.txt").write_text(crash, encoding="utf-8")
+    return logcat
+
+
+def extract_crash(logcat: str, package: str = PACKAGE) -> str:
+    lines = logcat.splitlines()
+    chunks = []
+    for index, line in enumerate(lines):
+        chunk = lines[index : index + 100]
+        if f"ANR in {package}" in line:
+            chunks.extend(chunk)
+        elif "FATAL EXCEPTION:" in line and any(package in item for item in chunk[:10]):
+            chunks.extend(chunk)
+    if not chunks:
+        return ""
+    return "\n".join(chunks).rstrip() + "\n"
+
+
+def parse_failure(crash: str) -> dict[str, Any] | None:
+    if not crash:
+        return None
+    if "ANR in" in crash:
+        return {"category": "anr", "stacktraceArtifact": "crash.txt"}
+    exception = re.search(r"(?:Caused by: )?([\w.$]+(?:Exception|Error))(?:: ([^\n]+))?", crash)
+    failure: dict[str, Any] = {
+        "category": "uncaught_exception",
+        "stacktraceArtifact": "crash.txt",
+    }
+    if exception:
+        failure["exceptionType"] = exception.group(1).split(".")[-1]
+        if exception.group(2):
+            failure["message"] = exception.group(2).strip()
+    frame = re.search(
+        r"at (io\.legado\.app\.[\w.$]+)\.([\w$<>]+)\(([^():]+):(\d+)\)",
+        crash,
+    )
+    if frame:
+        failure["suspectedFrame"] = {
+            "class": frame.group(1),
+            "method": frame.group(2),
+            "file": frame.group(3),
+            "line": int(frame.group(4)),
+            "confidence": "high",
+        }
+    return failure
+
+
+def sha256_text(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()
+
+
+def make_summary(
+    *,
+    session_id: str,
+    scenario: dict[str, Any],
+    fixture: str,
+    build: dict[str, Any],
+    ready: bool,
+    activity_visible: bool,
+    content_changed: bool,
+    action_count: int,
+    failure: dict[str, Any] | None,
+) -> tuple[dict[str, Any], int]:
+    assertion_values = {
+        "reader_ready": ready,
+        "reader_activity_visible": activity_visible,
+        "page_content_changed": content_changed,
+        "no_fatal_exception": failure is None,
+    }
+    assertions = [
+        {"name": name, "passed": assertion_values[name]}
+        for name in scenario["assertions"]
+    ]
+    if failure is not None:
+        status, exit_code = "failed", EXIT_CRASH
+    elif not all(item["passed"] for item in assertions):
+        status, exit_code = "failed", EXIT_ASSERTION
+    else:
+        status, exit_code = "passed", 0
+    summary: dict[str, Any] = {
+        "schemaVersion": 1,
+        "status": status,
+        "exitCode": exit_code,
+        "sessionId": session_id,
+        "scenario": scenario["name"],
+        "fixture": fixture,
+        "build": build,
+        "actions": {"performed": action_count},
+        "assertions": assertions,
+        "artifacts": {
+            "metadata": "metadata.json",
+            "logcat": "logcat.txt",
+            "crash": "crash.txt",
+            "beforeScreenshot": "before.png",
+            "afterScreenshot": "after.png",
+        },
+    }
+    if failure is not None:
+        summary["failure"] = failure
+    return summary, exit_code
+
+
+def error_summary(
+    session_id: str,
+    scenario_name: str,
+    fixture: str,
+    exit_code: int,
+    category: str,
+    message: str,
+    build: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    summary: dict[str, Any] = {
+        "schemaVersion": 1,
+        "status": "failed",
+        "exitCode": exit_code,
+        "sessionId": session_id,
+        "scenario": scenario_name,
+        "fixture": fixture,
+        "failure": {"category": category, "message": message},
+        "artifacts": {"metadata": "metadata.json"},
+    }
+    if build is not None:
+        summary["build"] = build
+    return summary
+
+
+def execute(args: argparse.Namespace) -> int:
+    try:
+        scenario = load_scenario(args.scenario)
+        fixture = args.fixture or scenario["fixture"]
+        fixture_data = validate_fixture(fixture)
+        markers = {chapter["readyMarker"] for chapter in fixture_data["chapters"]}
+        if scenario["readyMarker"] not in markers:
+            raise ScenarioError("scenario readyMarker is not provided by the fixture")
+        if args.action_repeat < 1 or args.action_repeat > 100:
+            raise ScenarioError("--action-repeat must be between 1 and 100")
+    except ScenarioError as error:
+        print(json.dumps({"status": "failed", "exitCode": EXIT_SCENARIO, "error": str(error)}, ensure_ascii=False))
+        return EXIT_SCENARIO
+
+    try:
+        device = select_device(args.device)
+    except (RuntimeError, CommandError, subprocess.TimeoutExpired, OSError) as error:
+        session_id, run_dir, _ = create_session(args.scenario, fixture, args.device)
+        summary = error_summary(session_id, args.scenario, fixture, EXIT_ENVIRONMENT, "environment", str(error))
+        write_json(run_dir / "summary.json", summary)
+        print(run_dir / "summary.json")
+        return EXIT_ENVIRONMENT
+
+    session_id, run_dir, metadata = create_session(args.scenario, fixture, device)
+    build: dict[str, Any] = {"status": "not_started"}
+    try:
+        build, _ = build_and_install(device, run_dir)
+    except (RuntimeError, CommandError, subprocess.TimeoutExpired) as error:
+        if isinstance(error, CommandError):
+            build = {
+                "status": "failed",
+                "artifact": "build.log" if (run_dir / "build.log").exists() else "install.log",
+            }
+        summary = error_summary(session_id, args.scenario, fixture, EXIT_BUILD, "build_or_install", str(error), build)
+        write_json(run_dir / "summary.json", summary)
+        print(run_dir / "summary.json")
+        return EXIT_BUILD
+
+    try:
+        launch_fixture(device, fixture, session_id)
+        ready, before_dump = wait_until_reader_ready(
+            device,
+            scenario["readyMarker"],
+            scenario.get("readyTimeoutSeconds", 30),
+        )
+        capture_screenshot(device, run_dir / "before.png")
+        action_count = perform_actions(device, scenario, args.action_repeat) if ready else 0
+        time.sleep(0.5)
+        after_dump = ui_dump(device)
+        capture_screenshot(device, run_dir / "after.png")
+        focus = current_focus(device)
+        activity_visible = PACKAGE in focus and MAIN_ACTIVITY in focus
+        content_changed = ready and sha256_text(before_dump) != sha256_text(after_dump)
+        logcat = collect_logcat(device, run_dir)
+        failure = parse_failure(extract_crash(logcat))
+        metadata.update(
+            {
+                "finishedAt": dt.datetime.now().astimezone().isoformat(timespec="milliseconds"),
+                "actionRepeat": args.action_repeat,
+                "uiBeforeSha256": sha256_text(before_dump),
+                "uiAfterSha256": sha256_text(after_dump),
+            }
+        )
+        write_json(run_dir / "metadata.json", metadata)
+        summary, exit_code = make_summary(
+            session_id=session_id,
+            scenario=scenario,
+            fixture=fixture,
+            build=build,
+            ready=ready,
+            activity_visible=activity_visible,
+            content_changed=content_changed,
+            action_count=action_count,
+            failure=failure,
+        )
+    except (RuntimeError, CommandError, subprocess.TimeoutExpired, OSError) as error:
+        try:
+            collect_logcat(device, run_dir)
+        except Exception:
+            pass
+        summary = error_summary(
+            session_id,
+            args.scenario,
+            fixture,
+            EXIT_ENVIRONMENT,
+            "environment",
+            str(error),
+            build,
+        )
+        exit_code = EXIT_ENVIRONMENT
+    write_json(run_dir / "summary.json", summary)
+    print(run_dir / "summary.json")
+    return exit_code
+
+
+def parse_args(argv: Sequence[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    subparsers = parser.add_subparsers(dest="command", required=True)
+    run_parser = subparsers.add_parser("run", help="build, install, and run one scenario")
+    run_parser.add_argument("scenario", help="scenario ID, for example reader/page-turn")
+    run_parser.add_argument("--device", help="adb device serial; optional when exactly one is ready")
+    run_parser.add_argument("--fixture", help="override the scenario fixture ID")
+    run_parser.add_argument("--action-repeat", type=int, default=1, help="repeat each domain action (default: 1)")
+    return parser.parse_args(argv)
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = parse_args(argv if argv is not None else sys.argv[1:])
+    if args.command == "run":
+        return execute(args)
+    return EXIT_SCENARIO
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/android/scenarios/reader/page-turn.json

@@ -0,0 +1,17 @@
+{
+  "schemaVersion": 1,
+  "name": "reader/page-turn",
+  "fixture": "long-chapter-v1",
+  "readyMarker": "LEGADO_DEBUG_LONG_CHAPTER_V1",
+  "readyTimeoutSeconds": 30,
+  "actions": [
+    { "type": "open_reader" },
+    { "type": "turn_page", "direction": "next", "count": 1 }
+  ],
+  "assertions": [
+    "reader_ready",
+    "reader_activity_visible",
+    "page_content_changed",
+    "no_fatal_exception"
+  ]
+}

tools/android/tests/test_runner.py

@@ -0,0 +1,99 @@
+import importlib.util
+from pathlib import Path
+import sys
+import unittest
+
+
+RUNNER_PATH = Path(__file__).parents[1] / "runner.py"
+SPEC = importlib.util.spec_from_file_location("legado_android_runner", RUNNER_PATH)
+runner = importlib.util.module_from_spec(SPEC)
+assert SPEC.loader is not None
+SPEC.loader.exec_module(runner)
+
+
+class ScenarioValidationTest(unittest.TestCase):
+    def test_repository_scenario_is_valid(self):
+        scenario = runner.load_scenario("reader/page-turn")
+        self.assertEqual("long-chapter-v1", scenario["fixture"])
+
+    def test_rejects_unknown_action(self):
+        scenario = {
+            "schemaVersion": 1,
+            "name": "reader/page-turn",
+            "fixture": "long-chapter-v1",
+            "readyMarker": "marker",
+            "actions": [{"type": "tap"}],
+            "assertions": ["reader_ready"],
+        }
+        with self.assertRaises(runner.ScenarioError):
+            runner.validate_scenario(scenario)
+
+    def test_rejects_missing_fixture(self):
+        with self.assertRaises(runner.ScenarioError):
+            runner.validate_fixture("missing-v1")
+
+    def test_rejects_invalid_fixture_before_device_work(self):
+        fixture = {
+            "schemaVersion": 1,
+            "id": "broken-v1",
+            "resetPolicy": "replace",
+            "book": {
+                "bookUrl": "debug://book",
+                "name": "broken",
+                "author": "debug",
+                "pageMode": "simulation",
+                "startChapter": 0,
+                "startPosition": 0,
+            },
+            "chapters": [{"title": "chapter", "readyMarker": "ready", "paragraph": "text", "repeat": 0}],
+        }
+        with self.assertRaises(runner.ScenarioError):
+            runner.validate_fixture_data(fixture, "broken-v1")
+
+
+class FailureParsingTest(unittest.TestCase):
+    def test_ignores_crash_from_another_package(self):
+        logcat = """FATAL EXCEPTION: main
+Process: com.example.other, PID: 42
+java.lang.IllegalStateException: unrelated
+"""
+        self.assertEqual("", runner.extract_crash(logcat))
+
+    def test_parses_app_frame_as_inference(self):
+        crash = """FATAL EXCEPTION: main
+java.lang.IndexOutOfBoundsException: Index 18 out of bounds
+    at io.legado.app.ui.book.read.ReadBookViewModel.loadChapter(ReadBookViewModel.kt:418)
+"""
+        failure = runner.parse_failure(crash)
+        self.assertEqual("uncaught_exception", failure["category"])
+        self.assertEqual("IndexOutOfBoundsException", failure["exceptionType"])
+        self.assertEqual("high", failure["suspectedFrame"]["confidence"])
+        self.assertEqual(418, failure["suspectedFrame"]["line"])
+
+    def test_crash_has_priority_over_assertion_failure(self):
+        scenario = runner.load_scenario("reader/page-turn")
+        summary, exit_code = runner.make_summary(
+            session_id="test",
+            scenario=scenario,
+            fixture="long-chapter-v1",
+            build={"status": "passed"},
+            ready=False,
+            activity_visible=False,
+            content_changed=False,
+            action_count=0,
+            failure={"category": "uncaught_exception", "stacktraceArtifact": "crash.txt"},
+        )
+        self.assertEqual(runner.EXIT_CRASH, exit_code)
+        self.assertEqual(runner.EXIT_CRASH, summary["exitCode"])
+
+
+class CommandDecodingTest(unittest.TestCase):
+    def test_replaces_invalid_utf8_in_command_output(self):
+        result = runner.run_command(
+            [sys.executable, "-c", "import sys; sys.stdout.buffer.write(b'log\\xc0tail')"],
+        )
+        self.assertEqual("log\ufffdtail", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()