test: 9.5-code review minors (M1 conversion test, M5 no-fingerprint audit test, M2/M4 wording)

- M1: test_goal_branch_via_convert_queued_then_claim actually calls
  server._goal_convert_queued on a queued auto-resumed task before
  claiming, instead of only inserting the post-conversion state.
- M5: test_worker_routing_audit_rows_carry_no_fingerprint drives a
  real run_one with an active policy and asserts worker_routing
  audit rows exist and carry no fingerprint (negative control
  verified: injecting fingerprint into the detail fails it).
- M2: 'all callers' wording tightened to the real counts
  (load_routing_policy: 2 production call sites; evaluate_shadow: 1).
- M4: content_fingerprint docstring now states the fingerprint covers
  the full normalized payload including fallback_models, which the
  router deliberately does not apply (fail-safe direction).
No behaviour change; M3 DESIGN.md edits are outside the repo.

cloud/models.py

@@ -418,8 +418,12 @@ class RoutingPolicy:
         """Content fingerprint of the parsed policy (9.5-code).
 
         sha256 of the canonical JSON (sorted keys, tight
-        separators) over the fields that decide routing and
-        capability checks: mode, tiers, tier_modes,
+        separators) over the full normalized policy payload
+        (not only the fields that decide routing and capability
+        checks: fallback_models is included even though the
+        router deliberately does not apply it, so the
+        fingerprint is fail-safe — it reports more changes,
+        never fewer): mode, tiers, tier_modes,
         tier_requirements, model_capabilities, fallback_models.
         The operator-declared ``version`` string is deliberately
         NOT part of the fingerprint: it is hand-filled and may
@@ -577,7 +581,9 @@ def load_routing_policy(env=None):
         # 置位，所以部分失败不会产生任务级审计；运营方
         # 只能从 worker 日志看到它，不能从审计里查到。
         # 这是已知缺口（非静默，但不如全失败可见），
-        # 改动返回类型会牵动全部调用方，本轨不改。
+        # 本轨不改返回类型：生产调用点只有 2 处（server
+        # 影子端点经 _load_routing_policy、worker 派发），
+        # 真正的成本是同步改调用方与测试，而不是“牵动全部”。
         _logger.warning(
             "routing policy partially failed (%s); continuing with"
             " the usable source", "; ".join(failures))

cloud/routing_shadow.py

@@ -71,9 +71,10 @@ def parse_worker_routing_detail(detail):
     9.5-code fail-open 清单：解析失败的行返回 None 后在
     evaluate_shadow 里直接跳过、不计数、不报错——样本无
     声消失（例如模型名含空白会使整行匹配失败，见 M7）。
-    这是已知缺口：改返回类型会牵动评估函数与端点，本轨
-    只在此处写明，不改语义；运营方无法从报告里看到丢
-    了多少行，只能靠审计行总数与样本数对不上来发现。
+    这是已知缺口：改返回类型只需同步评估函数的唯一生产
+    调用方（server 影子端点）与测试，本轨只在此处写明，
+    不改语义；运营方无法从报告里看到丢了多少行，只能靠
+    审计行总数与样本数对不上来发现。
     """
     if not isinstance(detail, str):
         return None

cloud/tests/test_95_code.py

