diff --git a/docs/stealing.md b/docs/stealing.md
new file mode 100644
index 0000000..07264ec
--- /dev/null
+++ b/docs/stealing.md
@@ -0,0 +1,41 @@
+# 偷菜与仓库输出规则
+
+## 偷菜
+
+仅支持发送 `偷菜 @用户`，不接受直接输入 QQ 号或在 @ 后附加文字。
+仅支持当前机器人、平台共享范围内的
+已开通农场，不允许偷自己。结果直接回复文字，避免额外绘图和图片发送。
+
+每次遍历目标全部地块；成熟、未枯萎且本季未被当前玩家偷过的地块，各随机
+偷取 1～3 个，直接进入偷取者的作物仓库。同名同编号的结果合并展示。
+未成熟、本季已偷过、防偷保护分别汇总跳过数量。
+
+本季总产量为土地加成后再按成熟时虫数减产的整数产量。剩余产量减去已偷
+数量；防偷下限为本季总产量的一半向上取整。临界时缩减本次偷取数量，
+保证不会穿过下限。例如本季产量 9 个，至少保留 5 个。
+
+保留每日最多 5 次成功偷菜规则，以北京时间零点刷新；一次访问无论偷取
+多少地块只算一次成功，完全未偷到不扣次数。操作记录持久化，重启不重置。
+同一消息重放返回原收据，不会重复扣产或入库。
+
+目标扣产、偷取者入库、去重记录、操作收据均在一个 SQLite 事务中提交。
+收获按剩余产量入库，并清除本季偷取记录；再生长进入下一季后可重新偷取。
+农场详述显示扣除偷取后的预计产出，不额外展示被偷数量。
+
+已有核心数据库版本 1～3 在启动时顺序迁移至版本 4。新增字段仅保存
+本季已被偷数量；独立记录表用于判定每个玩家是否偷过当前季。
+
+## 仓库图片质量
+
+`我的种子`、`我的作物` 支持以下输出，页码可省略，默认第 1 页：
+
+| 指令参数 | 尺寸 | JPEG 质量 | 用途 |
+| --- | --- | --- | --- |
+| `[页码]` | 768 × 1152 | 70 | 默认标准压缩图 |
+| `[页码] 高清` | 1024 × 1536 | 85 | 放大查看文字 |
+| `[页码] 文字` | 无图片 | 无 | 老机器、网络慢时使用 |
+
+每页最多 12 项；标准档按实际输出尺寸绘制。相同快照、分类、页码、质量
+可复用有界内存缓存，并共享进行中的绘图任务；绘图并发数为 1，队列有上限。
+默认 AUTO 输出 JPEG；部署者显式配置其他图片格式时仍尊重该设置。
+图片日志记录尺寸、质量、字节数；实际耗时随机器、字体、缓存而变化。
diff --git a/integrations/zhenxun/commands.py b/integrations/zhenxun/commands.py
index 1fa9277..7628f98 100644
--- a/integrations/zhenxun/commands.py
+++ b/integrations/zhenxun/commands.py
@@ -76,12 +76,32 @@ def handler_for(action):
         if message_id is None:
             message_id = getattr(event, "id", None)
         try:
