From 9d6430a4d02d004c6b566899c6c4b59435c2a833 Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 05:06:28 +0800 Subject: [PATCH] =?UTF-8?q?=E6=88=90=E6=9C=AC=E9=85=8D=E9=A2=9D=E4=B8=8E?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E5=8F=AA=E8=AF=BB=E6=A0=87=E8=AE=B0=EF=BC=9A?= =?UTF-8?q?task-cost/quota-status/task-files?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cloud/marker_tools.py | 123 ++++++++++- cloud/sandbox.py | 21 ++ cloud/tests/test_cost_files_markers.py | 269 +++++++++++++++++++++++++ cloud/worker.py | 260 ++++++++++++++++++++++-- docs/marker-tools.md | 52 +++++ 5 files changed, 711 insertions(+), 14 deletions(-) create mode 100644 cloud/tests/test_cost_files_markers.py diff --git a/cloud/marker_tools.py b/cloud/marker_tools.py index 72ff897..5abe246 100644 --- a/cloud/marker_tools.py +++ b/cloud/marker_tools.py @@ -1,6 +1,6 @@ """任务内标记工具（marker tools）：task-create / task-poll / task-read / wait-threads / task-cancel / task-retry / todo-list / todo-create / -todo-update / task-list。 +todo-update / task-list / task-cost / quota-status / task-files。 设计（与 peer_messages、task_spawn 同款原则，不另起一套）： @@ -56,11 +56,24 @@ todo-update / task-list。 任务 id 时也能分诊——查运行中与失败的任务、捡失败任务 重试。不传 status 时查全部状态并附各状态计数；只读， 跨项目任务不出现，报告最多列 limit 条（默认 20，上限 50）。 +- task-cost / quota-status：只读成本与配额。task-cost 查单个 + 任务的成本（美元只计有价行，无价 token 单列）；quota-status + 查发送方任务创建人当月的配额、已用与无价 token——派生子 + 任务前先看配额，避免配额门在执行时才拦住。两者都不接受 + 查别人的参数：quota 永远按发送方创建人算，task-cost 只 + 放行派生链内或同创建人的目标（比服务端只要求 read 更严， + 防止在项目内猜 id 看别人的花费）。 +- task-files：读派生链内或同创建人任务的 worktree 文件。 + 不传 path 时列顶层，传相对路径读文件（上限 20000 字， + 二进制文件拒绝）；绝对路径、.. 逃逸、符号链接逃逸一律 + 拒绝；cloud 任务与 worktree 在别机的任务拒绝（文件不在 + 本机）；目标没有本地 worktree 时按 404 报告，不编内容。 上限口径（本模块最终定的值，文档与代码一致）： - 每轮创建、轮询、取消、重试各最多 5 块；todo-list、 - todo-create、todo-update、task-list 各最多 5 块；wait + todo-create、todo-update、task-list、task-cost、 + quota-status、task-files 各最多 5 块；wait 每轮最多 2 块，每块最多 8 个目标。8 这个数是刻意与服务端 POST /tasks/{id}/wait 的“一次最多 8 个线程”对齐的； spawn 分支原先的每块 10 个在此收紧到 8 个，不要再用 10。 @@ -94,6 +107,7 @@ import。DB 助手函数只用 ? 占位符，与 peer_messages 同款写法。 """ import json +import os import re import time import uuid @@ -131,6 +145,12 @@ TASK_STATUSES = ("queued", "running", "completed", "failed", MAX_TASK_LISTS_PER_TURN = 5 DEFAULT_TASK_LIST_LIMIT = 20 MAX_TASK_LIST_LIMIT = 50 +MAX_TASK_COSTS_PER_TURN = 5 +MAX_QUOTA_PER_TURN = 5 +MAX_TASK_FILES_PER_TURN = 5 +MAX_TASK_FILE_CHARS = 20000 +MAX_TASK_FILE_LIST = 50 +MAX_TASK_FILE_PATH_CHARS = 1000 _CREATE_RE = re.compile(r"```task-create\s*\n(.*?)\n```", re.DOTALL) _POLL_RE = re.compile( @@ -150,6 +170,12 @@ _TODO_UPDATE_RE = re.compile( r"```(todo-update|todo_update)\s*\n(.*?)\n```", re.DOTALL) _TASK_LIST_RE = re.compile( r"```(task-list|task_list)\s*\n(.*?)\n```", re.DOTALL) +_TASK_COST_RE = re.compile( + r"```(task-cost|task_cost)\s*\n(.*?)\n```", re.DOTALL) +_QUOTA_RE = re.compile( + r"```(quota-status|quota_status)\s*\n(.*?)\n```", re.DOTALL) +_TASK_FILES_RE = re.compile( + r"```(task-files|task_files)\s*\n(.*?)\n```", re.DOTALL) def row_get(row, name): @@ -434,6 +460,83 @@ def extract_task_lists(text): return out +def extract_task_costs(text): + """提取 task-cost 块：{\"task_id\": \"...\"}，与 task-cancel + 同款别名口径（task_id / to / to_task_id / thread_id）； + 每轮最多 5 块。""" + return _extract_controls(text, _TASK_COST_RE) + + +def extract_quota_status(text): + """提取 quota-status 块：块体必须是 JSON 对象（可为空对象）， + 不接受任何参数——配额永远按发送方任务的创建人算，块里写 + 别人 id 也无效，避免用标记块查别人的配额；每轮最多 5 块 + （查询便宜，但多块只是重复报告）。""" + if not text: + return [] + out = [] + for match in _QUOTA_RE.finditer(str(text)): + if len(out) >= MAX_QUOTA_PER_TURN: + break + try: + payload = json.loads(match.group(2)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + out.append({}) + return out + + +def extract_task_files(text): + """提取 task-files 块：task_id 必填，path 可选（默认列顶层）； + path 非字符串或超长时整块跳过；绝对路径与 .. 逃逸不在此 + 层跳过——worker 的 safe_worktree_path 会拒绝并在报告里写 + 清原因，让模型知道错在哪；每轮最多 5 块。""" + if not text: + return [] + out = [] + for match in _TASK_FILES_RE.finditer(str(text)): + if len(out) >= MAX_TASK_FILES_PER_TURN: + break + try: + payload = json.loads(match.group(2)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + task_id = payload.get("task_id") + if not isinstance(task_id, str) or not task_id.strip(): + continue + path = payload.get("path", "") + if path is None: + path = "" + if not isinstance(path, str): + continue + if len(path) > MAX_TASK_FILE_PATH_CHARS: + continue + out.append({"task_id": task_id.strip(), "path": path}) + return out + + +def safe_worktree_path(worktree, rel): + """worktree 内相对路径的安全解析：与服务端 + _safe_worktree_path 同款——绝对路径、.. 逃逸、符号链接 + 逃逸一律返回 None；空串返回 worktree 本身。""" + if not rel: + return worktree + if os.path.isabs(rel): + return None + base = os.path.realpath(worktree) + target = os.path.realpath(os.path.join(base, rel)) + try: + if os.path.commonpath([base, target]) != base: + return None + except ValueError: + return None + return target + + _SPAWN_SUBJECT_ZH = ("新建的任务", "派生任务", "新建任务", "子任务", "子代理", "新任务", "新会话") _SPAWN_SUBJECT_EN = ("child tasks", "spawned tasks", "new tasks", @@ -895,6 +998,22 @@ def marker_report_header(): return "[任务工具报告]" +def format_cost(task_id, report): + """成本单行：美元数只覆盖有价行，无价 actual token 单独列， + 不会被静默算成 0 美元（与 CostStore._report 同款口径）。""" + usd = (report.get("usd_micros") or 0) / 1000000.0 + line = ("任务 %s 成本：$%.4f，总 token %d（actual %d，" + "estimated %d），轮次 %d" + % (task_id, usd, report.get("total_tokens", 0), + report.get("actual_tokens", 0), + report.get("estimated_tokens", 0), + report.get("turns", 0))) + unpriced = report.get("unpriced_actual_tokens", 0) + if unpriced: + line += "，其中无价 actual token %d 未计入美元" % unpriced + return line + + def format_todo(todo, max_title=100): """待办单行（报告与首轮注入共用）：状态 + 标题（截断）+ id + 优先级，due_at 有值时附 UTC 时间；标题里的换行压成空格， diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 0157522..3c96dc8 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -22,8 +22,11 @@ import subprocess import sys import time +from cloud.marker_tools import extract_quota_status as _extract_quota_status from cloud.marker_tools import extract_task_cancels as _extract_task_cancels +from cloud.marker_tools import extract_task_costs as _extract_task_costs from cloud.marker_tools import extract_task_creates as _extract_task_creates +from cloud.marker_tools import extract_task_files as _extract_task_files from cloud.marker_tools import extract_task_lists as _extract_task_lists from cloud.marker_tools import extract_task_polls as _extract_task_polls from cloud.marker_tools import extract_task_retries as _extract_task_retries @@ -409,6 +412,9 @@ class LocalExecutor(Executor): task_cancels = [] task_retries = [] task_lists = [] + task_costs = [] + quota_status = [] + task_files = [] todo_lists = [] todo_creates = [] todo_updates = [] @@ -453,6 +459,9 @@ class LocalExecutor(Executor): task_cancels = _extract_task_cancels(parsed.get("text")) task_retries = _extract_task_retries(parsed.get("text")) task_lists = _extract_task_lists(parsed.get("text")) + task_costs = _extract_task_costs(parsed.get("text")) + quota_status = _extract_quota_status(parsed.get("text")) + task_files = _extract_task_files(parsed.get("text")) todo_lists = _extract_todo_lists(parsed.get("text")) todo_creates = _extract_todo_creates(parsed.get("text")) todo_updates = _extract_todo_updates(parsed.get("text")) @@ -481,6 +490,9 @@ class LocalExecutor(Executor): "task_cancels": task_cancels, "task_retries": task_retries, "task_lists": task_lists, + "task_costs": task_costs, + "quota_status": quota_status, + "task_files": task_files, "todo_lists": todo_lists, "todo_creates": todo_creates, "todo_updates": todo_updates, @@ -1926,6 +1938,9 @@ def _parse_host_result(task_id, fake): task_cancels = [] task_retries = [] task_lists = [] + task_costs = [] + quota_status = [] + task_files = [] todo_lists = [] todo_creates = [] todo_updates = [] @@ -1957,6 +1972,9 @@ def _parse_host_result(task_id, fake): task_cancels = _extract_task_cancels(parsed.get("text")) task_retries = _extract_task_retries(parsed.get("text")) task_lists = _extract_task_lists(parsed.get("text")) + task_costs = _extract_task_costs(parsed.get("text")) + quota_status = _extract_quota_status(parsed.get("text")) + task_files = _extract_task_files(parsed.get("text")) todo_lists = _extract_todo_lists(parsed.get("text")) todo_creates = _extract_todo_creates(parsed.get("text")) todo_updates = _extract_todo_updates(parsed.get("text")) @@ -1982,6 +2000,9 @@ def _parse_host_result(task_id, fake): "task_cancels": task_cancels, "task_retries": task_retries, "task_lists": task_lists, + "task_costs": task_costs, + "quota_status": quota_status, + "task_files": task_files, "todo_lists": todo_lists, "todo_creates": todo_creates, "todo_updates": todo_updates, diff --git a/cloud/tests/test_cost_files_markers.py b/cloud/tests/test_cost_files_markers.py new file mode 100644 index 0000000..35c48fe --- /dev/null +++ b/cloud/tests/test_cost_files_markers.py @@ -0,0 +1,269 @@ +"""成本/配额/文件标记单测：task-cost / quota-status / task-files +解析、sandbox 提取、worker 投递与 worktree 解析。 + +不跑真实模型、不联网：解析是纯函数；worker 投递用 init_db +建的临时 SQLite 库，直接调 _deliver_marker_tools，不走 +run_one 全量流程；文件读取用临时目录冒充 worktree（测 +试里直接给 Worker 换 tenants_root 与 host_py_path）。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.cost import CostStore + from cloud.marker_tools import ( + extract_quota_status, + extract_task_costs, + extract_task_files, + format_cost, + safe_worktree_path, + ) + from cloud.sandbox import LocalExecutor, _parse_host_result + from cloud.server import init_db + from cloud.worker import Worker +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.cost import CostStore + from cloud.marker_tools import ( + extract_quota_status, + extract_task_costs, + extract_task_files, + format_cost, + safe_worktree_path, + ) + from cloud.sandbox import LocalExecutor, _parse_host_result + from cloud.server import init_db + from cloud.worker import Worker + +FAKE_ENV = {"XCM_API_KEY": "test-placeholder"} + + +class ExtractTestCase(unittest.TestCase): + def test_cost_and_files(self): + self.assertEqual(extract_task_costs( + "```task-cost\n{\"task_id\": \"t1\"}\n```"), + [{"task_id": "t1"}]) + self.assertEqual(extract_task_costs( + "```task-cost\n{\"to\": \"t1\"}\n```"), + [{"task_id": "t1"}]) + self.assertEqual(extract_task_costs( + "```task-cost\n{}\n```"), []) + self.assertEqual(extract_task_files( + "```task-files\n{\"task_id\": \"t1\"}\n```"), + [{"task_id": "t1", "path": ""}]) + self.assertEqual(extract_task_files( + "```task-files\n{\"task_id\": \"t1\", \"path\": \"a.txt\"}\n```"), + [{"task_id": "t1", "path": "a.txt"}]) + self.assertEqual(extract_task_files( + "```task-files\n{\"path\": \"a.txt\"}\n```"), []) + + def test_quota_only_object(self): + self.assertEqual(extract_quota_status( + "```quota-status\n{}\n```"), [{}]) + self.assertEqual(extract_quota_status( + "```quota-status\n{\"owner\": \"someone\"}\n```"), [{}]) + self.assertEqual(extract_quota_status( + "```quota-status\n[]\n```"), []) + + def test_safe_worktree_path(self): + self.assertIsNone(safe_worktree_path("/tmp/w", "../etc/passwd")) + self.assertIsNone(safe_worktree_path("/tmp/w", "/etc/passwd")) + self.assertEqual(safe_worktree_path("/tmp/w", ""), + "/tmp/w") + self.assertEqual(safe_worktree_path("/tmp/w", "a/b.txt"), + "/tmp/w/a/b.txt") + + def test_format_cost_unpriced_separate(self): + line = format_cost("t1", {"usd_micros": 0, "total_tokens": 10, + "actual_tokens": 10, + "estimated_tokens": 0, "turns": 1, + "unpriced_actual_tokens": 10}) + self.assertIn("$0.0000", line) + self.assertIn("无价 actual token 10 未计入美元", line) + + def test_sandbox_extracts_both_executors(self): + text = ("```task-cost\n{\"task_id\": \"t1\"}\n```\n" + "```quota-status\n{}\n```\n" + "```task-files\n{\"task_id\": \"t1\"}\n```") + fake = SimpleNamespace( + returncode=0, + stdout=json.dumps({"status": "completed", "text": text}), + stderr="") + local = LocalExecutor("/fake/host.py")._result_from_runner("t1", fake) + self.assertEqual(len(local["task_costs"]), 1) + self.assertEqual(len(local["quota_status"]), 1) + self.assertEqual(len(local["task_files"]), 1) + docker = _parse_host_result("t1", fake) + self.assertEqual(len(docker["task_costs"]), 1) + self.assertEqual(len(docker["quota_status"]), 1) + self.assertEqual(len(docker["task_files"]), 1) + + +class WorkerCostFilesTestCase(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._insert("PM", status="running", created_by="tester") + self._member("tester", "maintainer") + + def tearDown(self): + self.tmp.cleanup() + + def _insert(self, task_id, status="queued", created_by="tester", + project_id="p1", prompt="do work", parent=None, + **cols): + now = time.time() + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat)" + " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)", + (task_id, project_id, prompt, status, created_by, + now, now)) + if parent is not None: + conn.execute( + "UPDATE tasks SET parent_task_id = ? WHERE id = ?", + (parent, task_id)) + for key, value in cols.items(): + conn.execute( + "UPDATE tasks SET %s = ? WHERE id = ?" % key, + (value, task_id)) + conn.commit() + finally: + conn.close() + + def _member(self, user_id, role): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS users(id TEXT PRIMARY KEY," + " username TEXT UNIQUE, api_key_hash TEXT," + " created_at REAL)") + conn.execute( + "CREATE TABLE IF NOT EXISTS project_members(" + "project_id TEXT, user_id TEXT, role TEXT," + " PRIMARY KEY(project_id, user_id))") + conn.execute( + "INSERT OR REPLACE INTO project_members" + "(project_id, user_id, role) VALUES('p1', ?, ?)", + (user_id, role)) + conn.execute( + "INSERT OR IGNORE INTO users(id, username, api_key_hash," + " created_at) VALUES(?, ?, '', 0)", (user_id, user_id)) + conn.commit() + finally: + conn.close() + + def _worker(self): + return Worker(self.db_path, "/fake/host.py", env=FAKE_ENV) + + def _followup_bodies(self, task_id="PM"): + conn = sqlite3.connect(self.db_path) + try: + rows = conn.execute( + "SELECT body FROM task_followups" + " WHERE task_id = ?", (task_id,)).fetchall() + return [row[0] for row in rows] + finally: + conn.close() + + def test_deliver_cost_and_quota(self): + self._insert("E1", status="completed", parent="PM") + store = CostStore(self.db_path) + try: + store.record("E1", "p1", "gpt-test", 100, 50, + usd_micros=12345, source="actual") + finally: + try: + store._conn.close() + except Exception: + pass + worker = self._worker() + worker._deliver_marker_tools(worker.get_task("PM"), { + "status": "completed", + "task_costs": [{"task_id": "E1"}], + "quota_status": [{}], + "task_files": []}) + bodies = self._followup_bodies() + self.assertEqual(len(bodies), 1) + self.assertIn("$0.0123", bodies[0]) + self.assertIn("配额", bodies[0]) + + def test_deliver_cost_other_creator_rejected(self): + self._insert("E1", status="completed", created_by="someone-else") + worker = self._worker() + worker._deliver_marker_tools(worker.get_task("PM"), { + "status": "completed", + "task_costs": [{"task_id": "E1"}], + "quota_status": [], + "task_files": []}) + bodies = self._followup_bodies() + self.assertIn("不在派生链内且创建人不同", bodies[0]) + + def test_deliver_files_list_and_read(self): + self._insert("E1", status="completed", parent="PM") + worker = self._worker() + # 冒充 worktree：租户路径下给 E1 建目录与文件。 + worktree = os.path.join(self.tmp.name, "tenants", "p1", + "work", "host-worktrees", "E1") + os.makedirs(worktree) + with open(os.path.join(worktree, "out.txt"), "w") as fh: + fh.write("hello-files") + with open(os.path.join(worktree, "bin.dat"), "wb") as fh: + fh.write(b"\x00\x01\x02") + worker.tenants_root = os.path.join(self.tmp.name, "tenants") + worker._deliver_marker_tools(worker.get_task("PM"), { + "status": "completed", + "task_costs": [], + "quota_status": [], + "task_files": [{"task_id": "E1", "path": ""}, + {"task_id": "E1", "path": "out.txt"}, + {"task_id": "E1", "path": "bin.dat"}, + {"task_id": "E1", "path": "../escape"}]}) + bodies = self._followup_bodies() + self.assertEqual(len(bodies), 1) + self.assertIn("out.txt", bodies[0]) + self.assertIn("hello-files", bodies[0]) + self.assertIn("二进制文件", bodies[0]) + self.assertIn("路径逃逸", bodies[0]) + + def test_deliver_files_cloud_and_no_worktree_rejected(self): + self._insert("C1", status="completed", parent="PM", + executor="cloud") + self._insert("N1", status="completed", parent="PM") + worker = self._worker() + worker.tenants_root = os.path.join(self.tmp.name, "tenants") + worker._deliver_marker_tools(worker.get_task("PM"), { + "status": "completed", + "task_costs": [], + "quota_status": [], + "task_files": [{"task_id": "C1", "path": ""}, + {"task_id": "N1", "path": ""}]}) + bodies = self._followup_bodies() + self.assertIn("不在本机", bodies[0]) + self.assertIn("没有本地 worktree", bodies[0]) + + def test_deliver_goal_sender_ignored(self): + self._insert("G1", status="running", goal_mode=1) + worker = self._worker() + worker._deliver_marker_tools(worker.get_task("G1"), { + "status": "completed", + "task_costs": [{"task_id": "G1"}], + "quota_status": [{}], + "task_files": []}) + self.assertEqual(self._followup_bodies("G1"), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index e990838..1c7dea1 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -154,8 +154,10 @@ from cloud.marker_tools import check_control_gate as _marker_gate from cloud.marker_tools import count_direct_spawn_children as _marker_count_direct from cloud.marker_tools import fetch_task_snapshot as _marker_snapshot from cloud.marker_tools import format_snapshot as _marker_format +from cloud.marker_tools import format_cost as _format_cost from cloud.marker_tools import format_task as _format_task from cloud.marker_tools import format_todo as _format_todo +from cloud.marker_tools import safe_worktree_path as _safe_worktree_path from cloud.marker_tools import marker_report_header as _marker_header from cloud.todos import create_todo as _todo_create_db from cloud.todos import get_todo as _todo_get_db @@ -1804,7 +1806,11 @@ class Worker: "```task-list", '{"status": "failed"}', "```", - "（不传 status 时查全部状态并附各状态计数。）"] + "（不传 status 时查全部状态并附各状态计数。）", + "成本、配额、子任务文件同款标记块：task-cost" + " {\"task_id\": \"...\"}、quota-status {}、" + "task-files {\"task_id\": \"...\", \"path\": \"\"}" + "（path 空串列顶层；只读派生链内或同创建人的任务）。"] prefix = "\n".join(lines) new_task = dict(task) new_task["prompt"] = prefix + "\n\n" + (task.get("prompt") or "") @@ -3366,19 +3372,57 @@ class Worker: return None, "权限检查失败：%s" % exc return bool(ok), "" + def _task_files_worktree(self, target): + """task-files 用的目标 worktree 解析，返回 (path, None) + 或 (None, reason)。 + + 候选顺序与服务端 _task_worktree_for_files 同款：租户 + Docker 任务走 tenants_root/<安全项目名>/work/ + host-worktrees/<安全任务名>，本地任务走 host.py 同级 + work/host-worktrees/<安全任务名>；只接受已存在的目录， + 不存在时按“没有本地 worktree”报告，绝不把共享挂载点 + 本身或默认仓库路径当成目标任务的目录（那才是错指的 + 来源）。cloud 任务与 worktree 在别机的任务直接拒绝。""" + if target.get("executor") == "cloud": + return None, "cloud 任务的 worktree 不在本机" + worktree_host = target.get("worktree_host") + if worktree_host and worktree_host != socket.gethostname(): + return None, ("任务 worktree 在主机 %s，不在本机" + % worktree_host) + safe_task = re.sub(r"[^A-Za-z0-9_.-]", "-", + str(target["id"])) + try: + safe_project = self._tenant_safe_name( + target.get("project_id")) + except Exception: + safe_project = None + if safe_project: + tenant_path = os.path.join( + self.tenants_root, safe_project, "work", + "host-worktrees", safe_task) + if os.path.isdir(tenant_path): + return tenant_path, None + host_dir = os.path.dirname( + os.path.abspath(self.host_py_path)) + local_path = os.path.join(host_dir, "work", + "host-worktrees", safe_task) + if os.path.isdir(local_path): + return local_path, None + return None, "目标没有本地 worktree（未启动或目录已清理）" + def _deliver_marker_tools(self, task, result): """执行本轮 agent 输出里的标记工具块（见 marker_tools）。 只执行 completed 轮次的块；goal 任务与协调员任务不解析 （只收不发，与 peer 同款）；父任务被取消时不执行。 执行顺序：创建 → 取消 → 重试 → 待办创建 → 待办更新 → - 待办列表 → 任务列表 → 轮询 → 等待——创建在前让同一轮 - 刚创建的任务立刻能被轮询，控制操作在快照之前让轮询/ - 等待读到取消或重试后的新状态；待办列表放在待办写操作 - 之后，让同一轮刚建/刚改的待办立刻能在列表里看到。 - 全部报告拼成一条系统后续消息排给父任务自己，由排空 - 循环在轮次边界跑下一轮；拒绝只记报告与审计，不炸 - run_one。""" + 待办列表 → 任务列表 → 任务成本 → 配额 → 任务文件 → + 轮询 → 等待——创建在前让同一轮刚创建的任务立刻能被 + 轮询，控制操作在快照之前让轮询/等待读到取消或重试后 + 的新状态；待办列表放在待办写操作之后，让同一轮刚建/ + 刚改的待办立刻能在列表里看到。全部报告拼成一条系统 + 后续消息排给父任务自己，由排空循环在轮次边界跑下 + 一轮；拒绝只记报告与审计，不炸 run_one。""" if not result or result.get("status") != "completed": return if self._was_cancelled(task["id"]): @@ -3396,9 +3440,13 @@ class Worker: todo_creates = result.get("todo_creates") or [] todo_updates = result.get("todo_updates") or [] task_lists = result.get("task_lists") or [] + task_costs = result.get("task_costs") or [] + quota_status = result.get("quota_status") or [] + task_files = result.get("task_files") or [] if not (creates or polls or waits or cancels or retries or todo_lists or todo_creates or todo_updates - or task_lists): + or task_lists or task_costs or quota_status + or task_files): return reports = [] created = 0 @@ -3581,7 +3629,8 @@ class Worker: # 一律拒绝，不静默放宽。跨项目待办按“不存在”拒绝； # 关联任务 id 跨项目时同样拒绝，避免把别的项目的任务 # 挂到本项目待办上。 - if todo_lists or todo_creates or todo_updates or task_lists: + if (todo_lists or todo_creates or todo_updates or task_lists + or task_costs or quota_status or task_files): read_ok, _read_reason = self._marker_auth(task, "read") write_ok, _write_reason = self._marker_auth( task, "create_task") @@ -3747,6 +3796,190 @@ class Worker: lines.append("（还有 %d 条未列出。）" % (len(tasks) - block["limit"])) reports.append("\n".join(lines)) + for block in task_costs: + if read_ok is not True: + reports.append("查任务 %s 成本：拒绝（发送方" + "创建人无 read 权限）。" + % block["task_id"]) + rejected += 1 + continue + target = self.get_task(block["task_id"]) + if (target is None + or target.get("project_id") + != task["project_id"]): + reports.append("查任务 %s 成本：不存在（或" + "属于别的项目），已跳过。" + % block["task_id"]) + rejected += 1 + continue + try: + conn = _db_connect(self.db_path) + try: + relation = _marker_ancestry( + conn, task, target) + finally: + conn.close() + except Exception: + relation = "other" + if relation not in ("descendant", "same_creator"): + reports.append("查任务 %s 成本：拒绝（目标不" + "在派生链内且创建人不同）。" + % block["task_id"]) + rejected += 1 + continue + try: + store = CostStore(self.db_path) + try: + report = store.report_by_task( + block["task_id"]) + finally: + try: + store._conn.close() + except Exception: + pass + reports.append(_format_cost(block["task_id"], + report)) + except Exception as exc: + reports.append("查任务 %s 成本失败：%s" + % (block["task_id"], exc)) + rejected += 1 + for _block in quota_status: + if read_ok is not True: + reports.append("查配额：拒绝（发送方创建人" + "无 read 权限）。") + rejected += 1 + continue + owner = task.get("created_by") + if not owner and task.get("parent_task_id"): + parent = self.get_task(task["parent_task_id"]) + if parent is not None: + owner = parent.get("created_by") + if not owner: + reports.append("查配额：拒绝（发送方无创建" + "人，无法确定配额属主）。") + rejected += 1 + continue + try: + quota, used, month, unpriced = ( + self._quota_limit_and_used(owner)) + except Exception as exc: + reports.append("查配额失败：%s" % exc) + rejected += 1 + continue + if quota is None: + reports.append("配额（%s）：不限额，已用" + " $%.4f，无价 actual token %d。" + % (month, used / 1000000.0, + unpriced)) + else: + reports.append("配额（%s）：总额 $%.4f，已用" + " $%.4f，剩余 $%.4f，无价" + " actual token %d。" + % (month, quota / 1000000.0, + used / 1000000.0, + (quota - used) / 1000000.0, + unpriced)) + for block in task_files: + if read_ok is not True: + reports.append("读任务 %s 文件：拒绝（发送方" + "创建人无 read 权限）。" + % block["task_id"]) + rejected += 1 + continue + target = self.get_task(block["task_id"]) + if (target is None + or target.get("project_id") + != task["project_id"]): + reports.append("读任务 %s 文件：不存在（或" + "属于别的项目），已跳过。" + % block["task_id"]) + rejected += 1 + continue + try: + conn = _db_connect(self.db_path) + try: + relation = _marker_ancestry( + conn, task, target) + finally: + conn.close() + except Exception: + relation = "other" + if relation not in ("descendant", "same_creator"): + reports.append("读任务 %s 文件：拒绝（目标不" + "在派生链内且创建人不同）。" + % block["task_id"]) + rejected += 1 + continue + worktree, reason = self._task_files_worktree(target) + if worktree is None: + reports.append("读任务 %s 文件：拒绝（%s）。" + % (block["task_id"], reason)) + rejected += 1 + continue + safe = _safe_worktree_path(worktree, block["path"]) + if safe is None: + reports.append("读任务 %s 文件 %s：拒绝" + "（路径逃逸出 worktree）。" + % (block["task_id"], + block["path"] or "（顶层）")) + rejected += 1 + continue + if os.path.isdir(safe): + try: + names = sorted(os.listdir(safe))[:50] + except Exception as exc: + reports.append("列任务 %s 目录失败：%s" + % (block["task_id"], exc)) + rejected += 1 + continue + lines = ["任务 %s 目录 %s（共 %d 项，" + "只列前 50 项）：" + % (block["task_id"], + block["path"] or "（顶层）", + len(names))] + for name in names: + full = os.path.join(safe, name) + lines.append(name + ("/" if os.path.isdir( + full) else "")) + reports.append("\n".join(lines)) + elif os.path.isfile(safe): + try: + with open(safe, "rb") as fh: + raw = fh.read(20001) + except Exception as exc: + reports.append("读任务 %s 文件 %s 失败：" + "%s" % (block["task_id"], + block["path"], exc)) + rejected += 1 + continue + try: + text = raw.decode("utf-8") + except UnicodeDecodeError: + text = None + if text is None or "\x00" in text: + # NUL 字节是二进制文件的可靠信号：纯控制 + # 字符也能通过 utf-8 解码，只靠解码失败 + # 拦不住，必须另查 NUL。 + reports.append("读任务 %s 文件 %s：拒绝" + "（二进制文件，不读）。" + % (block["task_id"], + block["path"])) + rejected += 1 + continue + truncated = len(raw) > 20000 + if truncated: + text = text[:20000] + reports.append("任务 %s 文件 %s%s：\n%s" + % (block["task_id"], + block["path"], + "（已截断前 20000 字）" + if truncated else "", + text)) + else: + reports.append("读任务 %s 文件 %s：不存在。" + % (block["task_id"], + block["path"])) + rejected += 1 for poll in polls: try: reports.append(self._poll_task_snapshot(task, poll)) @@ -3763,11 +3996,14 @@ class Worker: self._queue_self_followup(task, body) detail = ("%s: spawned=%d rejected=%d polls=%d waits=%d " "cancels=%d retries=%d todo_lists=%d " - "todo_creates=%d todo_updates=%d task_lists=%d" + "todo_creates=%d todo_updates=%d task_lists=%d " + "task_costs=%d quota=%d task_files=%d" % (task["id"], created, rejected, len(polls), len(waits), len(cancels), len(retries), len(todo_lists), len(todo_creates), - len(todo_updates), len(task_lists))) + len(todo_updates), len(task_lists), + len(task_costs), len(quota_status), + len(task_files))) self._audit(task["project_id"], "worker_marker_tools", detail) def _deliver_peer_messages(self, task, result): diff --git a/docs/marker-tools.md b/docs/marker-tools.md index ad9b121..52f2ae7 100644 --- a/docs/marker-tools.md +++ b/docs/marker-tools.md @@ -300,3 +300,55 @@ Docker 执行器都经 sandbox 的结果解析提取标记块（与 task-message - 贴的是数据加一行格式示例，不是完整工具指南；完整格式 只在本文件里。注入顺序：项目上下文在最前，其次待办、 记忆，最后是原 prompt。 + +## 成本、配额与文件标记（task-cost / quota-status / task-files） + +### task-cost：查单个任务成本（只读） + +```text +```task-cost +{"task_id": "<目标任务 id>"} +``` +``` + +- 报告：美元数（只计有价行）、总 token（actual / estimated + 分列）、轮次；无价 actual token 单独列，不会被静默算成 + 0 美元（与 CostStore._report 同款口径）。 +- 门禁比服务端只要求 read 更严：目标须在本项目内，且在 + 派生链内或与发送方同创建人（自己也算同创建人），防止在 + 项目内猜 id 看别人的花费；每轮最多 5 块。 + +### quota-status：查发送方创建人当月配额（只读） + +```text +```quota-status +{} +``` +``` + +- 块体必须是 JSON 对象，不接受任何参数：配额永远按发送方 + 任务的创建人算（创建人缺失时回查父任务，仍查不到时拒绝）， + 块里写别人 id 也无效。报告：总额、已用、剩余、无价 + actual token；显式不限额时明写不限额，不编一个数字。 +- 每轮最多 5 块；发送方创建人须有项目 read 权限，否则拒绝。 + +### task-files：读派生链内或同创建人任务的文件（只读） + +```text +```task-files +{"task_id": "<目标任务 id>", "path": "out.txt"} +``` +``` + +- 不传 path（或空串）时列顶层（只列前 50 项，目录名带 /）； + 传相对路径读文件，上限 20000 字，超出写明截断。 +- 门禁：目标须在本项目内且在派生链内或同创建人；cloud + 任务与 worktree 在别机的任务拒绝（文件不在本机）；目标 + 没有本地 worktree 时按“没有本地 worktree”报告，不编内容。 +- 路径安全：绝对路径、.. 逃逸、符号链接逃逸一律拒绝并在 + 报告里写清；二进制文件（解码失败或含 NUL 字节）拒绝； + worktree 解析只接受已存在的任务级目录（租户路径或 + host.py 同级 work/host-worktrees），绝不把共享挂载点本身 + 当成目标任务的目录。 +- 每轮最多 5 块；删除、写文件不走标记块（模型改文件走它 + 自己的执行器工作区，不是读别人 worktree 的路）。