fix: 9.5-code reachable items (parse unification, goal branch, fail-open logs, M7 test, fingerprint)

Parse: agent_tools/tool_approvals column parsing unified into agent_def shared helpers (parse None = undeclared; server keeps None, worker maps to []/{} at _task_from_row, dispatch behaviour unchanged). Goal: _apply_resume_preamble goal branch proven reachable via _goal_convert_queued on a queued auto-resumed task, kept with a test and corrected docs (not dead code). Fail-open: audit write failure and preamble lookup failure now log warnings (best-effort unchanged); partial policy failure and shadow parse drops documented as known gaps. M7: shadow endpoint explicit_capability_failures_truncated test. Fingerprint: RoutingPolicy.content_fingerprint (canonical JSON sha256, version excluded) surfaced only in dry-run output, never in audit rows. M1 wording: legacy gap stated as fixed stock persisting per-task until completion, not converging.

cloud/agent_def.py

@@ -200,6 +200,53 @@ class AgentDefinition:
         }
 
 
+def parse_agent_tools_column(raw):
+    """Parse one tasks.agent_tools column value (the single shared
+    parser for server and worker; 9.5-code unification).
+
+    Returns a list copy, or None when the value is missing, is not
+    a JSON list, or is otherwise dirty. None is the parse-layer
+    default and means "undeclared", not "no required tools": the
+    server keeps None in task serialization (an old row with a
+    dirty value serializes exactly as before), while the worker
+    maps None to [] at the _task_from_row boundary, because
+    dispatch treats "no declaration" and "no required tools"
+    identically (both skip the v3 gate, and every worker use site
+    already reads ``task.get("agent_tools") or []``). Unifying the
+    parse logic therefore changes no observable behaviour on
+    either side; the old server/worker defaults (None vs []) were
+    only different spellings of the same dispatch outcome.
+    """
+    if raw is None:
+        return None
+    if isinstance(raw, list):
+        return list(raw)
+    try:
+        parsed = json.loads(raw)
+    except (TypeError, ValueError):
+        return None
+    return list(parsed) if isinstance(parsed, list) else None
+
+
+def parse_agent_tool_approvals_column(raw):
+    """Parse one tasks.agent_tool_approvals column value (the
+    single shared parser for server and worker; same None
+    convention as parse_agent_tools_column: None means
+    "undeclared", the server keeps it, the worker maps it to {}
+    at the _task_from_row boundary because every worker use site
+    already reads ``task.get("agent_tool_approvals") or {}``).
+    """
+    if raw is None:
+        return None
+    if isinstance(raw, dict):
+        return dict(raw)
+    try:
+        parsed = json.loads(raw)
+    except (TypeError, ValueError):
+        return None
+    return dict(parsed) if isinstance(parsed, dict) else None
+
+
 def effective_tool_approvals(tool_approvals, host_approval):
     """What host.py will actually answer per declared tool, today.
 

cloud/models.py

@@ -414,6 +414,37 @@ class RoutingPolicy:
         """Effective mode for one tier: explicit tier mode, else global."""
         return self.tier_modes.get(tier, self.mode)
 
+    def content_fingerprint(self):
+        """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,
+        tier_requirements, model_capabilities, fallback_models.
+        The operator-declared ``version`` string is deliberately
+        NOT part of the fingerprint: it is hand-filled and may
+        stay unchanged while requires change, so it cannot be
+        trusted to detect a change. The fingerprint is only for
+        dry-run output and debugging; it is never written into
+        worker_routing audit rows, so the shadow parser (which
+        splits audit detail on whitespace) is untouched.
+        Two policies that parse to the same effective fields
+        share a fingerprint even if their raw documents differ
+        in key order or in dropped/invalid tiers.
+        """
+        import hashlib
+        payload = {
+            "mode": self.mode,
+            "tiers": self.tiers,
+            "tier_modes": self.tier_modes,
+            "tier_requirements": self.tier_requirements,
+            "model_capabilities": self.model_capabilities,
+            "fallback_models": self.fallback_models,
+        }
+        canonical = json.dumps(payload, sort_keys=True,
+                               separators=(",", ":"))
+        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
 
 def parse_routing_policy(data):
     """Parse one policy document, or return None when it is unusable.
