diff --git a/.env.example b/.env.example
index 1f53db8..646b275 100644
--- a/.env.example
+++ b/.env.example
@@ -27,6 +27,11 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem
 JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify
 JG_REQUEST_TIMEOUT_SEC=15
 
+# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)=====
+HEARTBEAT_MONITOR_ENABLED=true
+HEARTBEAT_TIMEOUT_MINUTES=10
+HEARTBEAT_SCAN_INTERVAL_SEC=60
+
 # ===== 短信 (mock 模式) =====
 # mock = true 时,任意 6 位数字均通过,且 /sms/send 不真发短信(只 log)。
 # 后续接阿里云/腾讯云短信时,改成 false 并填供应商相关 key。
diff --git a/alembic/versions/device_kill_alert_pending.py b/alembic/versions/device_kill_alert_pending.py
new file mode 100644
index 0000000..b9eb475
--- /dev/null
+++ b/alembic/versions/device_kill_alert_pending.py
@@ -0,0 +1,39 @@
+"""device.kill_alert_pending (无障碍掉线「后置检测」待提醒标记)
+
+Revision ID: device_kill_alert_pending
+Revises: device_liveness_table
+Create Date: 2026-06-19
+
+后置检测 pull 版(spec/accessibility-liveness-pull-prompt.md):worker 检出心跳掉线即置此标记,
+客户端下次进 App 拉取到 → 弹「开启自启动」引导,交互后 ack 清回。与 liveness_state 解耦,故单列。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'device_kill_alert_pending'
+down_revision: Union[str, Sequence[str], None] = 'device_liveness_table'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # server_default=sa.false() 跨方言: SQLite 渲染 0 / PG 渲染 false(别用 sa.text("0"),PG 布尔严格会报错),
+    # 给存量行回填 False。
+    with op.batch_alter_table('device', schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                'kill_alert_pending',
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.false(),
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('device', schema=None) as batch_op:
+        batch_op.drop_column('kill_alert_pending')
diff --git a/alembic/versions/device_table.py b/alembic/versions/device_table.py
new file mode 100644
index 0000000..194a9a0
--- /dev/null
+++ b/alembic/versions/device_table.py
@@ -0,0 +1,53 @@
+"""device table (无障碍保护存活检测 + 极光推送)
+
+Revision ID: device_liveness_table
+Revises: ceb286289426
+Create Date: 2026-06-15 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'device_liveness_table'
+down_revision: Union[str, Sequence[str], None] = 'ceb286289426'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'device',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=False),
+        sa.Column('device_id', sa.String(length=128), nullable=False),
+        sa.Column('registration_id', sa.String(length=64), nullable=True),
+        sa.Column('platform', sa.String(length=16), nullable=False),
+        sa.Column('app_version', sa.String(length=32), nullable=True),
+        sa.Column('ever_protected', sa.Boolean(), nullable=False),
+        sa.Column('last_heartbeat_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('last_report_protection_on', sa.Boolean(), nullable=False),
+        sa.Column('liveness_state', sa.String(length=16), nullable=False),
+        sa.Column('notified_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('user_id', 'device_id', name='uq_device_user_device'),
+    )
+    with op.batch_alter_table('device', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_device_user_id'), ['user_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_device_device_id'), ['device_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_device_last_heartbeat_at'), ['last_heartbeat_at'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('device', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_device_last_heartbeat_at'))
+        batch_op.drop_index(batch_op.f('ix_device_device_id'))
+        batch_op.drop_index(batch_op.f('ix_device_user_id'))
+
+    op.drop_table('device')
diff --git a/app/api/v1/device.py b/app/api/v1/device.py
new file mode 100644
index 0000000..56a5a85
--- /dev/null
+++ b/app/api/v1/device.py
@@ -0,0 +1,96 @@
+"""设备注册 / 心跳 endpoint(无障碍保护存活检测)。
+
+路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。
+  POST /register   注册设备 / 更新 registration_id(App 前台、拿到 push token 时调)
+  POST /heartbeat  上报心跳(无障碍服务存活时周期调,刷新存活)
+
+后端 heartbeat_monitor_worker 据此发现心跳超时的设备并极光推送告警。
+见 spec: spec/accessibility-liveness-push.md。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter
+
+from app.api.deps import CurrentUser, DbSession
+from app.repositories import device as device_repo
+from app.schemas.device import (
+    DeviceOut,
+    DeviceRegisterRequest,
+    HeartbeatRequest,
+    LivenessAckRequest,
+    LivenessOut,
+    OkResponse,
+)
+
+logger = logging.getLogger("shagua.device")
+
+router = APIRouter(prefix="/api/v1/device", tags=["device"])
+
+
+@router.post("/register", response_model=DeviceOut, summary="注册设备/更新推送token")
+def register_device(
+    req: DeviceRegisterRequest,
+    user: CurrentUser,
+    db: DbSession,
+) -> DeviceOut:
+    device = device_repo.register_or_update(
+        db,
+        user_id=user.id,
+        device_id=req.device_id,
+        registration_id=req.registration_id,
+        platform=req.platform,
+        app_version=req.app_version,
+    )
+    logger.info(
+        "device register user_id=%d device_id=%s reg=%s",
+        user.id,
+        req.device_id,
+        bool(req.registration_id),
+    )
+    return DeviceOut.model_validate(device)
+
+
+@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳")
+def report_heartbeat(
+    req: HeartbeatRequest,
+    user: CurrentUser,
+    db: DbSession,
+) -> OkResponse:
+    device_repo.touch_heartbeat(
+        db,
+        user_id=user.id,
+        device_id=req.device_id,
+        accessibility_enabled=req.accessibility_enabled,
+        registration_id=req.registration_id,
+    )
+    return OkResponse()
+
+
+@router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)")
+def get_liveness(
+    device_id: str,
+    user: CurrentUser,
+    db: DbSession,
+) -> LivenessOut:
+    """客户端进 App 时拉取: 本机是否被判定掉线过(kill_alert_pending)。
+
+    设备从未注册过 → 返回默认(无告警)。命中 → 客户端弹「开启自启动」引导, 之后调 /liveness/ack 清。
+    见 spec: spec/accessibility-liveness-pull-prompt.md。
+    """
+    device = device_repo.get_device(db, user_id=user.id, device_id=device_id)
+    if device is None:
+        return LivenessOut()
+    return LivenessOut.model_validate(device)
+
+
+@router.post("/liveness/ack", response_model=OkResponse, summary="确认已提醒(清掉线告警)")
+def ack_liveness(
+    req: LivenessAckRequest,
+    user: CurrentUser,
+    db: DbSession,
+) -> OkResponse:
+    """客户端已弹过「开启自启动」引导 → 清掉「待提醒」标记(下次真掉线 worker 再置)。"""
+    device_repo.ack_kill_alert(db, user_id=user.id, device_id=req.device_id)
+    return OkResponse()
diff --git a/app/core/config.py b/app/core/config.py
index 0520923..8b3fe68 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -66,6 +66,11 @@ class Settings(BaseSettings):
     JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify"
     JG_REQUEST_TIMEOUT_SEC: int = 15
 
+    # 无障碍保护存活监控后台任务(pull 后置检测;本期不接推送)
+    HEARTBEAT_MONITOR_ENABLED: bool = True       # 总开关
+    HEARTBEAT_TIMEOUT_MINUTES: int = 10          # 多久没心跳算掉线(≈3 个客户端心跳周期)
+    HEARTBEAT_SCAN_INTERVAL_SEC: int = 60        # 扫描周期
+
     # ===== 短信 =====
     SMS_MOCK: bool = True
     SMS_CODE_TTL_SEC: int = 300
diff --git a/app/core/heartbeat_monitor_worker.py b/app/core/heartbeat_monitor_worker.py
new file mode 100644
index 0000000..6072d26
--- /dev/null
+++ b/app/core/heartbeat_monitor_worker.py
@@ -0,0 +1,148 @@
+"""无障碍保护存活监控后台任务。
+
+周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了),
+**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机
+推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。
+结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。
+
+见 spec: spec/accessibility-liveness-push.md。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import time
+from collections.abc import Iterator
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.repositories import device as device_repo
+
+logger = logging.getLogger("shagua.heartbeat_monitor")
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "heartbeat_monitor.lock"
+
+
+def _touch_lock() -> None:
+    with contextlib.suppress(FileNotFoundError):
+        os.utime(_LOCK_PATH, None)
+
+
+@contextlib.contextmanager
+def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
+    """同机多进程保护:同一时间只允许一个监控 worker 运行。"""
+    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd: int | None = None
+    try:
+        try:
+            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+        except FileExistsError:
+            try:
+                age = time.time() - _LOCK_PATH.stat().st_mtime
+            except FileNotFoundError:
+                age = stale_after_sec + 1
+            if age > stale_after_sec:
+                with contextlib.suppress(FileNotFoundError):
+                    _LOCK_PATH.unlink()
+                try:
+                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+                except FileExistsError:
+                    fd = None
+
+        if fd is None:
+            yield False
+            return
+
+        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
+        yield True
+    finally:
+        if fd is not None:
+            os.close(fd)
+            with contextlib.suppress(FileNotFoundError):
+                _LOCK_PATH.unlink()
+
+
+def _silent_seconds(last: datetime | None) -> int | None:
+    """距上次心跳的秒数(兼容 sqlite 取回的 naive datetime)。"""
+    if last is None:
+        return None
+    ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow()
+    return int((ref - last).total_seconds())
+
+
+def _scan_once(timeout_minutes: int) -> dict:
+    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。
+
+    本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端,
+    并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。
+    """
+    notified = 0
+    with SessionLocal() as db:
+        overdue = device_repo.list_overdue(db, timeout_minutes=timeout_minutes)
+        for device in overdue:
+            silent = _silent_seconds(device.last_heartbeat_at)
+            logger.warning(
+                "[掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)"
+                " → 判定 App 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 App 将弹「开启自启动」引导(后置检测);推送本期未接】",
+                device.user_id,
+                device.device_id,
+                silent if silent is not None else "?",
+                timeout_minutes,
+            )
+            device_repo.mark_notified(db, device_id_pk=device.id)
+            notified += 1
+        return {"checked": len(overdue), "notified": notified}
+
+
+async def _run_loop() -> None:
+    interval = max(10, int(settings.HEARTBEAT_SCAN_INTERVAL_SEC))
+    timeout_minutes = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES))
+    lock_stale_after = max(interval * 3, 600)
+    with _single_instance_lock(lock_stale_after) as lock_acquired:
+        if not lock_acquired:
+            logger.warning("heartbeat monitor skipped: another worker owns lock")
+            return
+        await _run_locked_loop(interval, timeout_minutes)
+
+
+async def _run_locked_loop(interval: int, timeout_minutes: int) -> None:
+    logger.info(
+        "heartbeat monitor started interval=%ss timeout=%sm",
+        interval,
+        timeout_minutes,
+    )
+    try:
+        while True:
+            try:
+                _touch_lock()
+                result = await asyncio.to_thread(_scan_once, timeout_minutes)
+                if result["notified"]:
+                    logger.info("heartbeat monitor result=%s", result)
+            except SQLAlchemyError:
+                logger.exception("heartbeat monitor db error")
+            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
+                logger.exception("heartbeat monitor unexpected error")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("heartbeat monitor stopped")
+        raise
+
+
+def start_heartbeat_monitor() -> asyncio.Task | None:
+    if not settings.HEARTBEAT_MONITOR_ENABLED:
+        logger.info("heartbeat monitor disabled")
+        return None
+    return asyncio.create_task(_run_loop(), name="heartbeat-monitor")
+
+
+async def stop_heartbeat_monitor(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
diff --git a/app/main.py b/app/main.py
index af011f8..5a4a6bd 100644
--- a/app/main.py
+++ b/app/main.py
@@ -20,6 +20,7 @@ from app.api.v1.compare import router as compare_router
 from app.api.v1.compare_milestone import router as compare_milestone_router
 from app.api.v1.compare_record import router as compare_record_router
 from app.api.v1.coupon import router as coupon_router
+from app.api.v1.device import router as device_router
 from app.api.v1.cps_redirect import router as cps_redirect_router
 from app.api.internal.app_version import router as internal_app_version_router
 from app.api.internal.launch_confirm import router as internal_launch_confirm_router
@@ -38,6 +39,10 @@ from app.api.v1.user import router as user_router
 from app.api.v1.wallet import router as wallet_router
 from app.api.v1.wxpay import router as wxpay_router
 from app.core.config import settings
+from app.core.heartbeat_monitor_worker import (
+    start_heartbeat_monitor,
+    stop_heartbeat_monitor,
+)
 from app.core.daily_exchange_worker import (
     start_daily_exchange_worker,
     stop_daily_exchange_worker,
@@ -63,10 +68,12 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.DATABASE_URL.split("://", 1)[0],
     )
     reconcile_task = start_withdraw_reconcile_worker()
+    heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
     try:
         yield
     finally:
+        await stop_heartbeat_monitor(heartbeat_task)
         await stop_withdraw_reconcile_worker(reconcile_task)
         await stop_daily_exchange_worker(daily_exchange_task)
         logger.info("shutting down")
@@ -100,6 +107,7 @@ app.include_router(user_router)
 app.include_router(feedback_router)
 app.include_router(invite_router)
 app.include_router(coupon_router)
+app.include_router(device_router)
 app.include_router(compare_router)
 app.include_router(compare_record_router)
 app.include_router(compare_milestone_router)
diff --git a/app/models/__init__.py b/app/models/__init__.py
index d88882f..f3491f1 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -12,6 +12,7 @@ from app.models.cps_link import CpsClick, CpsLink  # noqa: F401
 from app.models.cps_order import CpsOrder  # noqa: F401
 from app.models.cps_wx_user import CpsWxUser  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
+from app.models.device import Device  # noqa: F401
 from app.models.coupon_state import (  # noqa: F401
     CouponClaimRecord,
     CouponDailyCompletion,
diff --git a/app/models/device.py b/app/models/device.py
new file mode 100644
index 0000000..290bc9f
--- /dev/null
+++ b/app/models/device.py
@@ -0,0 +1,89 @@
+"""设备表(无障碍保护存活检测 + 极光推送)。
+
+每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。
+客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报
+registration_id(极光推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、
+现在心跳超时」的设备,通过极光推送提醒用户重开无障碍。
+
+liveness_state 状态机(防刷屏,一次掉线只推一条):
+  unknown → alive(收到 service 心跳)→ silent/notified(扫描发现超时并已推送)
+  心跳恢复时 handler 重置回 alive。
+见 spec: spec/accessibility-liveness-push.md。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class Device(Base):
+    __tablename__ = "device"
+    __table_args__ = (
+        UniqueConstraint("user_id", "device_id", name="uq_device_user_device"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("user.id"), index=True, nullable=False
+    )
+    # 客户端 DeviceId.get() 生成的 per-install id(如 device_Pixel_ab12cd34)
+    device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False)
+    # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发)
+    registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android")
+    app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
+
+    # 收到过 service 心跳即 true(=该设备开过无障碍,功能对它有意义)
+    ever_protected: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+    # 最近一次 service 心跳时间(存活证明);超时即视为保护掉线
+    last_heartbeat_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), index=True, nullable=True
+    )
+    # 最近一次上报的无障碍开关状态(观测用)
+    last_report_protection_on: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+    # unknown / alive / silent / notified
+    liveness_state: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="unknown"
+    )
+    # 最近一次推送告警时间
+    notified_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    # 掉线告警「待客户端提醒」标记(后置检测 pull 版, 见 spec accessibility-liveness-pull-prompt.md)。
+    # 与 liveness_state 解耦: worker 检出掉线即置 True; touch_heartbeat(心跳恢复)不动它
+    # → 规避「服务随 App 重启先发心跳、state 被重置回 alive → 客户端进 App 漏看」竞态; 只由客户端 ack 清。
+    kill_alert_pending: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<Device id={self.id} user_id={self.user_id} "
+            f"device_id={self.device_id} state={self.liveness_state}>"
+        )
diff --git a/app/repositories/device.py b/app/repositories/device.py
new file mode 100644
index 0000000..b1c40d8
--- /dev/null
+++ b/app/repositories/device.py
@@ -0,0 +1,124 @@
+"""device 表读写(设备注册 / 心跳 / 超时扫描)。"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.device import Device
+
+
+def _get(db: Session, *, user_id: int, device_id: str) -> Device | None:
+    stmt = select(Device).where(
+        Device.user_id == user_id, Device.device_id == device_id
+    )
+    return db.execute(stmt).scalar_one_or_none()
+
+
+def register_or_update(
+    db: Session,
+    *,
+    user_id: int,
+    device_id: str,
+    registration_id: str | None,
+    platform: str = "android",
+    app_version: str | None = None,
+) -> Device:
+    """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。"""
+    device = _get(db, user_id=user_id, device_id=device_id)
+    if device is None:
+        device = Device(
+            user_id=user_id,
+            device_id=device_id,
+            registration_id=registration_id,
+            platform=platform or "android",
+            app_version=app_version,
+        )
+        db.add(device)
+    else:
+        if registration_id:
+            device.registration_id = registration_id
+        if platform:
+            device.platform = platform
+        if app_version:
+            device.app_version = app_version
+    db.commit()
+    db.refresh(device)
+    return device
+
+
+def touch_heartbeat(
+    db: Session,
+    *,
+    user_id: int,
+    device_id: str,
+    accessibility_enabled: bool,
+    registration_id: str | None,
+) -> Device:
+    """处理一次心跳(心跳也能自注册)。
+
+    service 心跳或 accessibility_enabled=true 时,刷新存活并把状态机重置回 alive、
+    清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。
+    """
+    now = datetime.now(timezone.utc)
+    device = _get(db, user_id=user_id, device_id=device_id)
+    if device is None:
+        device = Device(user_id=user_id, device_id=device_id)
+        db.add(device)
+
+    if registration_id:
+        device.registration_id = registration_id
+    device.last_report_protection_on = accessibility_enabled
+
+    if accessibility_enabled:
+        device.last_heartbeat_at = now
+        device.ever_protected = True
+        device.liveness_state = "alive"
+        device.notified_at = None
+
+    db.commit()
+    db.refresh(device)
+    return device
+
+
+def list_overdue(db: Session, *, timeout_minutes: int) -> list[Device]:
+    """掉线设备:曾经保护过、当前 alive、心跳超时。
+
+    本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。
+    """
+    cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
+    stmt = select(Device).where(
+        Device.ever_protected.is_(True),
+        Device.liveness_state == "alive",
+        Device.last_heartbeat_at.is_not(None),
+        Device.last_heartbeat_at < cutoff,
+    )
+    return list(db.execute(stmt).scalars().all())
+
+
+def mark_notified(db: Session, *, device_id_pk: int) -> None:
+    """标记掉线已检出(状态机进入 notified,避免每轮重复处理)+ 置「待客户端提醒」标记。
+
+    kill_alert_pending 与 state 解耦(见 spec accessibility-liveness-pull-prompt.md):
+    心跳恢复(touch_heartbeat)只重置 state、不动此标记,故客户端下次进 App 必能拉到这次掉线。
+    """
+    device = db.get(Device, device_id_pk)
+    if device is not None:
+        device.liveness_state = "notified"
+        device.notified_at = datetime.now(timezone.utc)
+        device.kill_alert_pending = True
+        db.commit()
+
+
+def get_device(db: Session, *, user_id: int, device_id: str) -> Device | None:
+    """按 (user_id, device_id) 取设备(liveness 查询用)。"""
+    return _get(db, user_id=user_id, device_id=device_id)
+
+
+def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None:
+    """客户端已弹过「开启自启动」引导 → 清「待提醒」标记(幂等; 下次真掉线 worker 再置)。"""
+    device = _get(db, user_id=user_id, device_id=device_id)
+    if device is not None and device.kill_alert_pending:
+        device.kill_alert_pending = False
+        db.commit()
diff --git a/app/schemas/device.py b/app/schemas/device.py
new file mode 100644
index 0000000..3254e06
--- /dev/null
+++ b/app/schemas/device.py
@@ -0,0 +1,50 @@
+"""设备注册 / 心跳相关 schema(无障碍保护存活检测)。"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class DeviceRegisterRequest(BaseModel):
+    device_id: str
+    registration_id: str | None = None
+    platform: str = "android"
+    app_version: str | None = None
+
+
+class HeartbeatRequest(BaseModel):
+    device_id: str
+    source: str = "service"          # service | app
+    accessibility_enabled: bool = True
+    registration_id: str | None = None
+
+
+class DeviceOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    device_id: str
+    registration_id: str | None
+    ever_protected: bool
+    liveness_state: str
+    last_heartbeat_at: datetime | None
+    updated_at: datetime | None
+
+
+class OkResponse(BaseModel):
+    ok: bool = True
+
+
+class LivenessOut(BaseModel):
+    """本机掉线告警状态(后置检测 pull 版)。设备从未注册过 → 全默认(= 无告警)。"""
+    model_config = ConfigDict(from_attributes=True)
+
+    kill_alert_pending: bool = False
+    liveness_state: str = "unknown"
+    ever_protected: bool = False
+    last_heartbeat_at: datetime | None = None
+
+
+class LivenessAckRequest(BaseModel):
+    device_id: str