+            target_user_id = None
+            if action == 'steal':
+                mentions = [str(seg.data.get('qq', seg.data.get(
+                    'user_id', seg.data.get('id', ''),
+                ))) for seg in args if seg.type in (
+                    'at', 'mention', 'mention_user', 'mention_everyone',
+                )]
+                if len(mentions) > 1:
+                    await MessageUtils.build_message(
+                        '请只指定一个偷菜目标。',
+                    ).send(reply_to=True)
+                    return
+                if mentions:
+                    target_user_id = mentions[0]
+                    if target_user_id in ('', 'all'):
+                        await MessageUtils.build_message(
+                            '请@一个具体用户。',
+                        ).send(reply_to=True)
+                        return
             avatar = (await avatar_for(user_id, platform)
                       if action in ("farm", "details", "plot_ids") else None)
             response = await runtime.execute(
                 action, args.extract_plain_text(), bot_id=str(bot.self_id),
                 platform=platform, user_id=user_id, message_id=message_id,
                 display_name=session.user.name or "农场玩家", avatar=avatar,
+                target_user_id=target_user_id,
             )
         except Exception as error:
             logger.error(
diff --git a/integrations/zhenxun/runtime.py b/integrations/zhenxun/runtime.py
index 5792942..0630f8f 100644
--- a/integrations/zhenxun/runtime.py
+++ b/integrations/zhenxun/runtime.py
@@ -30,10 +30,13 @@ COMMANDS = {
     "播种":"sow", "收获":"harvest", "铲除":"eradicate",
     "作物加锁":"lock", "作物解锁":"unlock", "出售作物":"sell",
     "农场除虫":"clear_pests", "除虫":"clear_pests", "地块编号":"plot_ids",
+    "偷菜":"steal",
 }
 HELP = (
     "开通农场 [名称]｜我的农场 [高清]｜农场详述 [页码]\n"
     "地块编号 [高清]｜查看当前土地编号及开垦位置\n"
+    "偷菜 @目标｜每日最多5次成功偷取\n"
+    "我的种子/我的作物 [页码] [高清/文字]，默认标准压缩图。\n"
     "种子商店 [页码]｜购买种子 作物 数量｜播种 作物 [数量]\n"
     "收获｜铲除｜农场除虫｜我的种子 [页码]｜我的作物 [页码]\n"
     "作物加锁 作物｜作物解锁 作物｜出售作物 [作物] [数量]\n"
@@ -46,6 +49,10 @@ HELP = (
     "旧版农场数据在首次使用时迁移，绑定当前机器人和平台。"
 )
 ERRORS = {
+    'cannot_steal_self': '不能偷自己的农场。',
+    'different_farm_scope': '只能偷取当前机器人、平台内的农场。',
+    'target_not_registered': '目标尚未开通当前农场。',
+    'steal_daily_limit': '今日成功偷菜已达5次，北京时间零点恢复。',
     "legacy_identity_bound":"旧农场已绑定其他机器人或平台，请联系管理员。",
     "legacy_account_conflict":"新旧农场账号冲突，未覆盖，请联系管理员。",
     "legacy_schema_unsupported":"旧农场数据库结构不兼容，请联系管理员。",
@@ -151,7 +158,7 @@ class Runtime:
 
     async def execute(
         self, action, arguments, *, bot_id, platform, user_id, message_id,
-        display_name="农场玩家", avatar=None,
+        display_name="农场玩家", avatar=None, target_user_id=None,
     ):
         if not self.started:
             return "农场尚未初始化，请检查启动日志。"
@@ -237,6 +244,7 @@ class Runtime:
                 )
             if action in ('crops', 'seeds'):
                 title = '我的作物' if action == 'crops' else '我的种子'
+                mode = args.pop() if args and args[-1] in ('高清', '文字') else ''
                 if len(args) > 1:
                     raise ValueError()
                 try:
@@ -249,10 +257,11 @@ class Runtime:
                     _, pages = warehouse_page(rows, page)
                 except ValueError:
                     return f'页码无效，请发送：{title} 1'
-                if not self.rendering:
+                if not self.rendering or mode == '文字':
                     return warehouse_text(rows, snapshot, action, page, pages)
                 return await self.rendering.warehouse(
                     rows, snapshot, action, page, pages, self.image_format,
+                    mode == '高清',
                 )
             if action == "inventory":
                 if args:
@@ -283,7 +292,15 @@ class Runtime:
                 return picture.content
             if op is None:
                 return "无法取得稳定消息标识，本次写操作未执行。"
-            if action == "register":
+            if action == 'steal':
+                if target_user_id is None:
+                    return '请发送：偷菜 @目标，不支持直接输入QQ号。'
+                if args:
+                    return '请只指定一个偷菜目标，不要附加QQ号或其他文字。'
+                target = Identity(who.deployment, who.realm, who.platform,
+                                  str(target_user_id))
+                receipt = await self.service.steal(who, target, op)
+            elif action == "register":
                 name = " ".join(args) if args else display_name
                 name = " ".join(str(name).split())[:64] or "农场玩家"
                 receipt = await self.service.register(who, name, op)
diff --git a/src/farm_core/application.py b/src/farm_core/application.py
index 4fa5f8d..7bdb8d9 100644
--- a/src/farm_core/application.py
+++ b/src/farm_core/application.py
@@ -2,6 +2,7 @@
 
 from dataclasses import asdict, replace
 import time
+import secrets
 
 from .catalogue import Catalogue
 from .domain import (
@@ -9,6 +10,7 @@ from .domain import (
 )
 from .ports import Store
 from .pests import advance_plot, start_growth
+from .yield_rules import season_yield, remaining_yield
 
 
 class FarmService:
@@ -151,8 +153,7 @@ class FarmService:
                     or plot.mature_at > now
                 ):
                     continue
-                amount = boosted(plant.harvest, plot.yield_bonus)
-                amount = amount * (100 - 5 * plot.maturity_pests) // 100
+                amount = remaining_yield(plot)
                 experience = boosted(
                     plant.experience, plot.experience_bonus,
                 )
@@ -168,7 +169,9 @@ class FarmService:
                     pest_count=0, pest_spawned=0, growth_remaining=None,
                     growth_updated_at=None, next_pest_at=None,
                     maturity_pests=0,
+                    stolen_count=0,
                 )
+                repository.clear_thefts(plot.slot)
                 repository.save_plot(
                     start_growth(updated, now) if count < plant.crop
                     else updated
@@ -200,11 +203,13 @@ class FarmService:
             if not withered:
                 raise FarmError("nothing_to_eradicate")
             for plot in withered:
+                repository.clear_thefts(plot.slot)
                 repository.save_plot(replace(
                     plot, plant=None, mature_at=None, harvest_count=0,
                     pest_count=0, pest_spawned=0, growth_remaining=None,
                     growth_updated_at=None, next_pest_at=None,
                     maturity_pests=0,
+                    stolen_count=0,
                 ))
             experience = integer(len(withered) * 2)
             repository.asset(Asset.EXPERIENCE, experience)
@@ -217,6 +222,64 @@ class FarmService:
             identity, operation_id, "eradicate", {}, operation,
         )
 
+    async def steal(self, identity: Identity, target: Identity,
+                    operation_id: str):
+        self._identity(identity)
+        self._identity(target)
+        if identity == target:
+            raise FarmError('cannot_steal_self')
+        if (identity.deployment, identity.realm, identity.platform) != (
+            target.deployment, target.realm, target.platform
+        ):
+            raise FarmError('different_farm_scope')
+
+        def operation(repository, now):
+            repository.account()
+            owner = repository.for_identity(target)
+            try:
+                owner.account()
+            except FarmError as error:
+                if error.code == 'not_registered':
+                    raise FarmError('target_not_registered') from error
+                raise
+            day_start = ((now+28800)//86400)*86400-28800
+            used = repository.successful_thefts(day_start)
+            if used >= 5:
+                raise FarmError('steal_daily_limit')
+            items = []
+            skipped = {'not_ripe': 0, 'already_stolen': 0, 'protected': 0}
+            for stored in owner.plots():
+                plot = advance_plot(stored, now)
+                if (plot.plant is None or plot.withered
+                        or plot.mature_at is None or plot.mature_at > now):
+                    skipped['not_ripe'] += 1
+                    continue
+                if owner.theft_seen(plot.slot, identity):
+                    skipped['already_stolen'] += 1
+                    continue
+                total = season_yield(plot)
+                available = remaining_yield(plot)-(total+1)//2
+                if available <= 0:
+                    skipped['protected'] += 1
+                    continue
+                amount = min(secrets.randbelow(3)+1, available)
+                repository.stock(plot.plant, crops=amount)
+                owner.save_plot(replace(
+                    plot, stolen_count=plot.stolen_count+amount,
+                ))
+                owner.record_theft(plot.slot, identity)
+                items.append({'slot': plot.slot, 'name': plot.plant.name,
+                              'plant_id': plot.plant.plant_id,
+                              'quantity': amount})
+            return self._result(
+                repository, items=items, skipped=skipped,
+                quantity=sum(i['quantity'] for i in items),
+                remaining_attempts=5-used-bool(items),
+            )
+
+        return await self._execute(identity, operation_id, 'steal',
+                                   {'target': target.key}, operation)
+
     async def clear_pests(self, identity: Identity, operation_id: str):
         def operation(repository, now):
             repository.account()
diff --git a/src/farm_core/domain.py b/src/farm_core/domain.py
index d7a5783..7893c06 100644
--- a/src/farm_core/domain.py
+++ b/src/farm_core/domain.py
@@ -213,8 +213,10 @@ class Plot:
     growth_updated_at: int | None = None  # 上次工作量结算的 UTC 整数秒。
     next_pest_at: int | None = None  # 下一次生虫时间，最多每季三次。
     maturity_pests: int = 0  # 成熟瞬间冻结的虫数，决定本季减产。
+    stolen_count: int = 0  # 本季已被偷走数量；收获或铲除后归零。
 
     def __post_init__(self):
+        integer(self.stolen_count)
         integer(self.pest_count, 0, 3)
         integer(self.pest_spawned, self.pest_count, 3)
         integer(self.maturity_pests, 0, 3)
diff --git a/src/farm_core/presentation.py b/src/farm_core/presentation.py
index 2d0fd2a..5c2a3ab 100644
--- a/src/farm_core/presentation.py
+++ b/src/farm_core/presentation.py
@@ -4,6 +4,25 @@ from .domain import Receipt, Snapshot
 
 
 def receipt_text(receipt: Receipt) -> str:
+    if receipt.action == 'steal':
+        data = receipt.data
+        items = data['items']
+        lines = [f"偷菜成功：{len(items)}块地，共{data['quantity']}个。"
+                 if items else '本次未偷到作物，不消耗次数。']
+        grouped = {}
+        for item in items:
+            key = (item['plant_id'], item['name'])
+            grouped[key] = grouped.get(key, 0)+item['quantity']
+        lines.extend(f'{name}（#{pid}）×{amount}'
+                     for (pid, name), amount in grouped.items())
+        reasons = {'already_stolen': '本季已偷过',
+                   'protected': '防偷保护', 'not_ripe': '无成熟作物'}
+        notes = [f'{reasons[key]} {count}块'
+                 for key, count in data['skipped'].items() if count]
+        if notes:
+            lines.append('跳过：'+'；'.join(notes))
+        lines.append(f"今日剩余：{data['remaining_attempts']}/5次（北京时间）")
+        return '\n'.join(lines)
     labels = {
         "register": "注册", "buy": "购买", "sow": "播种",
         "harvest": "收获", "eradicate": "铲除",
diff --git a/src/farm_core/repository.py b/src/farm_core/repository.py
index 6842192..917a6d0 100644
--- a/src/farm_core/repository.py
+++ b/src/farm_core/repository.py
@@ -138,7 +138,7 @@ class SQLiteRepository:
                     r["yield_bonus"], r["experience_bonus"],
                     r["pest_count"], r["pest_spawned"], r["growth_remaining"],
                     r["growth_updated_at"], r["next_pest_at"],
-                    r["maturity_pests"],
+                    r["maturity_pests"], r['stolen_count'],
                 )
                 for r in rows
             )
@@ -154,14 +154,41 @@ class SQLiteRepository:
             "UPDATE plots SET plant=?, mature_at=?, harvest_count=?, "
             "yield_bonus=?, experience_bonus=?, pest_count=?, pest_spawned=?, "
             "growth_remaining=?, growth_updated_at=?, next_pest_at=?, "
-            "maturity_pests=? WHERE identity=? AND slot=?",
+            "maturity_pests=?, stolen_count=? WHERE identity=? AND slot=?",
             (
                 data, plot.mature_at, plot.harvest_count, plot.yield_bonus,
                 plot.experience_bonus, plot.pest_count, plot.pest_spawned,
                 plot.growth_remaining, plot.growth_updated_at,
-                plot.next_pest_at, plot.maturity_pests,
+                plot.next_pest_at, plot.maturity_pests, plot.stolen_count,
                 self._identity, plot.slot,
             ),
         )
         if cursor.rowcount != 1:
             raise FarmError("unknown_plot")
