code-to10 v6 A3: enforce signer separation on audit-verdict submission

process_audit_verdict now calls validate_signer_separation
before any state save or ledger write (previously the function
had no production caller on this path: it only ran inside
collect_failed_checks via decide_pushback, which itself has no
production caller). Signer is bound to the authenticated
caller -- a body signer differing from the caller is rejected
-- and executor/coordinator_approver/qa_tester must be
declared and distinct from the caller; missing field,
self-audit, or forged signer raises ValueError, mapped by
_loop_run to HTTP 400 with no side effects. No new route.
Honest limits are in the helper docstring: declared-identity
enforcement only (task rows store no authoritative executor /
coordinator / QA identity; task executor is a backend, not a
person), and audit-fix reaudit verdicts are out of scope
(the fix poster is not necessarily the re-auditor, so binding
signer to the fix poster would be a mis-binding).

(cherry picked from commit 56dce5165f4d08f38fe1730b9eea3b78f90cb97d)

Reconcile-port: handler_lifecycle.py docstring hunk redirected into cloud/server.py audit-verdict handler (target keeps 6-mixin split; enforcement itself is in audit_loop_wiring.py and applied cleanly).

cloud/tests/test_audit_loop_wiring.py

@@ -58,6 +58,11 @@ def fail_verdict(text="check x failed", source="audit", **extra):
         "failures": [{"check": "ci_green", "detail": text}],
         "passed_items": ["a", "b", "c"],
         "contract_version": VERSION,
+        # A3: signer is the caller (auditor-1 in _verdict); the
+        # other three roles must be declared and distinct.
+        "executor": "executor-1",
+        "coordinator_approver": "coordinator-1",
+        "qa_tester": "qa-1",
     }
     verdict.update(extra)
     return verdict
@@ -71,6 +76,10 @@ def pass_verdict(**extra):
         "failures": [],
         "passed_items": ["a", "b", "c"],
         "contract_version": VERSION,
+        # A3: see fail_verdict.
+        "executor": "executor-1",
+        "coordinator_approver": "coordinator-1",
+        "qa_tester": "qa-1",
     }
     verdict.update(extra)
     return verdict
@@ -141,7 +150,7 @@ class WiringTestCase(unittest.TestCase):
         self.assertTrue(verify_chain(self.store.conn)["ok"])
 
     def test_full_chain_fail_fix_reaudit_pass_unblocks(self):
-        first = self._verdict(fail_verdict())
+        self._verdict(fail_verdict())
         fix_task = {"id": "task-fix-1", "project_id": "proj-1",
                     "module_id": MODULE, "contract_version": VERSION}
         result = wiring.process_fix_completion(
@@ -435,5 +444,76 @@ class WiringTestCase(unittest.TestCase):
             decision["code"], "release_blocked_state_unavailable")
 
 
+class SignerSeparationTestCase(unittest.TestCase):
+    """A3: validate_signer_separation 在提交路径上的真实调用方。
+
+    覆盖：自审（调用方 = 声明的 executor/coordinator/QA 任一）
+    被拒且不落链不落状态；缺字段被拒；体里 signer 与调用方
+    不符被拒；声明三方均与调用方不同时正常提交。此前该校验
+    只活在重审清单函数里（collect_failed_checks，经无生产
+    调用方的 decide_pushback），提交路径上从未执行。
+
+    注意：本类故意不继承 WiringTestCase——继承会使父类全部
+    测试方法在本类下再跑一遍（全量计数虚增），故只复用其
+    setUp 形状，不复用其用例。
+    """
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.ledger_path = os.path.join(self.tmp.name, "ledger.db")
+        self.store = LedgerStore(self.ledger_path)
+        self.conn = sqlite3.connect(":memory:")
+        self.task = {"id": "task-parent", "project_id": "proj-1",
+                     "module_id": MODULE, "contract_version": VERSION}
+
+    def tearDown(self):
+        try:
+            self.store.close()
+        except Exception:
+            pass
+        self.conn.close()
+        self.tmp.cleanup()
+
+    def _verdict(self, verdict, **kwargs):
+        kwargs.setdefault("user_id", "auditor-1")
+        return wiring.process_audit_verdict(
+            self.conn, self.store, self.task, verdict, **kwargs)
+
+    def _chain_rows(self):
+        return self.store.conn.execute(
+            "SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
+
+    def test_self_audit_rejected_without_side_effects(self):
+        for field in ("executor", "coordinator_approver", "qa_tester"):
+            with self.subTest(field=field):
+                verdict = fail_verdict(**{field: "auditor-1"})
+                with self.assertRaises(ValueError) as caught:
+                    self._verdict(verdict)
+                self.assertIn("signer", str(caught.exception))
+                self.assertIsNone(
+                    wiring.load_loop_state(self.conn, MODULE, VERSION))
+                self.assertEqual(self._chain_rows(), 0)
+
+    def test_missing_identity_field_rejected(self):
+        verdict = fail_verdict()
+        del verdict["qa_tester"]
+        with self.assertRaises(ValueError):
+            self._verdict(verdict)
+        self.assertIsNone(
+            wiring.load_loop_state(self.conn, MODULE, VERSION))
+        self.assertEqual(self._chain_rows(), 0)
+
+    def test_forged_signer_rejected(self):
+        verdict = fail_verdict(signer="someone-else")
+        with self.assertRaises(ValueError):
+            self._verdict(verdict)
+        self.assertEqual(self._chain_rows(), 0)
+
+    def test_distinct_declared_roles_submit(self):
+        result = self._verdict(pass_verdict())
+        self.assertEqual(result["action_kind"], "pass")
+        self.assertFalse(result["merge_blocked"])
+
+
 if __name__ == "__main__":
     unittest.main()