@@ -40,6 +40,7 @@ from cloud.auth import AuthStore
 from cloud.models import parse_routing_policy
 from cloud.routing_dryrun import dry_run_policy_document
 from cloud.server import (
+    CloudControlPlaneHandler,
     _agent_tool_approvals_from_row,
     _agent_tools_from_row,
     create_server,
@@ -174,6 +175,64 @@ class GoalBranchTestCase(WorkerCase):
             conn.close()
         self.assertEqual(row[0], "做目标")
 
+    def test_goal_branch_via_convert_queued_then_claim(self):
+        # M1 strength (9.5-code minors): actually execute the
+        # conversion step itself, not just insert the
+        # post-conversion state. A queued non-goal task that was
+        # auto-resumed as platform_fresh is converted in place
+        # by server._goal_convert_queued, then claimed: the
+        # claim must reach the goal branch of
+        # _apply_resume_preamble (no _resume, no breakpoint).
+        _insert_task(self.db_path, "t2", prompt="做活",
+                     auto_resume_count=1)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            ensure_attempts_table(conn)
+            record_attempt(conn, "t2", 1, "platform_fresh",
+                           "worker_shutdown", -1,
+                           "做到一半：已建 a.py", "被平台杀掉")
+            conn.commit()
+        finally:
+            conn.close()
+        handler = CloudControlPlaneHandler.__new__(
+            CloudControlPlaneHandler)
+        handler.server = SimpleNamespace(db_path=self.db_path)
+        converted = handler._goal_convert_queued("t2", "做目标")
+        self.assertIsNotNone(converted)
+        self.assertEqual(converted["goal"], "做目标")
+        self.assertEqual(converted["goal_mode"], 1)
+        # Conversion is in-place on the queued row only: count
+        # and ledger untouched, no resume message planted.
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT status, goal_mode, auto_resume_count,"
+                " resume_message FROM tasks WHERE id = ?",
+                ("t2",)).fetchone()
+        finally:
+            conn.close()
+        self.assertEqual(row[0], "queued")
+        self.assertEqual(row[1], 1)
+        self.assertEqual(row[2], 1)
+        self.assertIsNone(row[3])
+        worker = self._worker()
+        task = worker.get_task("t2")
+        self.assertFalse(task.get("_resume"))
+        self.assertFalse(task.get("_goal_resume_breakpoint"))
+        injected = worker._apply_resume_preamble(task)
+        self.assertIn("上次尝试说明", injected["goal"])
+        self.assertTrue(injected["goal"].endswith("做目标"))
+        self.assertIn("上次尝试说明", injected["prompt"])
+        self.assertTrue(injected["prompt"].endswith("做活"))
+        # Memory only: the stored goal is untouched.
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute("SELECT goal FROM tasks WHERE id = ?",
+                               ("t2",)).fetchone()
+        finally:
+            conn.close()
+        self.assertEqual(row[0], "做目标")
+
 
 class FailOpenLogTestCase(WorkerCase):
     def test_audit_failure_logs_without_breaking_dispatch(self):
@@ -234,6 +293,71 @@ class FingerprintTestCase(unittest.TestCase):
                          a.content_fingerprint())
 
 
+def _completed_runner(cmd):
+    name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
+    return SimpleNamespace(
+        returncode=0,
+        stdout=json.dumps({"task": name, "status": "completed",
+                           "text": "done"}),
+        stderr="")
+
+
+class RoutingAuditNoFingerprintTestCase(unittest.TestCase):
+    # M5 (9.5-code minors): "fingerprint never enters audit rows"
+    # must be asserted on real audit rows, not just on the dry-run
+    # report. Drive a real run_one with an active policy so the
+    # worker writes a worker_routing audit row, then read the
+    # audit_log table directly. The test is sensitive: if anyone
+    # writes the fingerprint into a worker_routing detail, the
+    # "fingerprint" substring assertion fails; if routing stops
+    # writing audit rows at all, the "at least one row" assertion
+    # fails instead of passing vacuously.
+    def setUp(self):
+        self._old_push_allow = os.environ.get(
+            "CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
+        os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+
+    def tearDown(self):
+        if self._old_push_allow is None:
+            os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
+        else:
+            os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = (
+                self._old_push_allow)
+        self.tmp.cleanup()
+
+    def test_worker_routing_audit_rows_carry_no_fingerprint(self):
+        _insert_task(self.db_path, "t9", prompt="做活")
+        worker = Worker(
+            self.db_path, "/fake/host.py", runner=_completed_runner,
+            env=self.env, max_retries=0, retry_backoff_seconds=0,
+            routing_policy={"mode": "active",
+                            "tiers": {"default": "m1"}})
+        out = worker.run_one()
+        self.assertEqual(out["status"], "completed")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            rows = conn.execute(
+                "SELECT action, detail FROM audit_log").fetchall()
+        finally:
+            conn.close()
+        routing = [d for (a, d) in rows if a == "worker_routing"]
+        self.assertGreaterEqual(
+            len(routing), 1,
+            "expected at least one worker_routing audit row; "
+            "without it the fingerprint assertion would be vacuous")
+        for detail in routing:
+            self.assertNotIn("fingerprint", detail)
+            # The written format is exactly tier/source/applied/
+            # suggested/chosen; nothing else may ride along.
+            self.assertTrue(detail.startswith("t9 tier=default "))
+        for (_action, detail) in rows:
+            self.assertNotIn("fingerprint", detail or "")
+
+
 class _HttpServerCase(unittest.TestCase):
     def setUp(self):
         self._old = {k: os.environ.get(k) for k in (