成本配额与文件只读标记:task-cost/quota-status/task-files

This commit is contained in:
host committed 2026-10-30 05:11:06 +08:00
1 parent 2fd73c1121
commit 9d6430a4d0
5 files changed
+711 -14

No files matched your search

+121 -2
View File
@@ -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 时间;标题里的换行压成空格,
+21
View File
@@ -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,
+269
View File
@@ -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()
+248 -12
View File
@@ -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):
+52
View File
@@ -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 的路)。