From 5db89d1b066a0d7b753e8a37754c9996ff44f268 Mon Sep 17 00:00:00 2001
From: host <host@test.local>
Date: Mon, 2 Nov 2026 03:41:50 +0000
Subject: [PATCH] code-to10 v6 A4: fix AuthStore.audit NameError (import
 sqlite3) + retry regression test

Pre-fix any first-attempt audit failure raised NameError at
auth.py:424 (sqlite3 never imported), aborting the 5-attempt
retry loop on attempt 1 and masking the original error. The
424-line coverage mystery is also explained: the only full-suite
hit is test_child_model.test_coordinator_inherits_child_defaults
via worker best-effort _audit swallowing an OperationalError
(database is locked); 422-423/426-427 stay uncovered because
rollback succeeds and NameError aborts before either raise.
Bubble audit (see report): no merge/release gate swallows the
NameError and proceeds, so this stays a downgrade, not a
production blocker.

(cherry picked from commit 039c6adbac3b120b5b9a1dcbafeff6a91407ce70)
---
 cloud/auth.py            |  1 +
 cloud/tests/test_auth.py | 48 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+)

diff --git a/cloud/auth.py b/cloud/auth.py
index 039659e..b7dd2f6 100644
--- a/cloud/auth.py
+++ b/cloud/auth.py
@@ -23,6 +23,7 @@ approve_merge -> maintainer，manage_project -> owner（仅 owner）。
 
 import hashlib
 import secrets
+import sqlite3
 from cloud.db import connect as _db_connect
 from cloud.ledger import append_entry as _ledger_append
 from cloud.ledger import ensure_schema as _ledger_ensure
diff --git a/cloud/tests/test_auth.py b/cloud/tests/test_auth.py
index 1557677..73493b5 100644
--- a/cloud/tests/test_auth.py
+++ b/cloud/tests/test_auth.py
@@ -153,6 +153,54 @@ class TestAuth(unittest.TestCase):
         self.assertFalse(
             self.store.check_permission(target_id, project_id, "create_task"))
 
+    def test_audit_retries_integrity_error_and_reraises_others(self):
+        """A4 回归：audit() 重试分支的 sqlite3 引用必须存在。
+
+        首调 IntegrityError 后重试成功：审计对恰好一行（audit_log
+        1 行 + audit_ledger 1 行），不是 0 行也不是多行；非完整
+        性错误原样抛回（RuntimeError，不是 NameError 掩盖原错）。
+        背景：auth.py 曾全文件无 import sqlite3，audit() 的
+        except 分支在 424 行求 isinstance(exc, sqlite3.IntegrityError)
+        时直接 NameError，任何首调失败都走不到重试或原错抛回。
+        """
+        import cloud.auth as auth_mod
+
+        original = auth_mod.audit_on_conn
+        calls = {"n": 0}
+
+        def _flaky(conn, *args, **kwargs):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise sqlite3.IntegrityError(
+                    "UNIQUE constraint failed: audit_ledger.entry_hash")
+            return original(conn, *args, **kwargs)
+
+        auth_mod.audit_on_conn = _flaky
+        try:
+            self.store.audit("u1", "p1", "act", "d")
+        finally:
+            auth_mod.audit_on_conn = original
+        self.assertEqual(calls["n"], 2)
+        self.assertEqual(len(self.store.list_audit("p1")), 1)
+        ledger_rows = self.store.conn.execute(
+            "SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
+        self.assertEqual(ledger_rows, 1)
+
+        def _broken(conn, *args, **kwargs):
+            raise RuntimeError("ledger boom")
+
+        auth_mod.audit_on_conn = _broken
+        try:
+            with self.assertRaises(RuntimeError) as caught:
+                self.store.audit("u1", "p1", "act", "d2")
+        finally:
+            auth_mod.audit_on_conn = original
+        self.assertEqual(str(caught.exception), "ledger boom")
+        self.assertEqual(len(self.store.list_audit("p1")), 1)
+        ledger_rows = self.store.conn.execute(
+            "SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
+        self.assertEqual(ledger_rows, 1)
+
     def test_audit_append_only(self):
         project_id, owner_id, _v, _c, _m, _o = self._project_with_roles()
         self.assertFalse(hasattr(AuthStore, "delete_audit"))