+
+    def for_identity(self, identity):
+        """同一事务内访问目标账户，禁止独立提交。"""
+        return SQLiteRepository(self._db, identity.key, self._operation_id,
+                                self._action, self._now)
+
+    def theft_seen(self, slot, thief):
+        return self._db.execute(
+            'SELECT 1 FROM thefts WHERE owner=? AND slot=? AND thief=?',
+            (self._identity, slot, thief.key),
+        ).fetchone() is not None
+
+    def record_theft(self, slot, thief):
+        self._db.execute('INSERT INTO thefts VALUES (?, ?, ?)',
+                         (self._identity, slot, thief.key))
+
+    def clear_thefts(self, slot):
+        self._db.execute('DELETE FROM thefts WHERE owner=? AND slot=?',
+                         (self._identity, slot))
+
+    def successful_thefts(self, day_start):
+        return self._db.execute(
+            "SELECT COUNT(*) FROM operations WHERE identity=? "
+            "AND action='steal' AND created_at>=? AND created_at<? "
+            "AND json_extract(result, '$.quantity')>0",
+            (self._identity, day_start, day_start+86400),
+        ).fetchone()[0]
diff --git a/src/farm_core/storage.py b/src/farm_core/storage.py
index 405ee73..f0b12bb 100644
--- a/src/farm_core/storage.py
+++ b/src/farm_core/storage.py
@@ -16,7 +16,7 @@ from .catalogue import Catalogue
 
 LOGGER = logging.getLogger(__name__)
 APPLICATION_ID = 0x4641524D
