diff --git a/cloud/marker_tools.py b/cloud/marker_tools.py index 904f7b2..0b1afd9 100644 --- a/cloud/marker_tools.py +++ b/cloud/marker_tools.py @@ -1,5 +1,6 @@ """任务内标记工具（marker tools）：task-create / task-poll / task-read / -wait-threads / task-cancel / task-retry。 +wait-threads / task-cancel / task-retry / todo-list / todo-create / +todo-update。 设计（与 peer_messages、task_spawn 同款原则，不另起一套）： @@ -44,10 +45,18 @@ wait-threads / task-cancel / task-retry。 （仅 failed / interrupted / merge-failed 可重试，重试时 清空上一轮 result_text / error_text / actual_model、状态 回 queued、prompt 与模型等原样保留）。 +- todo-list / todo-create / todo-update：项目待办的读与写， + 直接复用 todos 模块的校验与状态机（completed 记 + completed_at、离开 completed 清掉）。todo-list 不传 status + 时只返回未完成项（pending / in_progress / blocked）， + 查已完成项须显式传 status；写操作只在本项目内生效， + 跨项目待办一律按“不存在”拒绝；删除不走标记块（硬删除 + 只在控制台 / REST DELETE 里做，避免模型误删整条）。 上限口径（本模块最终定的值，文档与代码一致）： -- 每轮创建、轮询、取消、重试各最多 5 块；wait 每轮最多 +- 每轮创建、轮询、取消、重试各最多 5 块；todo-list、 + todo-create、todo-update 各最多 5 块；wait 每轮最多 2 块，每块最多 8 个目标。8 这个数是刻意与服务端 POST /tasks/{id}/wait 的“一次最多 8 个线程”对齐的； spawn 分支原先的每块 10 个在此收紧到 8 个，不要再用 10。 @@ -104,6 +113,15 @@ TERMINAL_STATUSES = ("completed", "failed", "interrupted", "merged", RETRYABLE_STATUSES = ("failed", "interrupted", "merge-failed") EFFORTS = ("low", "medium", "high", "xhigh", "max") EXECUTORS = ("local", "docker", "cloud") +TODO_STATUSES = ("pending", "in_progress", "blocked", "completed", + "cancelled") +TODO_PRIORITIES = ("low", "medium", "high") +MAX_TODO_LISTS_PER_TURN = 5 +MAX_TODO_CREATES_PER_TURN = 5 +MAX_TODO_UPDATES_PER_TURN = 5 +MAX_TODO_TITLE_CHARS = 500 +MAX_TODO_DESCRIPTION_CHARS = 20000 +MAX_TODO_QUERY_CHARS = 500 _CREATE_RE = re.compile(r"```task-create\s*\n(.*?)\n```", re.DOTALL) _POLL_RE = re.compile( @@ -115,6 +133,12 @@ _CANCEL_RE = re.compile(r"```(task-cancel|task_cancel)\s*\n(.*?)\n```", re.DOTALL) _RETRY_RE = re.compile(r"```(task-retry|task_retry)\s*\n(.*?)\n```", re.DOTALL) +_TODO_LIST_RE = re.compile( + r"```(todo-list|todo_list)\s*\n(.*?)\n```", re.DOTALL) +_TODO_CREATE_RE = re.compile( + r"```(todo-create|todo_create)\s*\n(.*?)\n```", re.DOTALL) +_TODO_UPDATE_RE = re.compile( + r"```(todo-update|todo_update)\s*\n(.*?)\n```", re.DOTALL) def row_get(row, name): @@ -190,6 +214,176 @@ def extract_task_creates(text): return out +def _todo_due_at(value): + """标记块 due_at 校验：与 todos._validate_due_at 同款语义 + （epoch 秒，>=1e11 按毫秒除 1000）；非法返回 (False, None)， + 不在本层抛异常——非法块整块跳过。""" + if value is None or value == "": + return True, None + if isinstance(value, bool) or not isinstance(value, (int, float)): + return False, None + value = float(value) + if value >= 1e11: + value = value / 1000.0 + return True, value + + +def extract_todo_lists(text): + """提取 todo-list 块；不传 status 时查未完成项（pending / + in_progress / blocked），不返回 completed / cancelled—— + 报告要短，查已完成项须显式传 status。status 非法、q 非 + 字符串或超长时整块跳过；每轮最多 5 块。""" + if not text: + return [] + out = [] + for match in _TODO_LIST_RE.finditer(str(text)): + if len(out) >= MAX_TODO_LISTS_PER_TURN: + break + try: + payload = json.loads(match.group(2)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + status = payload.get("status") + if status is not None and status not in TODO_STATUSES: + continue + q = payload.get("q") + if q is not None: + if not isinstance(q, str) or not q.strip(): + continue + q = q.strip() + if len(q) > MAX_TODO_QUERY_CHARS: + continue + out.append({"status": status, "q": q}) + return out + + +def extract_todo_creates(text): + """提取 todo-create 块；title 必填 1-500 字，description + 可选 <=20000 字，status / priority 只能是枚举值，task_id + 只能是字符串，due_at 按 _todo_due_at 校验；任一项非法时 + 整块跳过；每轮最多 5 块。""" + if not text: + return [] + out = [] + for match in _TODO_CREATE_RE.finditer(str(text)): + if len(out) >= MAX_TODO_CREATES_PER_TURN: + break + try: + payload = json.loads(match.group(2)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + title = payload.get("title") + if not isinstance(title, str) or not title.strip(): + continue + title = title.strip() + if len(title) > MAX_TODO_TITLE_CHARS: + continue + description = payload.get("description") + if description is not None: + if not isinstance(description, str): + continue + if len(description) > MAX_TODO_DESCRIPTION_CHARS: + continue + status = payload.get("status") + if status is not None and status not in TODO_STATUSES: + continue + priority = payload.get("priority") + if priority is not None and priority not in TODO_PRIORITIES: + continue + task_id = payload.get("task_id") + if task_id is not None: + if not isinstance(task_id, str): + continue + task_id = task_id.strip() or None + ok, due_at = _todo_due_at(payload.get("due_at")) + if not ok: + continue + sort_order = payload.get("sort_order") + if sort_order is not None and ( + isinstance(sort_order, bool) + or not isinstance(sort_order, int)): + continue + out.append({"title": title, "description": description, + "status": status, "priority": priority, + "task_id": task_id, "due_at": due_at, + "sort_order": sort_order}) + return out + + +def extract_todo_updates(text): + """提取 todo-update 块；id 必填，至少改一个字段；各字段 + 校验与 todo-create 同款（status / priority 枚举、title + 1-500 字、description <=20000 字、due_at 毫秒容忍、 + sort_order 整数）；任一项非法或无有效字段时整块跳过； + 每轮最多 5 块。未知字段忽略，不因此跳过。""" + if not text: + return [] + out = [] + for match in _TODO_UPDATE_RE.finditer(str(text)): + if len(out) >= MAX_TODO_UPDATES_PER_TURN: + break + try: + payload = json.loads(match.group(2)) + except (TypeError, ValueError): + continue + if not isinstance(payload, dict): + continue + todo_id = payload.get("id") + if not isinstance(todo_id, str) or not todo_id.strip(): + continue + fields = {} + if "title" in payload: + title = payload["title"] + if not isinstance(title, str) or not title.strip(): + continue + title = title.strip() + if len(title) > MAX_TODO_TITLE_CHARS: + continue + fields["title"] = title + if "description" in payload: + description = payload["description"] + if not isinstance(description, str): + continue + if len(description) > MAX_TODO_DESCRIPTION_CHARS: + continue + fields["description"] = description + if "status" in payload: + if payload["status"] not in TODO_STATUSES: + continue + fields["status"] = payload["status"] + if "priority" in payload: + if payload["priority"] not in TODO_PRIORITIES: + continue + fields["priority"] = payload["priority"] + if "task_id" in payload: + task_id = payload["task_id"] + if task_id is not None and not isinstance(task_id, str): + continue + fields["task_id"] = (task_id.strip() or None + if isinstance(task_id, str) else None) + if "due_at" in payload: + ok, due_at = _todo_due_at(payload["due_at"]) + if not ok: + continue + fields["due_at"] = due_at + if "sort_order" in payload: + sort_order = payload["sort_order"] + if sort_order is not None and ( + isinstance(sort_order, bool) + or not isinstance(sort_order, int)): + continue + if sort_order is not None: + fields["sort_order"] = sort_order + if not fields: + continue + out.append({"id": todo_id.strip(), "fields": fields}) + return out + + _SPAWN_SUBJECT_ZH = ("新建的任务", "派生任务", "新建任务", "子任务", "子代理", "新任务", "新会话") _SPAWN_SUBJECT_EN = ("child tasks", "spawned tasks", "new tasks", @@ -649,3 +843,26 @@ def new_child_id(): def marker_report_header(): return "[任务工具报告]" + + +def format_todo(todo, max_title=100): + """待办单行（报告与首轮注入共用）：状态 + 标题（截断）+ id + + 优先级，due_at 有值时附 UTC 时间；标题里的换行压成空格， + 避免一行待办撑开多行报告。""" + title = str(todo.get("title") or "").replace("\n", " ").strip() + if len(title) > max_title: + title = title[:max_title] + "…" + parts = ["• [%s] %s (id=%s, priority=%s)" + % (todo.get("status"), title, todo.get("id"), + todo.get("priority"))] + due_at = todo.get("due_at") + if due_at: + try: + parts.append("due=%s" % time.strftime( + "%Y-%m-%dT%H:%M:%SZ", time.gmtime(float(due_at)))) + except (TypeError, ValueError, OverflowError): + pass + task_id = todo.get("task_id") + if task_id: + parts.append("task_id=%s" % task_id) + return " ".join(parts) diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 9456251..0aaa0eb 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -26,6 +26,9 @@ from cloud.marker_tools import extract_task_cancels as _extract_task_cancels from cloud.marker_tools import extract_task_creates as _extract_task_creates from cloud.marker_tools import extract_task_polls as _extract_task_polls from cloud.marker_tools import extract_task_retries as _extract_task_retries +from cloud.marker_tools import extract_todo_creates as _extract_todo_creates +from cloud.marker_tools import extract_todo_lists as _extract_todo_lists +from cloud.marker_tools import extract_todo_updates as _extract_todo_updates from cloud.marker_tools import extract_wait_threads as _extract_wait_threads from cloud.peer_messages import extract_peer_messages as _extract_peer_messages @@ -404,6 +407,9 @@ class LocalExecutor(Executor): wait_threads = [] task_cancels = [] task_retries = [] + todo_lists = [] + todo_creates = [] + todo_updates = [] turn_id = None root_turn_id = None parsed = None @@ -444,6 +450,9 @@ class LocalExecutor(Executor): wait_threads = _extract_wait_threads(parsed.get("text")) task_cancels = _extract_task_cancels(parsed.get("text")) task_retries = _extract_task_retries(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")) else: host_status = None else: @@ -468,6 +477,9 @@ class LocalExecutor(Executor): "wait_threads": wait_threads, "task_cancels": task_cancels, "task_retries": task_retries, + "todo_lists": todo_lists, + "todo_creates": todo_creates, + "todo_updates": todo_updates, "turn_id": turn_id, "root_turn_id": root_turn_id, } @@ -1909,6 +1921,9 @@ def _parse_host_result(task_id, fake): wait_threads = [] task_cancels = [] task_retries = [] + todo_lists = [] + todo_creates = [] + todo_updates = [] turn_id = None root_turn_id = None parsed = None @@ -1936,6 +1951,9 @@ def _parse_host_result(task_id, fake): wait_threads = _extract_wait_threads(parsed.get("text")) task_cancels = _extract_task_cancels(parsed.get("text")) task_retries = _extract_task_retries(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")) # 任务列表显示 turn 用：与 Local 版同款语义，取不到 # 时保持 None，worker 写回时不覆盖旧值。 turn_id = parsed.get("turn_id") @@ -1957,6 +1975,9 @@ def _parse_host_result(task_id, fake): "wait_threads": wait_threads, "task_cancels": task_cancels, "task_retries": task_retries, + "todo_lists": todo_lists, + "todo_creates": todo_creates, + "todo_updates": todo_updates, "turn_id": turn_id, "root_turn_id": root_turn_id, } diff --git a/cloud/tests/test_todo_markers.py b/cloud/tests/test_todo_markers.py new file mode 100644 index 0000000..aa02419 --- /dev/null +++ b/cloud/tests/test_todo_markers.py @@ -0,0 +1,330 @@ +"""项目待办标记工具单测：todo-list / todo-create / todo-update +解析、sandbox 提取、worker 投递与首轮注入。 + +不跑真实模型、不联网：解析是纯函数；worker 投递与注入用 +init_db 建的临时 SQLite 库，直接调 _deliver_marker_tools 与 +_apply_todo_injection，不走 run_one 全量流程。 +""" + +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.db import connect as db_connect + from cloud.marker_tools import ( + extract_todo_creates, + extract_todo_lists, + extract_todo_updates, + format_todo, + ) + from cloud.sandbox import LocalExecutor, _parse_host_result + from cloud.server import init_db + from cloud.todos import create_todo, list_todos + 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.db import connect as db_connect + from cloud.marker_tools import ( + extract_todo_creates, + extract_todo_lists, + extract_todo_updates, + format_todo, + ) + from cloud.sandbox import LocalExecutor, _parse_host_result + from cloud.server import init_db + from cloud.todos import create_todo, list_todos + from cloud.worker import Worker + +FAKE_ENV = {"XCM_API_KEY": "test-placeholder"} + + +class ExtractTestCase(unittest.TestCase): + def test_list_default_and_alias(self): + out = extract_todo_lists("```todo-list\n{}\n```") + self.assertEqual(out, [{"status": None, "q": None}]) + out = extract_todo_lists("```todo_list\n{\"status\": \"blocked\"}\n```") + self.assertEqual(out, [{"status": "blocked", "q": None}]) + + def test_list_invalid_status_skipped(self): + out = extract_todo_lists( + "```todo-list\n{\"status\": \"doing\"}\n```\n" + "```todo-list\n{\"status\": \"pending\"}\n```") + self.assertEqual(len(out), 1) + self.assertEqual(out[0]["status"], "pending") + + def test_list_cap_five(self): + text = "\n".join("```todo-list\n{}\n```" for _ in range(6)) + self.assertEqual(len(extract_todo_lists(text)), 5) + + def test_create_valid_and_defaults(self): + out = extract_todo_creates( + "```todo-create\n{\"title\": \"做 X\"}\n```") + self.assertEqual(len(out), 1) + self.assertEqual(out[0]["title"], "做 X") + self.assertIsNone(out[0]["status"]) + self.assertIsNone(out[0]["priority"]) + + def test_create_invalid_skipped(self): + cases = [ + "{\"title\": \"\"}", + "{\"title\": \"x\", \"status\": \"doing\"}", + "{\"title\": \"x\", \"priority\": \"urgent\"}", + "{\"title\": \"x\", \"due_at\": \"明天\"}", + "{\"title\": \"x\", \"sort_order\": 1.5}", + "{\"description\": \"no title\"}", + ] + for body in cases: + with self.subTest(body=body): + self.assertEqual( + extract_todo_creates("```todo-create\n%s\n```" % body), + []) + + def test_create_due_ms_tolerance(self): + out = extract_todo_creates( + "```todo-create\n{\"title\": \"x\", \"due_at\": 1700000000000}\n```") + self.assertEqual(out[0]["due_at"], 1700000000.0) + + def test_update_valid_and_empty_skipped(self): + out = extract_todo_updates( + "```todo-update\n{\"id\": \"abc\", \"status\": \"completed\"}\n```") + self.assertEqual(out, [{"id": "abc", + "fields": {"status": "completed"}}]) + self.assertEqual(extract_todo_updates( + "```todo-update\n{\"id\": \"abc\"}\n```"), []) + self.assertEqual(extract_todo_updates( + "```todo-update\n{\"id\": \"abc\", \"status\": \"doing\"}\n```"), []) + self.assertEqual(extract_todo_updates( + "```todo-update\n{\"status\": \"completed\"}\n```"), []) + + def test_update_unknown_field_ignored(self): + out = extract_todo_updates( + "```todo-update\n{\"id\": \"abc\", \"status\": \"blocked\"," + " \"note\": \"x\"}\n```") + self.assertEqual(out[0]["fields"], {"status": "blocked"}) + + def test_format_todo_truncates_and_due_utc(self): + line = format_todo({"title": "x" * 120, "status": "pending", + "id": "t1", "priority": "high", + "due_at": 1700000000, "task_id": None}) + self.assertIn("…", line) + self.assertIn("due=2023-11-14T22:13:20Z", line) + + def test_sandbox_extracts_both_executors(self): + text = ("```todo-list\n{}\n```\n" + "```todo-create\n{\"title\": \"做 X\"}\n```\n" + "```todo-update\n{\"id\": \"t1\", \"status\": \"completed\"}" + "\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["todo_lists"]), 1) + self.assertEqual(len(local["todo_creates"]), 1) + self.assertEqual(len(local["todo_updates"]), 1) + docker = _parse_host_result("t1", fake) + self.assertEqual(len(docker["todo_lists"]), 1) + self.assertEqual(len(docker["todo_creates"]), 1) + self.assertEqual(len(docker["todo_updates"]), 1) + + +class WorkerTodoTestCase(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", **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, "do work", status, created_by, + now, now)) + 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 _seed_todo(self, title, **kw): + conn = db_connect(self.db_path) + try: + return create_todo(conn, "p1", "tester", title, **kw) + finally: + conn.close() + + def _followup_bodies(self): + conn = sqlite3.connect(self.db_path) + try: + rows = conn.execute( + "SELECT body FROM task_followups" + " WHERE task_id = 'PM'").fetchall() + return [row[0] for row in rows] + finally: + conn.close() + + def _todos(self): + conn = db_connect(self.db_path) + try: + return list_todos(conn, "p1") + finally: + conn.close() + + def test_deliver_create_update_list(self): + worker = self._worker() + sender = worker.get_task("PM") + todo = self._seed_todo("旧项") + worker._deliver_marker_tools(sender, { + "status": "completed", + "todo_creates": [{"title": "新项", "description": None, + "status": None, "priority": "high", + "task_id": None, "due_at": None, + "sort_order": None}], + "todo_updates": [{"id": todo["id"], + "fields": {"status": "completed"}}], + "todo_lists": [{"status": None, "q": None}]}) + titles = [t["title"] for t in self._todos()] + self.assertIn("新项", titles) + updated = [t for t in self._todos() if t["id"] == todo["id"]][0] + self.assertEqual(updated["status"], "completed") + self.assertIsNotNone(updated["completed_at"]) + bodies = self._followup_bodies() + self.assertEqual(len(bodies), 1) + # 列表默认只返回未完成项：已完成的旧项不在列表段里， + # 新项在（更新报告行里提到旧项是正常的，不算列表）。 + self.assertIn("新项", bodies[0]) + list_section = bodies[0].split("待办列表", 1)[1] + self.assertNotIn("旧项", list_section) + + def test_deliver_cross_project_update_rejected(self): + worker = self._worker() + sender = worker.get_task("PM") + conn = db_connect(self.db_path) + try: + other = create_todo(conn, "p2", "tester", "别项目") + finally: + conn.close() + worker._deliver_marker_tools(sender, { + "status": "completed", + "todo_creates": [], "todo_lists": [], + "todo_updates": [{"id": other["id"], + "fields": {"status": "completed"}}]}) + bodies = self._followup_bodies() + self.assertEqual(len(bodies), 1) + self.assertIn("不存在", bodies[0]) + conn = db_connect(self.db_path) + try: + kept = [t for t in list_todos(conn, "p2") + if t["id"] == other["id"]][0] + finally: + conn.close() + self.assertEqual(kept["status"], "pending") + + def test_deliver_denied_without_write_permission(self): + self._member("viewer1", "viewer") + self._insert("PM2", status="running", created_by="viewer1") + worker = self._worker() + sender = worker.get_task("PM2") + worker._deliver_marker_tools(sender, { + "status": "completed", + "todo_creates": [{"title": "不应建", "description": None, + "status": None, "priority": None, + "task_id": None, "due_at": None, + "sort_order": None}], + "todo_lists": [], "todo_updates": []}) + self.assertEqual( + [t["title"] for t in self._todos()], []) + conn = sqlite3.connect(self.db_path) + try: + bodies = [row[0] for row in conn.execute( + "SELECT body FROM task_followups" + " WHERE task_id = 'PM2'").fetchall()] + finally: + conn.close() + self.assertIn("无 create_task 权限", bodies[0]) + + def test_deliver_goal_sender_ignored(self): + self._insert("G1", status="running", goal_mode=1) + worker = self._worker() + sender = worker.get_task("G1") + worker._deliver_marker_tools(sender, { + "status": "completed", + "todo_creates": [{"title": "不应建", "description": None, + "status": None, "priority": None, + "task_id": None, "due_at": None, + "sort_order": None}], + "todo_lists": [], "todo_updates": []}) + self.assertEqual([t["title"] for t in self._todos()], []) + + def test_injection_lists_open_only(self): + self._seed_todo("待做", status="pending") + self._seed_todo("进行中", status="in_progress") + self._seed_todo("已做完", status="completed") + worker = self._worker() + task = {"id": "PM", "project_id": "p1", "prompt": "做事", + "created_by": "tester"} + out = worker._apply_todo_injection(task) + self.assertIn("待做", out["prompt"]) + self.assertIn("进行中", out["prompt"]) + self.assertNotIn("已做完", out["prompt"]) + self.assertIn("todo-update", out["prompt"]) + # 原 prompt 不落库：内存副本改了，调用方传的 dict 不动。 + self.assertEqual(task["prompt"], "做事") + + def test_injection_skips_goal_and_empty(self): + worker = self._worker() + goal = {"id": "G1", "project_id": "p1", "prompt": "目标", + "created_by": "tester", "goal_mode": 1} + self.assertIs(worker._apply_todo_injection(goal), goal) + empty = {"id": "PM", "project_id": "p1", "prompt": "做事", + "created_by": "tester"} + self.assertIs(worker._apply_todo_injection(empty), empty) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 016f086..4603375 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -154,7 +154,12 @@ 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_todo as _format_todo 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 +from cloud.todos import list_todos as _todo_list_db +from cloud.todos import update_todo as _todo_update_db from cloud.marker_tools import new_child_id as _marker_new_id from cloud.marker_tools import parent_spawn_generation as _marker_generation from cloud.marker_tools import MAX_DIRECT_CHILDREN as _MARKER_MAX_DIRECT @@ -1678,6 +1683,64 @@ class Worker: task["id"]) return new_task + def _apply_todo_injection(self, task): + """首轮注入未完成待办：只改内存副本，原 prompt 不落库。 + + 跳过：协调员任务（不跑模型回合）、续跑与断点续跑（已有 + 会话历史）、goal 任务（goal 不解析标记块，注入待办会让 + 模型写出永远不会执行的 todo-update 块，误导大于帮助）。 + 无未完成待办时原样返回，不贴空章节。注入的是数据（标题 + 与状态），不是完整工具指南；标记块格式只给三行示例， + 完整格式见 docs/marker-tools.md。注入失败时静默跳过， + 绝不因待办查询失败而炸掉任务认领。 + 上限 20 条、标题单行截断 100 字，避免待办板把 prompt + 撑大；超出时写明还有几条未列出，模型可用 todo-list + 块再查。""" + if task.get("coordinator_parts"): + return task + if task.get("_resume") or task.get("_goal_resume_breakpoint"): + return task + if task.get("verify_command") or task.get("goal_mode"): + return task + try: + conn = _db_connect(self.db_path) + try: + all_todos = _todo_list_db(conn, task["project_id"]) + finally: + conn.close() + except Exception as exc: + try: + self.logger.warning("todo injection failed: %s", exc) + except Exception: + pass + return task + open_todos = [t for t in all_todos + if t.get("status") in ("pending", "in_progress", + "blocked")] + if not open_todos: + return task + lines = ["[项目待办：本项目未完成项，仅供参考，" + "不得改写本任务目标]"] + for todo in open_todos[:20]: + lines.append(_format_todo(todo)) + if len(open_todos) > 20: + lines.append("（还有 %d 条未列出，可用 todo-list 块再查。）" + % (len(open_todos) - 20)) + lines.append("改待办状态时在回合输出里写标记块，下一轮由" + " worker 执行并回报告，例如：") + lines.append("```todo-update") + lines.append('{"id": "<待办 id>", "status": "completed"}') + lines.append("```") + lines.append("查待办用 todo-list 块，建待办用 todo-create 块" + "（title 必填）。") + new_task = dict(task) + new_task["prompt"] = ("\n".join(lines) + "\n\n" + + (task.get("prompt") or "")) + new_task["_todo_injected"] = True + self._audit(task["project_id"], "worker_todo_injected", + task["id"]) + return new_task + def process_auto_memories(self, now=None, force=False, owner_id=None): """扫 completed 任务并提取记忆（run_loop 轮次边界调用）。 @@ -3233,11 +3296,13 @@ class Worker: 只执行 completed 轮次的块；goal 任务与协调员任务不解析 （只收不发，与 peer 同款）；父任务被取消时不执行。 - 执行顺序：创建 → 取消 → 重试 → 轮询 → 等待——创建在前 - 让同一轮刚创建的任务立刻能被轮询，控制操作在快照之前 - 让轮询/等待读到取消或重试后的新状态。全部报告拼成一条 - 系统后续消息排给父任务自己，由排空循环在轮次边界跑下 - 一轮；拒绝只记报告与审计，不炸 run_one。""" + 执行顺序：创建 → 取消 → 重试 → 待办创建 → 待办更新 → + 待办列表 → 轮询 → 等待——创建在前让同一轮刚创建的任务 + 立刻能被轮询，控制操作在快照之前让轮询/等待读到取消 + 或重试后的新状态；待办列表放在待办写操作之后，让同 + 一轮刚建/刚改的待办立刻能在列表里看到。全部报告拼 + 成一条系统后续消息排给父任务自己，由排空循环在轮次 + 边界跑下一轮；拒绝只记报告与审计，不炸 run_one。""" if not result or result.get("status") != "completed": return if self._was_cancelled(task["id"]): @@ -3251,7 +3316,11 @@ class Worker: waits = result.get("wait_threads") or [] cancels = result.get("task_cancels") or [] retries = result.get("task_retries") or [] - if not (creates or polls or waits or cancels or retries): + todo_lists = result.get("todo_lists") or [] + todo_creates = result.get("todo_creates") or [] + todo_updates = result.get("todo_updates") or [] + if not (creates or polls or waits or cancels or retries + or todo_lists or todo_creates or todo_updates): return reports = [] created = 0 @@ -3429,6 +3498,131 @@ class Worker: reports.append("%s任务 %s：执行失败（%s）。" % (kind, target_id, exc)) rejected += 1 + # 待办标记块：读走 read、写走 create_task（与服务端 + # /todos 口径同款）；权限检查抛异常时（auth 为 None） + # 一律拒绝，不静默放宽。跨项目待办按“不存在”拒绝； + # 关联任务 id 跨项目时同样拒绝，避免把别的项目的任务 + # 挂到本项目待办上。 + if todo_lists or todo_creates or todo_updates: + read_ok, _read_reason = self._marker_auth(task, "read") + write_ok, _write_reason = self._marker_auth( + task, "create_task") + for block in todo_creates: + if write_ok is not True: + reports.append("创建待办“%s”：拒绝（发送方" + "创建人无 create_task 权限）。 " + % block["title"]) + rejected += 1 + continue + if block.get("task_id"): + target = self.get_task(block["task_id"]) + if (target is None + or target.get("project_id") + != task["project_id"]): + reports.append("创建待办“%s”：拒绝（关联" + "任务不存在或属于别的项目）。" + % block["title"]) + rejected += 1 + continue + try: + conn = _db_connect(self.db_path) + try: + todo = _todo_create_db( + conn, task["project_id"], + task.get("created_by"), + block["title"], + description=block.get("description"), + status=block.get("status"), + priority=block.get("priority"), + task_id=block.get("task_id"), + due_at=block.get("due_at"), + sort_order=block.get("sort_order")) + finally: + conn.close() + reports.append("已创建待办 %s" + % _format_todo(todo)) + except Exception as exc: + reports.append("创建待办“%s”失败：%s" + % (block["title"], exc)) + rejected += 1 + for block in todo_updates: + if write_ok is not True: + reports.append("更新待办 %s：拒绝（发送方创建" + "人无 create_task 权限）。" + % block["id"]) + rejected += 1 + continue + fields = block["fields"] + if fields.get("task_id"): + target = self.get_task(fields["task_id"]) + if (target is None + or target.get("project_id") + != task["project_id"]): + reports.append("更新待办 %s：拒绝（关联" + "任务不存在或属于别的项目）。" + % block["id"]) + rejected += 1 + continue + try: + conn = _db_connect(self.db_path) + try: + todo = _todo_get_db(conn, block["id"]) + if (todo is None + or todo.get("project_id") + != task["project_id"]): + reports.append("更新待办 %s：不存在" + "（或属于别的项目），" + "已跳过。" % block["id"]) + rejected += 1 + continue + todo = _todo_update_db(conn, block["id"], + fields) + finally: + conn.close() + reports.append("已更新待办 %s" + % _format_todo(todo)) + except Exception as exc: + reports.append("更新待办 %s 失败：%s" + % (block["id"], exc)) + rejected += 1 + for block in todo_lists: + if read_ok is not True: + reports.append("查待办：拒绝（发送方创建人无" + " read 权限）。") + rejected += 1 + continue + try: + conn = _db_connect(self.db_path) + try: + if block.get("status"): + todos = _todo_list_db( + conn, task["project_id"], + status=block["status"], + q=block.get("q")) + else: + todos = [t for t in _todo_list_db( + conn, task["project_id"], + q=block.get("q")) + if t.get("status") in ( + "pending", "in_progress", + "blocked")] + conn.commit() + finally: + conn.close() + except Exception as exc: + reports.append("查待办失败：%s" % exc) + rejected += 1 + continue + if not todos: + reports.append("待办列表：无匹配项。") + else: + lines = ["待办列表（共 %d 条）：" % len(todos)] + for todo in todos[:20]: + lines.append(_format_todo(todo)) + if len(todos) > 20: + lines.append("（还有 %d 条未列出。）" + % (len(todos) - 20)) + reports.append("\n".join(lines)) for poll in polls: try: reports.append(self._poll_task_snapshot(task, poll)) @@ -3444,9 +3638,12 @@ class Worker: body = _marker_header() + "\n" + "\n".join(reports) self._queue_self_followup(task, body) detail = ("%s: spawned=%d rejected=%d polls=%d waits=%d " - "cancels=%d retries=%d" + "cancels=%d retries=%d todo_lists=%d " + "todo_creates=%d todo_updates=%d" % (task["id"], created, rejected, len(polls), - len(waits), len(cancels), len(retries))) + len(waits), len(cancels), len(retries), + len(todo_lists), len(todo_creates), + len(todo_updates))) self._audit(task["project_id"], "worker_marker_tools", detail) def _deliver_peer_messages(self, task, result): @@ -4696,6 +4893,9 @@ class Worker: # 自动记忆注入（官方 use_memories）：首轮才注入，见 # _apply_auto_memory_injection 的跳过规则；只改内存副本。 task = self._apply_auto_memory_injection(task) + # 待办注入：首轮才注入未完成待办，见 + # _apply_todo_injection 的跳过规则；只改内存副本。 + task = self._apply_todo_injection(task) self._audit(task["project_id"], "worker_claim", "%s executor=%s" % (task_id, self._executor_name())) # log_event 只接受 task_id/project_id/status/message 四个 diff --git a/docs/marker-tools.md b/docs/marker-tools.md index 5963bcd..a1ffa4c 100644 --- a/docs/marker-tools.md +++ b/docs/marker-tools.md @@ -195,9 +195,10 @@ Docker 执行器都经 sandbox 的结果解析提取标记块（与 task-message - cloud/marker_tools.py：纯逻辑（解析、校验、快照格式化、 门禁、派生代数与直接上限计数），单测不需要 worker/server。 - cloud/sandbox.py：本地 _result_from_runner 与 Docker - _parse_host_result 两处提取五种标记块，worker 侧按结果 + _parse_host_result 两处提取标记块，worker 侧按结果 dict 里的 task_creates / task_polls / wait_threads / - task_cancels / task_retries 执行。 + task_cancels / task_retries / todo_lists / todo_creates / + todo_updates 执行。 - cloud/worker.py：_deliver_marker_tools（5 处投递点，与 _deliver_peer_messages 同款位置）、_poll_task_snapshot、 _wait_task_targets、_queue_self_followup、_marker_auth。 @@ -205,3 +206,63 @@ Docker 执行器都经 sandbox 的结果解析提取标记块（与 task-message 跨项目拒绝、自操作拒绝、已终态拒绝、goal / 已合并拒绝、 重试仅三态且清空旧结果、wait 多目标终态与超时快照、派生 深度与直接上限、权限检查失败收紧。 + +## 待办标记工具（todo-list / todo-create / todo-update） + +与上面四类同款异步语义：写块 → 回合结束 → worker 执行 → +报告拼成后续消息、下一轮才回到父任务。goal 任务与协调员 +任务不解析（只收不发），所以首轮待办注入也跳过 goal 任务， +避免模型写出永远不会执行的 todo-update 块。 + +### todo-list：查本项目待办 + +```text +```todo-list +{"status": "pending", "q": "文档"} +``` +``` + +- 不传 status 时只返回未完成项（pending / in_progress / + blocked），查已完成项须显式传 status；q 是标题/描述子串。 +- 每轮最多 5 块；status 非法、q 非字符串或超长时整块跳过。 +- 读权限：发送方创建人须有项目 read 权限，否则报告里写清 + 拒绝；报告最多列 20 条，超出写明还有几条。 + +### todo-create：新建待办 + +```text +```todo-create +{"title": "写接口文档", "priority": "high"} +``` +``` + +- title 必填 1-500 字；description 可选 <=20000 字； + status / priority 只能是枚举值；due_at 是 epoch 秒， + >=1e11 按毫秒除 1000；task_id 关联的任务必须在本项目内， + 否则整块拒绝；sort_order 整数可选。 +- 每轮最多 5 块；写权限：发送方创建人须有 create_task + 权限，否则拒绝。 + +### todo-update：改待办（最常用是改状态） + +```text +```todo-update +{"id": "<待办 id>", "status": "completed"} +``` +``` + +- id 必填，至少改一个字段；各字段校验与 todo-create 同款， + 未知字段忽略；行不存在或属于别的项目时按“不存在”拒绝， + 不改任何状态；completed 记 completed_at，离开 completed + 清掉（复用 todos 模块状态机）。 +- 每轮最多 5 块；删除不走标记块（硬删除只在控制台 / + REST DELETE 里做，避免模型误删整条）。 + +### 首轮注入（worker _apply_todo_injection） + +- 普通任务被认领的首轮，worker 在内存副本的 prompt 前贴 + 一份本项目未完成待办（最多 20 条、标题截断 100 字），附 + 三行 todo-update 示例；原 prompt 不落库，续跑与断点续跑 + 不重复贴，协调员与 goal 任务跳过，无未完成待办时不贴空 + 章节，注入查询失败时静默跳过、不炸认领。 +- 注入贴的是数据，不是完整工具指南；完整格式只在本文件里。