rehearsal: union task_attempts schema (main old DDL + gap-3 ledger cols), relax 2 hook no-table asserts to no-rows, gateway-env fake tolerates main routing reads, extend single-writer fixture list by name (test-only inserts)

cloud/tests/test_attempt_hook.py

@@ -149,6 +149,19 @@ def _table_exists(db_path, table):
     return row[0] == 1
 
 
+def _task_attempts_row_count(db_path):
+    """task_attempts 行数（表不存在时计 0；COUNT(*) 与列集无关，
+    旧 DDL 表与并集表均可查）。"""
+    conn = sqlite3.connect(db_path)
+    try:
+        if not _table_exists(db_path, "task_attempts"):
+            return 0
+        return conn.execute(
+            "SELECT COUNT(*) FROM task_attempts").fetchone()[0]
+    finally:
+        conn.close()
+
+
 def _completed_runner(cmd):
     """Fake CLI: dispatch/run always succeed, host.py-style JSON."""
     if "dispatch" in cmd:
@@ -466,7 +479,11 @@ class TestWorkerHook(unittest.TestCase):
         self.assertFalse(worker.attempt_ledger_hook)
         result = worker.run_one()
         self.assertEqual(result["status"], "completed")
-        self.assertFalse(_table_exists(self.db_path, "task_attempts"))
+        # 演练注：main 侧 init_db 已先建同名旧 DDL 的
+        # task_attempts 表（见 attempt_ledger 并集表说明），
+        # “表不存在”在合入后不再是可观测口径；保留的意图是
+        # “关时不写行”，故此处断言零行而非表不存在。
+        self.assertEqual(_task_attempts_row_count(self.db_path), 0)
         self.assertFalse(
             _table_exists(self.db_path, "task_attempt_origin_markers"))
 
@@ -559,7 +576,9 @@ class TestWorkerHook(unittest.TestCase):
         worker = self._worker(attempt_ledger_hook=True, env={})
         result = worker.run_one()
         self.assertEqual(result["status"], "failed")
-        self.assertFalse(_table_exists(self.db_path, "task_attempts"))
+        # 演练注：同上，门失败不调钩子、故零行；表本身由
+        # main 侧 init_db 预建，不再断言表不存在。
+        self.assertEqual(_task_attempts_row_count(self.db_path), 0)
 
     def test_hook_on_cancelled_dispatch_writes_interrupted_row(self):
         # Established pattern (test_worker