@@ -538,6 +569,15 @@ def load_routing_policy(env=None):
                 " worker default", "; ".join(failures))
         return None
     if failures:
+        # 9.5-code fail-open 清单：部分失败是 fail-open——
+        # 可用来源继续生效、失败来源被丢弃，派发不炸。
+        # 可见性只有这条警告日志：load_routing_policy 仍
+        # 返回一个 policy，worker 的
+        # routing_policy_load_failed 标志只在全失败时
+        # 置位，所以部分失败不会产生任务级审计；运营方
+        # 只能从 worker 日志看到它，不能从审计里查到。
+        # 这是已知缺口（非静默，但不如全失败可见），
+        # 改动返回类型会牵动全部调用方，本轨不改。
         _logger.warning(
             "routing policy partially failed (%s); continuing with"
             " the usable source", "; ".join(failures))

cloud/routing_dryrun.py

@@ -118,6 +118,7 @@ def dry_run_policy_document(data, env_mode_override=None,
     if not isinstance(data, dict):
         return {
             "valid": False, "blocked": True, "version": None,
+            "content_fingerprint": None,
             "global_mode": None, "global_mode_source": None,
             "tiers": tiers_report, "dropped_tiers": dropped,
             "unknown_tiers": unknown,
@@ -141,6 +142,7 @@ def dry_run_policy_document(data, env_mode_override=None,
             "entirely and workers would keep the worker default")
         return {
             "valid": False, "blocked": True, "version": None,
+            "content_fingerprint": None,
             "global_mode": None, "global_mode_source": None,
             "tiers": tiers_report, "dropped_tiers": dropped,
             "unknown_tiers": unknown, "warnings": warnings,
@@ -311,7 +313,9 @@ def dry_run_policy_document(data, env_mode_override=None,
     blocked = bool(dropped or unknown or cap_blocked or _gw_blocked)
     return {
         "valid": True, "blocked": blocked,
-        "version": policy.version, "global_mode": global_mode,
+        "version": policy.version,
+        "content_fingerprint": policy.content_fingerprint(),
+        "global_mode": global_mode,
         "global_mode_source": global_source,
         "tiers": tiers_report, "dropped_tiers": dropped,
         "unknown_tiers": unknown, "warnings": warnings,

cloud/routing_shadow.py

@@ -68,6 +68,12 @@ def parse_worker_routing_detail(detail):
 
     Old format (no suggested/chosen) parses with both models None.
     A ``-`` placeholder also means None, never a model named "-".
+    9.5-code fail-open 清单：解析失败的行返回 None 后在
+    evaluate_shadow 里直接跳过、不计数、不报错——样本无
+    声消失（例如模型名含空白会使整行匹配失败，见 M7）。
+    这是已知缺口：改返回类型会牵动评估函数与端点，本轨
+    只在此处写明，不改语义；运营方无法从报告里看到丢
+    了多少行，只能靠审计行总数与样本数对不上来发现。
     """
     if not isinstance(detail, str):
         return None

cloud/server.py

@@ -60,6 +60,8 @@ from urllib.parse import parse_qs, quote, urlparse
 from cloud.agent_def import default_agents_dir as _agent_default_dir
 from cloud.agent_def import list_known_skills as _agent_known_skills
 from cloud.agent_def import parse_agent_definition as _parse_agent_definition
+from cloud.agent_def import parse_agent_tool_approvals_column as _parse_tool_approvals_col
+from cloud.agent_def import parse_agent_tools_column as _parse_tools_col
 from cloud.agent_def import resolve_registry_agent as _resolve_agent
 from cloud.attempts import ensure_attempts_table as _ensure_attempts
 from cloud.auth import ROLES as _MEMBER_ROLES
@@ -1798,32 +1800,19 @@ def _row_get(row, name):
 
 
 def _agent_tools_from_row(row):
-    """agent_tools 列的容错解析：JSON 数组，非数组/脏数据为 None。"""
-    raw = _row_get(row, "agent_tools")
-    if raw is None:
-        return None
-    if isinstance(raw, list):
-        return list(raw)
-    try:
-        parsed = json.loads(raw)
-    except (TypeError, ValueError):
-        return None
-    return list(parsed) if isinstance(parsed, list) else None
+    """agent_tools 列的容错解析：唯一实现在
+    agent_def.parse_agent_tools_column（9.5-code 统一）；本
+    函数只取列值并转发，缺列/脏数据为 None（= 未声明），
+    序列化口径与统一前逐字一致。"""
+    return _parse_tools_col(_row_get(row, "agent_tools"))
 
 
 def _agent_tool_approvals_from_row(row):
-    """agent_tool_approvals 列的容错解析：JSON 对象，非对象/脏
+    """agent_tool_approvals 列的容错解析：唯一实现在
+    agent_def.parse_agent_tool_approvals_column；缺列/脏
     数据为 None（NULL = 未声明，与 v3 tools 同款语义）。"""
-    raw = _row_get(row, "agent_tool_approvals")
-    if raw is None:
-        return None
-    if isinstance(raw, dict):
-        return dict(raw)
-    try:
-        parsed = json.loads(raw)
-    except (TypeError, ValueError):
-        return None
-    return dict(parsed) if isinstance(parsed, dict) else None
+    return _parse_tool_approvals_col(
+        _row_get(row, "agent_tool_approvals"))
 
 
 def _task_to_dict(row, attachments=None, total_tokens=None):

cloud/tests/test_95_code.py

@@ -0,0 +1,321 @@
+"""9.5-code track: code-reachable items only.
+
+- Parse unification: agent_tools / tool_approvals column parsing
+  lives only in agent_def.py; server keeps None (undeclared),
+  worker maps None to [] / {} at the _task_from_row boundary.
+- Goal branch of _apply_resume_preamble is reachable (via
+  _goal_convert_queued on a queued auto-resumed task), so it is
+  kept and tested, not deleted as dead code.
+- Fail-open inventory: audit write failure and preamble lookup
+  failure log a warning without breaking dispatch.
+- M7: shadow endpoint explicit_capability_failures_truncated flag.
+- Policy content fingerprint: stable for the same document,
+  changes when requires change, never touches audit rows.
+
+Run: python3 -m unittest discover -s cloud/tests -p "test_95_code.py" -v
+No real models, no network, no real gateway, no real PG.
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+import cloud.worker as worker_module
+from cloud.agent_def import (
+    parse_agent_tool_approvals_column,
+    parse_agent_tools_column,
+)
+from cloud.attempts import ensure_attempts_table, record_attempt
+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 (
+    _agent_tool_approvals_from_row,
+    _agent_tools_from_row,
+    create_server,
+    init_db,
+)
+from cloud.worker import (
+    Worker,
+    _agent_tool_approvals_from_col,
+    _agent_tools_from_col,
+)
+
+
+class ParseUnificationTestCase(unittest.TestCase):
+    def test_shared_parser_is_single_source(self):
+        # Missing / dirty / wrong-type all parse to None.
+        for raw in (None, "not json", '{"a": 1}', '["a", 1]', 42):
+            if raw == '["a", 1]':
+                # A JSON list parses even if items are not strings;
+                # the parser only guarantees list shape, not items.
+                self.assertEqual(parse_agent_tools_column(raw),
+                                 ["a", 1])
+                continue
+            self.assertIsNone(parse_agent_tools_column(raw))
+        for raw in (None, "not json", '["a"]', '{"a": 1}', 42):
+            if raw == '{"a": 1}':
+                self.assertEqual(
+                    parse_agent_tool_approvals_column(raw), {"a": 1})
+                continue
+            self.assertIsNone(parse_agent_tool_approvals_column(raw))
+        # Valid values are copied, not aliased.
+        src = ["websearch"]
+        self.assertEqual(parse_agent_tools_column(src), src)
+        self.assertIsNot(parse_agent_tools_column(src), src)
+        src = {"websearch": "ask"}
+        self.assertEqual(parse_agent_tool_approvals_column(src), src)
+        self.assertIsNot(parse_agent_tool_approvals_column(src), src)
+
+    def test_server_keeps_none_worker_maps_to_empty(self):
+        # Server serialization: missing/dirty stays None (undeclared).
+        self.assertIsNone(_agent_tools_from_row({}))
+        self.assertIsNone(_agent_tools_from_row({"agent_tools": "junk"}))
+        self.assertEqual(
+            _agent_tools_from_row({"agent_tools": '["websearch"]'}),
+            ["websearch"])
+        self.assertIsNone(_agent_tool_approvals_from_row({}))
+        self.assertIsNone(
+            _agent_tool_approvals_from_row(
+                {"agent_tool_approvals": "junk"}))
+        # Worker dispatch: same parse, None mapped to [] / {}.
+        self.assertEqual(_agent_tools_from_col(None), [])
+        self.assertEqual(_agent_tools_from_col("junk"), [])
+        self.assertEqual(_agent_tools_from_col('["websearch"]'),
+                         ["websearch"])
+        self.assertEqual(_agent_tool_approvals_from_col(None), {})
+        self.assertEqual(_agent_tool_approvals_from_col("junk"), {})
+        self.assertEqual(
+            _agent_tool_approvals_from_col('{"websearch": "ask"}'),
+            {"websearch": "ask"})
+
+
+def _insert_task(db_path, task_id, prompt="做活", status="queued",
+                 project_id="p1", **cols):
+    now = time.time()
+    conn = sqlite3.connect(db_path)
+    try:
+        base = {"id": task_id, "project_id": project_id,
+                "prompt": prompt, "status": status,
+                "idempotency_key": None, "depends_on": "[]",
+                "priority": 0, "created_by": "tester",
+                "created_at": now, "updated_at": now,
+                "last_heartbeat": None}
+        base.update(cols)
+        keys = list(base.keys())
+        conn.execute(
+            "INSERT INTO tasks(%s) VALUES(%s)"
+            % (", ".join(keys), ", ".join("?" * len(keys))),
+            [base[k] for k in keys])
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class WorkerCase(unittest.TestCase):
+    def setUp(self):
+        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):
+        self.tmp.cleanup()
+
+    def _worker(self):
+        def _runner(cmd):
+            return SimpleNamespace(returncode=0, stdout="{}",
+                                   stderr="")
+        return Worker(self.db_path, "/fake/host.py",
+                      runner=_runner, env=self.env)
+
+
+class GoalBranchTestCase(WorkerCase):
+    def test_goal_branch_reached_for_converted_goal_task(self):
+        # Reachability proof (9.5-code): a non-goal task auto-resumed
+        # as platform_fresh (count=1, queued) that is then converted
+        # to a goal while still queued reaches the goal branch:
+        # no _resume (resume_message was cleared by auto_resume),
+        # no _goal_resume_breakpoint, count>0, latest attempt fresh.
+        _insert_task(self.db_path, "t1", prompt="做活",
+                     goal="做目标", goal_mode=1, auto_resume_count=1)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            ensure_attempts_table(conn)
+            record_attempt(conn, "t1", 1, "platform_fresh",
+                           "worker_shutdown", -1,
+                           "做到一半：已建 a.py", "被平台杀掉")
+            conn.commit()
+        finally:
+            conn.close()
+        worker = self._worker()
+        task = worker.get_task("t1")
+        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 = ?",
+                               ("t1",)).fetchone()
+        finally:
+            conn.close()
+        self.assertEqual(row[0], "做目标")
+
+
+class FailOpenLogTestCase(WorkerCase):
+    def test_audit_failure_logs_without_breaking_dispatch(self):
+        worker = self._worker()
+        calls = []
+        with mock.patch.object(
+                worker_module, "AuthStore",
+                side_effect=RuntimeError("db down")), \
+                mock.patch.object(
+                    worker_module, "log_event",
+                    side_effect=lambda *a, **k: calls.append(a[1])):
+            # Must not raise: audit is best-effort by design.
+            worker._audit("p1", "worker_claim", "t1")
+        self.assertEqual(calls, ["worker_audit_failed"])
+
+    def test_preamble_lookup_failure_logs_and_returns_task(self):
+        worker = self._worker()
+        task = {"id": "t1", "project_id": "p1", "prompt": "做活",
+                "auto_resume_count": 1}
+        calls = []
+        with mock.patch.object(
+                worker_module, "_db_connect",
+                side_effect=RuntimeError("db down")), \
+                mock.patch.object(
+                    worker_module, "log_event",
+                    side_effect=lambda *a, **k: calls.append(a[1])):
+            out = worker._apply_resume_preamble(task)
+        self.assertIs(out["prompt"], "做活")
+        self.assertEqual(
+            calls, ["worker_resume_preamble_lookup_failed"])
+
+
+class FingerprintTestCase(unittest.TestCase):
+    def test_stable_and_sensitive_to_requires(self):
+        doc = {"mode": "shadow",
+               "tiers": {"default": "m1"},
+               "model_capabilities": {"m1": {"tool_calling": True}}}
+        a = parse_routing_policy(doc)
+        b = parse_routing_policy(
+            {"model_capabilities": {"m1": {"tool_calling": True}},
+             "tiers": {"default": "m1"}, "mode": "shadow"})
+        self.assertEqual(a.content_fingerprint(),
+                         b.content_fingerprint())
+        c = parse_routing_policy({
+            "mode": "shadow",
+            "tiers": {"default": {"model": "m1",
+                                  "requires": {"tool_calling": True}}},
+            "model_capabilities": {"m1": {"tool_calling": True}}})
+        self.assertNotEqual(a.content_fingerprint(),
+                            c.content_fingerprint())
+        # Operator version string does not affect the fingerprint.
+        d = parse_routing_policy(dict(doc, version="v99"))
+        self.assertEqual(a.content_fingerprint(),
+                         d.content_fingerprint())
+        # Dry-run surfaces it; audit rows never carry it.
+        report = dry_run_policy_document(doc)
+        self.assertEqual(report["content_fingerprint"],
+                         a.content_fingerprint())
+
+
+class _HttpServerCase(unittest.TestCase):
+    def setUp(self):
+        self._old = {k: os.environ.get(k) for k in (
+            "CLOUD_ROUTING_POLICY_JSON", "CLOUD_ROUTING_POLICY_FILE")}
+        os.environ.pop("CLOUD_ROUTING_POLICY_JSON", None)
+        os.environ.pop("CLOUD_ROUTING_POLICY_FILE", None)
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        for k, v in self._old.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _project(self, name="p"):
+        status, project = self._req(
+            "POST", "/projects", {"name": name, "auto_pr": 1},
+            self.owner_key)
+        self.assertEqual(status, 201, project)
+        return project["id"]
+
+
+class ExplicitTruncationTestCase(_HttpServerCase):
+    def test_explicit_failures_truncated_flag(self):
+        pid = self._project()
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.executemany(
+                "INSERT INTO audit_log(user_id, project_id, action,"
+                " detail, created_at) VALUES(?, ?, ?, ?, ?)",
+                [("tester", pid,
+                  "worker_capability_gateway_explicit",
+                  "E1 model=m1 tier=default mode=active"
+                  " missing=tool_calling unknown=False blocked=True",
+                  time.time())] * 10001)
+            conn.commit()
+        finally:
+            conn.close()
+        status, report = self._req(
+            "GET", "/routing-shadow?project_id=" + pid,
+            key=self.owner_key)
+        self.assertEqual(status, 200, report)
+        self.assertTrue(
+            report["explicit_capability_failures_truncated"])
+        self.assertEqual(
+            len(report["explicit_capability_failures"]), 10000)

cloud/worker.py

@@ -137,6 +137,8 @@ from types import SimpleNamespace
 from cloud.agent_def import AgentDefinition as _AgentDefinition
 from cloud.agent_def import format_instructions_header as _agent_header
 from cloud.agent_def import missing_required_tools as _missing_tools
+from cloud.agent_def import parse_agent_tool_approvals_column as _parse_tool_approvals_col
+from cloud.agent_def import parse_agent_tools_column as _parse_tools_col
 from cloud.attempts import build_resume_preamble as _resume_preamble
 from cloud.attempts import ensure_attempts_table as _ensure_attempts
 from cloud.attempts import latest_attempt as _latest_attempt
@@ -362,30 +364,22 @@ def _resolve_model(model=None, config_path=None):
 
 
 def _agent_tools_from_col(raw):
-    """agent_tools 列的容错解析：JSON 数组，非数组/脏数据为 []。"""
-    if raw is None:
-        return []
-    if isinstance(raw, list):
-        return list(raw)
-    try:
-        parsed = json.loads(raw)
-    except (TypeError, ValueError):
-        return []
-    return list(parsed) if isinstance(parsed, list) else []
+    """agent_tools 列的容错解析：唯一实现在
+    agent_def.parse_agent_tools_column（9.5-code 统一）；本
+    函数只把解析层的 None（= 未声明）映射为派发默认值
+    []——worker 每个使用点本来就读 ``or []``，所以映射后
+    的派发行为与统一前逐字一致，缺列/脏数据不炸认领。"""
+    parsed = _parse_tools_col(raw)
+    return parsed if parsed is not None else []
 
 
 def _agent_tool_approvals_from_col(raw):
-    """agent_tool_approvals 列的容错解析：JSON 对象，非对象/
-    脏数据为 {}（= 未声明），不炸认领。"""
-    if raw is None:
-        return {}
-    if isinstance(raw, dict):
-        return dict(raw)
-    try:
-        parsed = json.loads(raw)
-    except (TypeError, ValueError):
-        return {}
-    return dict(parsed) if isinstance(parsed, dict) else {}
+    """agent_tool_approvals 列的容错解析：唯一实现在
+    agent_def.parse_agent_tool_approvals_column；解析层的
+    None（= 未声明）映射为派发默认值 {}，理由同上（每个
+    使用点本来就读 ``or {}``），派发行为与统一前一致。"""
+    parsed = _parse_tool_approvals_col(raw)
+    return parsed if parsed is not None else {}
 
 
 def _task_from_row(row):
@@ -1772,7 +1766,14 @@ class Worker:
             conn.close()
 
     def _audit(self, project_id, action, detail=""):
-        """best-effort 审计：失败不影响任务状态。"""
+        """best-effort 审计：失败不影响任务状态（9.5-code
+        fail-open 清单第 1 项）。
+
+        此处是 fail-open：审计写失败时派发照常进行，运营
+        方只能从这条警告日志看到审计丢了——审计失败绝不
+        炸派发（语义不变），但失败本身不再静默。日志调用
+        本身也加 guard，日志失败时仍不炸派发。
+        """
         try:
             store = AuthStore(self.db_path)
             try:
@@ -1782,8 +1783,14 @@ class Worker:
                     store.conn.close()
                 except Exception:
                     pass
-        except Exception:
-            pass
+        except Exception as exc:
+            try:
+                log_event(self.logger, "worker_audit_failed",
+                          project_id=project_id, status="warning",
+                          message="audit write failed (%s): %s"
+                          % (action, exc))
+            except Exception:
+                pass
 
     def _notify_task_finished(self, task_id, project_id, event_type,
                               error=None):
@@ -2013,8 +2020,8 @@ class Worker:
         是模型最先看到的一段之一；位置是刻意的前置（重
         排说明应当先被看到），不是紧贴原 prompt。goal
         任务若意外走到这里也同步贴 goal（执行器取 goal
-        优先于 prompt）；该分支正常流程不可达（见下），
-        只是防御性保留。查账本失败时原样返回，不炸认领。
+        优先于 prompt）；goal 分支在正常流程可达（见下），
+        不是死代码，绝不能删。查账本失败时原样返回，不炸认领。
         """
         if task.get("coordinator_parts"):
             return task
@@ -2028,7 +2035,20 @@ class Worker:
                 attempt = _latest_attempt(conn, task["id"])
             finally:
                 conn.close()
-        except Exception:
+        except Exception as exc:
+            # 9.5-code fail-open 清单：账本查不到时不贴前言、
+            # 不炸认领（语义不变），但失败记一条警告日志，
+            # 否则“前言丢了”在运行期不可见。
+            try:
+                log_event(self.logger,
+                          "worker_resume_preamble_lookup_failed",
+                          task_id=task["id"],
+                          project_id=task.get("project_id"),
+                          status="warning",
+                          message="resume preamble lookup failed: %s"
+                          % exc)
+            except Exception:
+                pass
             return task
         preamble = _resume_preamble(attempt)
         if not preamble:
@@ -2037,13 +2057,19 @@ class Worker:
         new_task["prompt"] = preamble + "\n\n" + (
             task.get("prompt") or "")
         if task.get("goal_mode") or task.get("verify_command"):
-            # Audit fix 3（M8）：此分支正常流程不可达——
-            # build_resume_preamble 只对 platform_fresh 出
-            # 前言，而 auto_resume 把 goal 任务一律归为
-            # goal_breakpoint；且 goal 断点任务在函数入口
-            # 已被 _goal_resume_breakpoint 挡回。保留不删：
-            # 万一未来有 platform_fresh 的 goal 尝试，goal
-            # 字段不同步贴前言会静默丢说明。
+            # 9.5-code 已证实可达（非 M8 所说的防御性保留）：
+            # platform_fresh 账本只由 auto_resume 给非 goal
+            # 任务记（_record_attempt 唯一调用点），但同一
+            # 任务随后可在 queued 状态下被
+            # server._goal_convert_queued 原地转成 goal
+            # （goal_mode=1，CAS 只改 queued 行，不动
+            # auto_resume_count 与账本）；认领时它既无
+            # _resume（auto_resume 已清 resume_message），
+            # 也无 _goal_resume_breakpoint，于是走到这里。
+            # 删掉本分支会让这类转 goal 任务静默丢前言，
+            # 故保留；复活路径（_goal_revive_terminal）因
+            # resume_message 置 _resume 而在入口被挡，不
+            # 走本分支，二者不要混淆。
             new_task["goal"] = preamble + "\n\n" + (
                 task.get("goal") or task.get("prompt") or "")
         new_task["_resume_preamble_injected"] = True
@@ -5835,6 +5861,14 @@ class Worker:
         也就是写回的来源。
         查找失败（DB 异常）时返回 False 并记警告日志：复核查
         不到依据就跳过，而不是拦任务。
+
+        残留缺口措辞（9.5-code，M1）：fixes3 时期由 shadow /
+        worker_default 决策写回的旧模型既无 persisted 行又
+        无 policy 行，启发式认不出来——这部分存量是固定
+        的、逐任务持续直至完成，不是“会收敛”（收敛一词
+        会误读为随时间自动消失；实际上每个存量任务在完
+        成前每次续跑都不经复核）。新任务不再产生这类写
+        回，所以缺口不扩大，但存量本身不会被重新识别。
         """
         try:
             conn = _db_connect(self.db_path)