From 1b8b6e525d7fb3ae25f8cd65d26fa10af2f2114c Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 06:28:02 +0000 Subject: [PATCH] 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 | 47 ++++++ cloud/models.py | 40 +++++ cloud/routing_dryrun.py | 6 +- cloud/routing_shadow.py | 6 + cloud/server.py | 33 ++-- cloud/tests/test_95_code.py | 321 ++++++++++++++++++++++++++++++++++++ cloud/worker.py | 102 ++++++++---- 7 files changed, 498 insertions(+), 57 deletions(-) create mode 100644 cloud/tests/test_95_code.py diff --git a/cloud/agent_def.py b/cloud/agent_def.py index 0a6f6c9..e0e3019 100644 --- a/cloud/agent_def.py +++ b/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. diff --git a/cloud/models.py b/cloud/models.py index f9af66e..20188b8 100644 --- a/cloud/models.py +++ b/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)) diff --git a/cloud/routing_dryrun.py b/cloud/routing_dryrun.py index e77a074..14a247b 100644 --- a/cloud/routing_dryrun.py +++ b/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, diff --git a/cloud/routing_shadow.py b/cloud/routing_shadow.py index 469af31..eab79b6 100644 --- a/cloud/routing_shadow.py +++ b/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 diff --git a/cloud/server.py b/cloud/server.py index 59c3cbe..69154a3 100644 --- a/cloud/server.py +++ b/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): diff --git a/cloud/tests/test_95_code.py b/cloud/tests/test_95_code.py new file mode 100644 index 0000000..dee0d72 --- /dev/null +++ b/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) diff --git a/cloud/worker.py b/cloud/worker.py index b8a7a31..49ad58c 100644 --- a/cloud/worker.py +++ b/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)