-SCHEMA_VERSION = 3
+SCHEMA_VERSION = 4
 
 PEST_COLUMNS = {
     "pest_count": (
@@ -72,11 +72,19 @@ SCHEMA = (
         harvest_count INTEGER NOT NULL DEFAULT 0 CHECK(harvest_count>=0),
         yield_bonus INTEGER NOT NULL DEFAULT 0,
         experience_bonus INTEGER NOT NULL DEFAULT 0,
+        stolen_count INTEGER NOT NULL DEFAULT 0 CHECK(stolen_count>=0),
         """ + ",\n        ".join(
             f"{name} {definition}" for name, definition in PEST_COLUMNS.items()
         ) + """,
         PRIMARY KEY(identity, slot)
     )""",
+    """CREATE TABLE thefts(
+        owner TEXT NOT NULL,
+        slot INTEGER NOT NULL,
+        thief TEXT NOT NULL REFERENCES accounts(identity),
+        PRIMARY KEY(owner, slot, thief),
+        FOREIGN KEY(owner, slot) REFERENCES plots(identity, slot)
+    )""",
     """CREATE TABLE ledger(
         id INTEGER PRIMARY KEY,
         identity TEXT NOT NULL,
@@ -137,6 +145,8 @@ class SQLiteStore:
             )
         }
         expected = {"accounts", "operations", "inventory", "plots", "ledger"}
+        if expected_version >= 4:
+            expected.add('thefts')
         if (
             app_id != APPLICATION_ID or version != expected_version
             or tables != expected
@@ -227,15 +237,28 @@ class SQLiteStore:
                         )
                     connection.execute("PRAGMA user_version=3")
                     migrated_pests = True
+                    version = 3
+                if app == APPLICATION_ID and version == 3:
+                    self._validate(connection, expected_version=3)
+                    connection.execute(
+                        'ALTER TABLE plots ADD COLUMN stolen_count '
+                        'INTEGER NOT NULL DEFAULT 0 CHECK(stolen_count>=0)'
+                    )
+                    connection.execute(next(
+                        sql for sql in SCHEMA
+                        if sql.startswith('CREATE TABLE thefts(')
+                    ))
+                    connection.execute('PRAGMA user_version=4')
+                    LOGGER.info('farm schema migrated version=4 theft=enabled')
                 self._validate(connection)
                 connection.commit()
                 if migrated_rows is not None:
                     LOGGER.info(
-                        "farm schema migrated version=3 inventory_rows=%s",
+                        "farm schema migrated version=4 inventory_rows=%s",
                         migrated_rows,
                     )
                 elif migrated_pests:
-                    LOGGER.info("farm schema migrated version=3 pests=enabled")
+                    LOGGER.info("farm schema migrated version=4 pests=enabled")
             except BaseException:
                 connection.rollback()
                 LOGGER.warning("farm schema initialization rolled back")
diff --git a/src/farm_core/yield_rules.py b/src/farm_core/yield_rules.py
new file mode 100644
index 0000000..3310bfa
--- /dev/null
+++ b/src/farm_core/yield_rules.py
@@ -0,0 +1,12 @@
+"""本季产量的唯一口径：土地加成、成熟虫害、实际偷取扣减。"""
+
+from .domain import boosted
+
+
+def season_yield(plot):
+    return (boosted(plot.plant.harvest, plot.yield_bonus)
+            * (100-5*plot.maturity_pests)//100)
+
+
+def remaining_yield(plot):
+    return max(0, season_yield(plot)-plot.stolen_count)
diff --git a/src/farm_render/details.py b/src/farm_render/details.py
index cb436fd..6534c8e 100644
--- a/src/farm_render/details.py
+++ b/src/farm_render/details.py
@@ -56,7 +56,8 @@ def detail_rows(renderer, identity, snapshot, now, page):
             continue
         ripe = plot.mature_at is not None and plot.mature_at <= now
         pests = plot.maturity_pests if ripe else plot.pest_count
-        amount = boosted(plant.harvest, plot.yield_bonus)*(100-5*pests)//100
+        amount = max(0, boosted(plant.harvest, plot.yield_bonus)
+                     *(100-5*pests)//100-plot.stolen_count)
         info = metadata['plants'].get(plant.name)
         total = len(set(info['phase'].split(','))) if info else None
         stage = ('成熟' if ripe else
diff --git a/src/farm_render/service.py b/src/farm_render/service.py
index 4f2d8cf..6255b12 100644
--- a/src/farm_render/service.py
+++ b/src/farm_render/service.py
@@ -118,17 +118,21 @@ class RenderService:
         return await asyncio.shield(task)
 
     async def warehouse(self, rows, snapshot, kind, page, pages,
-                        image_format='AUTO'):
+                        image_format='AUTO', high=False):
         if self._closed:
             raise RuntimeError('Render service closed')
-        key = ('warehouse', rows, snapshot, kind, page, pages, image_format)
+        key = ('warehouse', rows, snapshot, kind, page, pages,
+               image_format, high)
+        cached = self.renderer.results.get(key)
+        if cached is not None:
+            return cached
         task = self._pending.get(key)
         if task is None:
             if len(self._pending) >= self._max_pending:
                 raise RuntimeError('Render queue full')
             task = asyncio.create_task(self._run_function(
                 render_warehouse, self.renderer, rows, snapshot, kind,
-                page, pages, image_format,
+                page, pages, image_format, high,
             ))
             self._pending[key] = task
 
@@ -138,7 +142,9 @@ class RenderService:
                     done.exception()
 
             task.add_done_callback(finished)
-        return await asyncio.shield(task)
+        content = await asyncio.shield(task)
+        self.renderer.results.put(key, content, len(content))
+        return content
 
     async def _run_function(self, function, *args):
         started = perf_counter()
diff --git a/src/farm_render/warehouse.py b/src/farm_render/warehouse.py
index 7adaa33..6c672d3 100644
--- a/src/farm_render/warehouse.py
+++ b/src/farm_render/warehouse.py
@@ -110,15 +110,25 @@ def warehouse_text(rows, snapshot, kind, page, pages):
 
 
 def render_warehouse(renderer, rows, snapshot, kind, page, pages,
-                     image_format='AUTO'):
-    options = RenderOptions(format=image_format, quality=80)
+                     image_format='AUTO', high=False):
+    options = RenderOptions(width=1024 if high else 768,
+                            format=image_format, quality=85 if high else 70)
     codec = options.codec()
+    scale = options.width/1024
+    s = lambda n: round(n*scale)
     canvas = Canvas(renderer.assets.image('webgl/shop-board.png',
-                                         (1024, 1536)))
-    patch = canvas.image.crop((300, 340, 880, 445))
-    canvas.image.paste(patch.resize((824, 930)), (100, 320))
-    canvas.rectangle((105, 322, 919, 1249), '#fff1cf', radius=15)
+                                         (s(1024), s(1536))))
+    patch = canvas.image.crop(tuple(s(n) for n in (300, 340, 880, 445)))
+    canvas.image.paste(patch.resize((s(824), s(930))), (s(100), s(320)))
+
+    def rectangle(box, color, radius):
+        canvas.rectangle(tuple(s(n) for n in box), color, radius=s(radius))
+
+    rectangle((105, 322, 919, 1249), '#fff1cf', 15)
     def text(value, xy, size=26, color='#603817', width=760):
