待办：todo-list/create/update 标记工具 + 首轮未完成待办注入

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)

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,
     }

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()

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 四个

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 任务跳过，无未完成待办时不贴空
+  章节，注入查询失败时静默跳过、不炸认领。
+- 注入贴的是数据，不是完整工具指南；完整格式只在本文件里。