diff --git a/cloud/auto_memory.py b/cloud/auto_memory.py
index c324b01..bb34e89 100644
--- a/cloud/auto_memory.py
+++ b/cloud/auto_memory.py
@@ -53,12 +53,15 @@ cloud/migrations.py 的对齐说明；PG DDL 同步进
 cloud/db.py POSTGRES_SCHEMA 与 MIGRATION_TABLES。
 """
 
-import os
 import re
 import time
 
 from cloud.db import connect as _db_connect
 from cloud.identity import IdentityStore
+# 行取值统一走 peer_messages.row_get（A7）：本模块旧私有版与之
+# 同款语义（缺列/缺键/空行回 default，默认 None），调用处仍
+# 用 _row_get 别名，行为不变。
+from cloud.peer_messages import row_get as _row_get
 from cloud.quota import default_quota_usd_micros as _quota_default
 from cloud.quota import monthly_usage as _quota_monthly_usage
 from cloud.quota import quota_for_owner as _quota_for_owner
@@ -152,13 +155,6 @@ def quota_remaining_percent(db_path, owner):
         conn.close()
 
 
-def _row_get(row, name, default=None):
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return default
-
-
 def _first_line(text, max_len=200):
     for line in (text or "").splitlines():
         line = " ".join(line.split())
diff --git a/cloud/lifecycle.py b/cloud/lifecycle.py
index 8e0eba3..1f3ae49 100644
--- a/cloud/lifecycle.py
+++ b/cloud/lifecycle.py
@@ -31,6 +31,13 @@ import json
 import time
 import uuid
 
+# 行取值统一走 peer_messages.row_get（A7）：本模块旧私有版在
+# 真实行类型（None / dict / sqlite3.Row）下与之结果完全一致
+# （None 特判、dict .get、缺列回 None），调用处仍用 _row_get
+# 别名，行为不变；旧版宽 Exception 兜底的病态行不在契约内，
+# 见 peer_messages.row_get docstring。
+from cloud.peer_messages import row_get as _row_get
+
 SNAPSHOT_STATUS_LIMIT = 4000
 SNAPSHOT_DIFF_LIMIT = 20000
 REF_LIMIT = 2000
@@ -39,17 +46,6 @@ REF_KINDS = ("doc", "build", "external", "pr", "file")
 BINDING_MODES = ("heartbeat", "cron")
 
 
-def _row_get(row, name):
-    if row is None:
-        return None
-    if isinstance(row, dict):
-        return row.get(name)
-    try:
-        return row[name]
-    except Exception:
-        return None
-
-
 # -- worktree 快照 ------------------------------------------------------
 
 def ensure_snapshots_table(conn):
diff --git a/cloud/peer_messages.py b/cloud/peer_messages.py
index 378b0b9..b990749 100644
--- a/cloud/peer_messages.py
+++ b/cloud/peer_messages.py
@@ -62,14 +62,22 @@ def peer_message_prefix(from_task_id):
     return "[来自任务 %s 的消息]\n" % from_task_id
 
 
-def row_get(row, name):
-    """行取值容错：sqlite3.Row 缺列报 IndexError、dict 缺键
-    报 KeyError、空行报 TypeError，一律回 None（worker 直连
-    老库、列还没迁移时排空不能因此停摆）。"""
+def row_get(row, name, default=None):
+    """行取值容错（全仓行取值的统一口，A7）：sqlite3.Row 缺列
+    报 IndexError、dict 缺键报 KeyError、空行报 TypeError，
+    一律回 default（默认 None；worker 直连老库、列还没迁移时
+    排空不能因此停摆）。handler 链、auto_memory、lifecycle
+    三处统一走本函数：三者旧语义在真实行类型（None / dict /
+    sqlite3.Row）下完全一致——auto_memory 旧版的 default 参数
+    默认就是 None，lifecycle 旧版的 None 特判与 dict .get 在
+    这些类型下与本函数结果相同；旧 lifecycle 版另带的宽
+    Exception 兜底只对“__getitem__ 抛其它异常的病态行”有
+    区别，那类输入不是 DB 行，不属于本函数契约，统一后此
+    类输入改为抛给调用方（不在测试锁定的真实行类型内）。"""
     try:
         return row[name]
     except (KeyError, IndexError, TypeError):
-        return None
+        return default
 
 
 def extract_peer_messages(text):
diff --git a/cloud/tests/test_row_get_unified.py b/cloud/tests/test_row_get_unified.py
new file mode 100644
index 0000000..65f10b2
--- /dev/null
+++ b/cloud/tests/test_row_get_unified.py
@@ -0,0 +1,92 @@
+"""row_get 簇统一单测（A7）：三处走同一函数，语义钉住。
+
+只用标准库 + 内存 SQLite，不联网。锁定点：
+- 统一身份：auto_memory._row_get 与 lifecycle._row_get 就是
+  peer_messages.row_get 本体（不是又一份复写）。
+- 真实行类型矩阵（None / dict / sqlite3.Row）下取值与回落
+  与三处旧版一致：现值原样回（含 0 / 空串等假值），缺列 /
+  缺键 / 空行回 default（默认 None）。
+- default 回落差异钉住：三处旧版的默认回落都是 None——
+  auto_memory 旧版 default 参数默认 None，lifecycle 旧版无
+  default 参数一律回 None，peer 旧版回 None；统一后两模块
+  调用处仍不传 default，缺失时仍为 None，不算改语义。
+- 统一路径实测：lifecycle.snapshot_to_dict 在 dict 缺键行与
+  sqlite3.Row 行上都经统一口出正确字段。
+"""
+
+import sqlite3
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    import cloud.auto_memory as auto_memory
+    import cloud.lifecycle as lifecycle
+    from cloud.peer_messages import row_get
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    import cloud.auto_memory as auto_memory
+    import cloud.lifecycle as lifecycle
+    from cloud.peer_messages import row_get
+
+
+def _sqlite_row():
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    try:
+        conn.execute("CREATE TABLE t(a TEXT, b INTEGER)")
+        conn.execute("INSERT INTO t(a, b) VALUES('x', 0)")
+        full = conn.execute("SELECT a, b FROM t").fetchone()
+        partial = conn.execute("SELECT a FROM t").fetchone()
+        return full, partial
+    finally:
+        conn.close()
+
+
+class RowGetUnifiedTestCase(unittest.TestCase):
+    def test_all_three_are_the_same_function(self):
+        self.assertIs(auto_memory._row_get, row_get)
+        self.assertIs(lifecycle._row_get, row_get)
+
+    def test_real_row_type_matrix(self):
+        full, partial = _sqlite_row()
+        # 现值原样回（含假值 0，不被当成缺失）。
+        self.assertEqual(row_get({"a": 0}, "a"), 0)
+        self.assertEqual(row_get({"a": ""}, "a"), "")
+        self.assertEqual(row_get(full, "a"), "x")
+        self.assertEqual(row_get(full, "b"), 0)
+        # 缺键 / 缺列 / 空行一律回 default（默认 None）。
+        self.assertIsNone(row_get({"a": 1}, "missing"))
+        self.assertIsNone(row_get(partial, "b"))
+        self.assertIsNone(row_get(None, "a"))
+        # 显式 default 只在缺失时生效，现值 None 也不被替换
+        #（旧 auto_memory 版同款语义：取到什么回什么）。
+        self.assertEqual(row_get({"a": 1}, "missing", default="d"), "d")
+        self.assertEqual(row_get(None, "a", default=7), 7)
+        self.assertIsNone(row_get({"a": None}, "a", default="d"))
+
+    def test_default_fallback_is_none_in_both_modules(self):
+        # 两模块调用处均不传 default；缺失回落必须仍是 None，
+        # 否则就是改了旧语义。
+        self.assertIsNone(auto_memory._row_get({}, "nope"))
+        self.assertIsNone(lifecycle._row_get({}, "nope"))
+        self.assertIsNone(auto_memory._row_get(None, "nope"))
+        self.assertIsNone(lifecycle._row_get(None, "nope"))
+
+    def test_lifecycle_snapshot_to_dict_via_unified_get(self):
+        out = lifecycle.snapshot_to_dict({"id": "s1", "task_id": "t1"})
+        self.assertEqual(out["id"], "s1")
+        self.assertEqual(out["task_id"], "t1")
+        self.assertIsNone(out["project_id"])
+        self.assertIsNone(out["branch"])
+        self.assertEqual(out["untracked"], [])
+        full, _partial = _sqlite_row()
+        out2 = lifecycle.snapshot_to_dict(
+            {"id": "s2", "untracked_json": '["a.txt"]'})
+        self.assertEqual(out2["untracked"], ["a.txt"])
+        # sqlite Row 行同样走统一口：缺列不炸。
+        self.assertIsNone(row_get(full, "untracked_json"))
+
+
+if __name__ == "__main__":
+    unittest.main()