+        size = max(1, s(size))
+        xy = tuple(s(n) for n in xy)
+        width = s(width)
         value = canvas.fit_text(value, renderer.font_path, size, width)
         box = font(renderer.font_path, size).getbbox(value)
         canvas.text((xy[0], xy[1]-box[1]), value,
@@ -136,7 +146,7 @@ def render_warehouse(renderer, rows, snapshot, kind, page, pages,
     for index, row in enumerate(chunk):
         if row.group != group:
             group = row.group
-            canvas.rectangle((123, y, 899, y+43), '#d8e5b6', radius=7)
+            rectangle((123, y, 899, y+43), '#d8e5b6', 7)
             count = sum(r.group == group for r in rows)
             label = f'{group}（{count}种）'
             if group == '已锁定':
@@ -144,12 +154,13 @@ def render_warehouse(renderer, rows, snapshot, kind, page, pages,
             text(label, (140, y+8), 24, '#607d30')
             y += 49
         if index % 2 == 0:
-            canvas.rectangle((123, y, 899, y+55), '#fff8e4', radius=5)
+            rectangle((123, y, 899, y+55), '#fff8e4', 5)
         icon = f'plant/{row.name}/icon.png'
         if icon in renderer.assets.metadata['images']:
             image = renderer.assets.image(icon)
-            image.thumbnail((44, 46), Image.Resampling.LANCZOS)
-            canvas.paste(image, (158-image.width//2, y+27-image.height//2))
+            image.thumbnail((s(44), s(46)), Image.Resampling.LANCZOS)
+            canvas.paste(image, (s(158)-image.width//2,
+                                 s(y+27)-image.height//2))
         else:
             text('?', (149, y+15), 26)
         text(f'#{row.plant_id}', (193, y+19), 21, '#607d30', 82)
@@ -186,6 +197,7 @@ def render_warehouse(renderer, rows, snapshot, kind, page, pages,
     x = round((1024-canvas.measure_text(nav, renderer.font_path, 23))/2)
     text(nav, (x, 1424), 23, width=510)
     content = canvas.encode(codec, options.quality, options.method)
-    logger.info('farm warehouse kind=%s page=%s bytes=%s codec=%s',
-                kind, page, len(content), codec)
+    logger.info('farm warehouse kind=%s page=%s bytes=%s codec=%s width=%s '
+                'quality=%s', kind, page, len(content), codec,
+                options.width, options.quality)
     return content
diff --git a/tests/test_crop_identity.py b/tests/test_crop_identity.py
index c209586..0bc329d 100644
--- a/tests/test_crop_identity.py
+++ b/tests/test_crop_identity.py
@@ -116,6 +116,8 @@ class MigrationTests(unittest.IsolatedAsyncioTestCase):
     def create_v1(self, path, unknown=False, version=1):
         with closing(sqlite3.connect(path)) as db, db:
             for statement in SCHEMA:
+                if statement.startswith('CREATE TABLE thefts('):
+                    continue
                 if version == 1 and statement.startswith("CREATE TABLE inventory("):
                     statement = """CREATE TABLE inventory(
                         identity TEXT, name TEXT, seeds INTEGER,
diff --git a/tests/test_stealing.py b/tests/test_stealing.py
new file mode 100644
index 0000000..1a416eb
--- /dev/null
+++ b/tests/test_stealing.py
@@ -0,0 +1,156 @@
+import asyncio
+from contextlib import closing
+from dataclasses import replace
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from farm_core import Catalogue, FarmService, Identity, SQLiteStore
+from farm_core.domain import FarmError, Plant, Plot
+from farm_core.repository import SQLiteRepository
+from farm_core.yield_rules import remaining_yield
+
+
+class StealingTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.store = SQLiteStore(Path(self.temp.name) / 'farm.db')
+        await self.store.initialize()
+        self.now = 100000
+        self.plant = Plant(1, '测试菜', 0, 1, False, 0, 2, 10,
+                           1, 100, 2, 100)
+        self.service = FarmService(self.store, Catalogue([self.plant]),
+                                   lambda: self.now)
+        self.owner = Identity('test', 'local', 'qq', '1')
+        self.thief = replace(self.owner, user_id='2')
+        for who in (self.owner, self.thief):
+            await self.service.register(who, who.user_id, 'register')
+        self.fixture = 0
+
+    async def asyncTearDown(self):
+        await self.store.close()
+        self.temp.cleanup()
+
+    async def prepare(self, plots, owner=None):
+        self.fixture += 1
+        await self.store.transact(
+            owner or self.owner, f'fixture-{self.fixture}', 'fixture', {},
+            self.now, lambda repo: [repo.save_plot(p) for p in plots],
+        )
+
+    async def test_each_target_plot_random_bounds_replay_and_repeat(self):
+        await self.prepare([Plot(i, self.plant, self.now) for i in range(1, 7)])
+        with patch('farm_core.application.secrets.randbelow',
+                   side_effect=[0, 2, 0, 2, 0, 2]):
+            result = await self.service.steal(self.thief, self.owner, 'first')
+        self.assertEqual([i['quantity'] for i in result.data['items']],
+                         [1, 3, 1, 3, 1, 3])
+        self.assertEqual(result.data['remaining_attempts'], 4)
+        self.assertEqual(await self.service.steal(
+            self.thief, self.owner, 'first'), result)
+        repeated = await self.service.steal(self.thief, self.owner, 'again')
+        self.assertEqual(repeated.data['quantity'], 0)
+        self.assertEqual(repeated.data['skipped']['already_stolen'], 6)
+        self.assertEqual(repeated.data['remaining_attempts'], 4)
+        self.assertEqual((await self.service.snapshot(self.thief))
+                         .inventory[0].crops, 12)
+
+    async def test_protection_odd_yield_pests_and_bonus(self):
+        odd = replace(self.plant, harvest=9)
+        await self.prepare([
+            Plot(1, odd, self.now, stolen_count=3),
+            Plot(2, self.plant, self.now, stolen_count=5),
+            Plot(3, self.plant, self.now, yield_bonus=20,
+                 maturity_pests=3, stolen_count=4),
+        ])
+        with patch('farm_core.application.secrets.randbelow', return_value=2):
+            result = await self.service.steal(self.thief, self.owner, 'steal')
+        self.assertEqual([i['quantity'] for i in result.data['items']], [1, 1])
+        self.assertEqual(result.data['skipped']['protected'], 1)
+        plots = (await self.service.snapshot(self.owner)).plots
+        self.assertEqual([remaining_yield(p) for p in plots[:3]], [5, 5, 5])
+
+    async def test_harvest_preserves_remaining_and_resets_next_season(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        with patch('farm_core.application.secrets.randbelow', return_value=2):
+            await self.service.steal(self.thief, self.owner, 'first')
+        receipt = await self.service.harvest(self.owner, 'harvest')
+        self.assertEqual(receipt.data['quantity'], 7)
+        self.assertEqual((await self.service.snapshot(self.owner))
+                         .plots[0].stolen_count, 0)
+        self.now += 1000
+        again = await self.service.steal(self.thief, self.owner, 'next')
+        self.assertGreater(again.data['quantity'], 0)
+
+    async def test_concurrent_thieves_never_cross_protected_half(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        thieves = [replace(self.thief, user_id=str(i)) for i in range(3, 13)]
+        for who in thieves:
+            await self.service.register(who, who.user_id, 'register')
+        with patch('farm_core.application.secrets.randbelow', return_value=2):
+            results = await asyncio.gather(*[
+                self.service.steal(who, self.owner, 'steal') for who in thieves
+            ])
+        self.assertEqual(sum(r.data['quantity'] for r in results), 5)
+        self.assertEqual(remaining_yield(
+            (await self.service.snapshot(self.owner)).plots[0]), 5)
+
+    async def test_daily_quota_failed_visits_free_persistent_and_midnight(self):
+        self.now = 86400-28800-1
+        empty = await self.service.steal(self.thief, self.owner, 'empty')
+        self.assertEqual(empty.data['remaining_attempts'], 5)
+        for i in range(5):
+            owner = replace(self.owner, user_id=str(i+20))
+            await self.service.register(owner, str(i), 'register')
+            await self.prepare([Plot(1, self.plant, self.now)], owner)
+            await self.service.steal(self.thief, owner, f'steal-{i}')
+        await self.store.close()
+        await self.store.initialize()
+        with self.assertRaisesRegex(FarmError, 'steal_daily_limit'):
+            await self.service.steal(self.thief, self.owner, 'limit')
+        self.now += 1
+        await self.prepare([Plot(1, self.plant, self.now)])
+        receipt = await self.service.steal(self.thief, self.owner, 'new-day')
+        self.assertEqual(receipt.data['remaining_attempts'], 4)
+
+    async def test_denied_targets_and_atomic_rollback(self):
+        for target, code in (
+            (self.thief, 'cannot_steal_self'),
+            (replace(self.owner, realm='other'), 'different_farm_scope'),
+            (replace(self.owner, user_id='missing'), 'target_not_registered'),
+        ):
+            with self.assertRaisesRegex(FarmError, code):
+                await self.service.steal(self.thief, target, code)
+        await self.prepare([Plot(1, self.plant, self.now)])
+        before = [await self.service.snapshot(w)
+                  for w in (self.owner, self.thief)]
+        with patch.object(SQLiteRepository, 'record_theft',
+                          side_effect=sqlite3.OperationalError('fixture')):
+            with self.assertRaises(Exception):
+                await self.service.steal(self.thief, self.owner, 'rollback')
+        self.assertEqual([await self.service.snapshot(w)
+                          for w in (self.owner, self.thief)], before)
+        receipt = await self.service.steal(self.thief, self.owner, 'rollback')
+        self.assertGreater(receipt.data['quantity'], 0)
+
+    async def test_v3_migration_preserves_plots_and_initializes_theft_state(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        before = await self.service.snapshot(self.owner)
+        await self.store.close()
+        with closing(sqlite3.connect(self.store.path)) as db, db:
+            db.execute('DROP TABLE thefts')
+            db.execute('ALTER TABLE plots DROP COLUMN stolen_count')
+            db.execute('PRAGMA user_version=3')
+        await self.store.initialize()
+        self.assertEqual(await self.service.snapshot(self.owner), before)
+        await self.service.steal(self.thief, self.owner, 'after-migration')
+
+    async def test_concurrent_harvest_and_steal_conserve_yield(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        results = await asyncio.gather(
+            self.service.harvest(self.owner, 'harvest'),
+            self.service.steal(self.thief, self.owner, 'steal'),
+        )
+        self.assertEqual(sum(r.data['quantity'] for r in results), 10)
diff --git a/tests/test_warehouse.py b/tests/test_warehouse.py
index 7ea488d..c4ca130 100644
--- a/tests/test_warehouse.py
+++ b/tests/test_warehouse.py
@@ -72,5 +72,10 @@ class WarehouseTests(unittest.TestCase):
                                        kind, 1, 1)
             with Image.open(BytesIO(content)) as image:
                 self.assertEqual(image.format, 'JPEG')
-                self.assertEqual(image.size, (1024, 1536))
+                self.assertEqual(image.size, (768, 1152))
             self.assertLess(len(content), 500000)
+            high = render_warehouse(renderer, rows, self.snapshot,
+                                    kind, 1, 1, high=True)
+            with Image.open(BytesIO(high)) as image:
+                self.assertEqual(image.size, (1024, 1536))
+            self.assertGreater(len(high), len(content))
diff --git a/tests/test_zhenxun_runtime.py b/tests/test_zhenxun_runtime.py
index 2c11199..e102290 100644
--- a/tests/test_zhenxun_runtime.py
+++ b/tests/test_zhenxun_runtime.py
@@ -98,6 +98,13 @@ class ZhenxunRuntimeTests(unittest.IsolatedAsyncioTestCase):
             query.assert_awaited_once()
         with Image.open(BytesIO(content)) as image:
             self.assertEqual(image.format, 'JPEG')
+            self.assertEqual(image.size, (768, 1152))
+        high = await self.call('seeds', '1 高清', message=None)
+        with Image.open(BytesIO(high)) as image:
+            self.assertEqual(image.size, (1024, 1536))
+        self.assertIn('胡萝卜（#1） 2粒', await self.call(
+            'seeds', '文字', message=None,
+        ))
         self.assertEqual(before, await self.runtime.service.snapshot(
             self.runtime_identity(),
         ))
@@ -115,6 +122,37 @@ class ZhenxunRuntimeTests(unittest.IsolatedAsyncioTestCase):
         finally:
             self.runtime.rendering = rendering
 
+    async def test_steal_target_arguments_and_text_receipts(self):
+        from farm_core.domain import Plot
+        await self.call('register')
+        await self.call('register', user_id='2')
+        owner = replace(self.runtime_identity(), user_id='2')
+        plant = self.runtime.service.catalogue.get('胡萝卜')
+        now = self.runtime.service.clock()
+        await self.runtime.service.store.transact(
+            owner, 'fixture', 'fixture', {}, now,
+            lambda repo: repo.save_plot(Plot(1, plant, now)),
+        )
+        self.assertIn('偷菜', await self.call('steal'))
+        before = await self.runtime.service.snapshot(owner)
+        for account in ('1', '2', '9999'):
+            self.assertIn('不支持直接输入QQ号', await self.call(
+                'steal', account, message='numeric-'+account,
+            ))
+        self.assertEqual(await self.runtime.service.snapshot(owner), before)
+        self.assertIn('自己', await self.call('steal', target_user_id='1'))
+        self.assertIn('开通', await self.call('steal', target_user_id='9999'))
+        result = await self.call('steal', message='steal', target_user_id='2')
+        self.assertIsInstance(result, str)
+        self.assertIn('胡萝卜', result)
+        self.assertIn('4/5', result)
+        repeated = await self.call('steal', message='again', target_user_id='2')
+        self.assertIn('本季已偷', repeated)
+        self.assertIn('4/5', repeated)
+        self.assertIn('目标', await self.call(
+            'steal', '2', target_user_id='2', message='invalid',
+        ))
+
     async def test_original_commands_transactions_and_png(self):
         self.assertNotIn("测试农场", self.module.COMMANDS)
         self.assertIn("开通农场", self.module.COMMANDS)
@@ -442,6 +480,27 @@ class ZhenxunRuntimeTests(unittest.IsolatedAsyncioTestCase):
                     SimpleNamespace(extract_plain_text=lambda:""),
                 )
                 self.assertTrue(sent)
+                from unittest.mock import AsyncMock
+
+                class Arguments(list):
+                    def extract_plain_text(self):
+                        return ''
+
+                for segment in (
+                    SimpleNamespace(type='at', data={'qq': '123'}),
+                    SimpleNamespace(type='mention_user',
+                                    data={'user_id': '123'}),
+                ):
+                    with patch.object(imported.runtime, 'execute',
+                                      new=AsyncMock(return_value='结果')) as run:
+                        await hooks['commands']['偷菜'](
+                            bot, SimpleNamespace(message_id='steal'),
+                            SimpleNamespace(user=SimpleNamespace(
+                                id='u', name='用户',
+                            )), Arguments([segment]),
+                        )
+                        self.assertEqual(run.await_args.kwargs[
+                            'target_user_id'], '123')
             finally:
                 await hooks["close"][0]()
             self.assertFalse(imported.runtime.started)
diff --git a/version.txt b/version.txt
index 1d5e9e0..63ebd3f 100644
--- a/version.txt
+++ b/version.txt
@@ -1 +1 @@
-1.5.3
\ No newline at end of file
+1.5.4
\ No newline at end of file