任务列表 task-list 标记 + 首轮项目上下文注入

cloud/marker_tools.py

@@ -1,6 +1,6 @@
 """任务内标记工具（marker tools）：task-create / task-poll / task-read /
 wait-threads / task-cancel / task-retry / todo-list / todo-create /
-todo-update。
+todo-update / task-list。
 
 设计（与 peer_messages、task_spawn 同款原则，不另起一套）：
 
@@ -52,12 +52,16 @@ todo-update。
   查已完成项须显式传 status；写操作只在本项目内生效，
   跨项目待办一律按“不存在”拒绝；删除不走标记块（硬删除
   只在控制台 / REST DELETE 里做，避免模型误删整条）。
+- task-list：列出本项目任务（新建倒序），让模型在不知道
+  任务 id 时也能分诊——查运行中与失败的任务、捡失败任务
+  重试。不传 status 时查全部状态并附各状态计数；只读，
+  跨项目任务不出现，报告最多列 limit 条（默认 20，上限 50）。
 
 上限口径（本模块最终定的值，文档与代码一致）：
 
 - 每轮创建、轮询、取消、重试各最多 5 块；todo-list、
-  todo-create、todo-update 各最多 5 块；wait 每轮最多
+  todo-create、todo-update、task-list 各最多 5 块；wait
-  2 块，每块最多 8 个目标。8 这个数是刻意与服务端
+  每轮最多 2 块，每块最多 8 个目标。8 这个数是刻意与服务端
   POST /tasks/{id}/wait 的“一次最多 8 个线程”对齐的；
   spawn 分支原先的每块 10 个在此收紧到 8 个，不要再用 10。
 - 超时口径与服务端 /wait 不同，且是故意的：服务端 /wait
@@ -122,6 +126,11 @@ MAX_TODO_UPDATES_PER_TURN = 5
 MAX_TODO_TITLE_CHARS = 500
 MAX_TODO_DESCRIPTION_CHARS = 20000
 MAX_TODO_QUERY_CHARS = 500
+TASK_STATUSES = ("queued", "running", "completed", "failed",
+                 "interrupted", "merged", "merge-failed")
+MAX_TASK_LISTS_PER_TURN = 5
+DEFAULT_TASK_LIST_LIMIT = 20
+MAX_TASK_LIST_LIMIT = 50
 
 _CREATE_RE = re.compile(r"```task-create\s*\n(.*?)\n```", re.DOTALL)
 _POLL_RE = re.compile(
@@ -139,6 +148,8 @@ _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)
+_TASK_LIST_RE = re.compile(
+    r"```(task-list|task_list)\s*\n(.*?)\n```", re.DOTALL)
 
 
 def row_get(row, name):
@@ -384,6 +395,45 @@ def extract_todo_updates(text):
     return out
 
 
+def extract_task_lists(text):
+    """提取 task-list 块；不传 status 时查全部状态（与 todo-list
+    默认只查未完成不同——任务分诊首先要看到失败与运行中的，
+    全部状态的计数与截断报告才是可用的概览）。q 是 prompt
+    子串，limit 钳制 1-50（默认 20）；status 非法、q 非字符
+    串或超长、limit 非整数时整块跳过；每轮最多 5 块。"""
+    if not text:
+        return []
+    out = []
+    for match in _TASK_LIST_RE.finditer(str(text)):
+        if len(out) >= MAX_TASK_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 TASK_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
+        limit = payload.get("limit")
+        if limit is None:
+            limit = DEFAULT_TASK_LIST_LIMIT
+        elif isinstance(limit, bool) or not isinstance(limit, int):
+            continue
+        else:
+            limit = max(1, min(MAX_TASK_LIST_LIMIT, limit))
+        out.append({"status": status, "q": q, "limit": limit})
+    return out
+
+
 _SPAWN_SUBJECT_ZH = ("新建的任务", "派生任务", "新建任务",
                      "子任务", "子代理", "新任务", "新会话")
 _SPAWN_SUBJECT_EN = ("child tasks", "spawned tasks", "new tasks",
@@ -866,3 +916,25 @@ def format_todo(todo, max_title=100):
     if task_id:
         parts.append("task_id=%s" % task_id)
     return " ".join(parts)
+
+
+def format_task(task, max_prompt=80):
+    """任务单行（task-list 报告用）：状态 + id + 创建人 + prompt
+    截断；prompt 里的换行压成空格，避免一行任务撑开多行报告；
+    创建时间有值时附 UTC 时间。"""
+    prompt = str(task.get("prompt") or "").replace("\n", " ").strip()
+    if len(prompt) > max_prompt:
+        prompt = prompt[:max_prompt] + "…"
+    line = ("• [%s] %s (by=%s)"
+            % (task.get("status"), task.get("id"),
+               task.get("created_by") or ""))
+    created_at = task.get("created_at")
+    if created_at:
+        try:
+            line += " created=%s" % time.strftime(
+                "%Y-%m-%dT%H:%M:%SZ", time.gmtime(float(created_at)))
+        except (TypeError, ValueError, OverflowError):
+            pass
+    if prompt:
+        line += " " + prompt
+    return line

cloud/sandbox.py

@@ -24,6 +24,7 @@ import time
 
 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_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
 from cloud.marker_tools import extract_todo_creates as _extract_todo_creates
@@ -407,6 +408,7 @@ class LocalExecutor(Executor):
         wait_threads = []
         task_cancels = []
         task_retries = []
+        task_lists = []
         todo_lists = []
         todo_creates = []
         todo_updates = []
@@ -450,6 +452,7 @@ 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"))
+                task_lists = _extract_task_lists(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"))
@@ -477,6 +480,7 @@ class LocalExecutor(Executor):
             "wait_threads": wait_threads,
             "task_cancels": task_cancels,
             "task_retries": task_retries,
+            "task_lists": task_lists,
             "todo_lists": todo_lists,
             "todo_creates": todo_creates,
             "todo_updates": todo_updates,
@@ -1921,6 +1925,7 @@ def _parse_host_result(task_id, fake):
     wait_threads = []
     task_cancels = []
     task_retries = []
+    task_lists = []
     todo_lists = []
     todo_creates = []
     todo_updates = []
@@ -1951,6 +1956,7 @@ 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"))
+            task_lists = _extract_task_lists(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"))
@@ -1975,6 +1981,7 @@ def _parse_host_result(task_id, fake):
         "wait_threads": wait_threads,
         "task_cancels": task_cancels,
         "task_retries": task_retries,
+        "task_lists": task_lists,
         "todo_lists": todo_lists,
         "todo_creates": todo_creates,
         "todo_updates": todo_updates,

cloud/tests/test_task_list_markers.py

@@ -0,0 +1,256 @@
+"""任务列表标记与项目注入单测：task-list 解析、sandbox 提取、
+worker 投递与首轮项目注入。
+
+不跑真实模型、不联网：解析是纯函数；worker 投递与注入用
+init_db 建的临时 SQLite 库，直接调 _deliver_marker_tools 与
+_apply_project_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.marker_tools import extract_task_lists, format_task
+    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.marker_tools import extract_task_lists, format_task
+    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_list_default_and_alias(self):
+        out = extract_task_lists("```task-list\n{}\n```")
+        self.assertEqual(out, [{"status": None, "q": None, "limit": 20}])
+        out = extract_task_lists(
+            "```task_list\n{\"status\": \"failed\", \"limit\": 5}\n```")
+        self.assertEqual(out, [{"status": "failed", "q": None,
+                                "limit": 5}])
+
+    def test_list_limit_clamped_and_invalid_skipped(self):
+        out = extract_task_lists(
+            "```task-list\n{\"limit\": 999}\n```")
+        self.assertEqual(out[0]["limit"], 50)
+        out = extract_task_lists(
+            "```task-list\n{\"limit\": 0}\n```")
+        self.assertEqual(out[0]["limit"], 1)
+        for body in ("{\"status\": \"doing\"}",
+                     "{\"limit\": \"many\"}",
+                     "{\"limit\": true}",
+                     "{\"q\": 123}"):
+            with self.subTest(body=body):
+                self.assertEqual(extract_task_lists(
+                    "```task-list\n%s\n```" % body), [])
+
+    def test_list_cap_five(self):
+        text = "\n".join("```task-list\n{}\n```" for _ in range(6))
+        self.assertEqual(len(extract_task_lists(text)), 5)
+
+    def test_format_task_truncates(self):
+        line = format_task({"id": "t1", "status": "failed",
+                            "prompt": "x" * 100, "created_by": "u1",
+                            "created_at": 1700000000})
+        self.assertIn("…", line)
+        self.assertIn("2023-11-14T22:13:20Z", line)
+
+    def test_sandbox_extracts_both_executors(self):
+        text = "```task-list\n{\"status\": \"failed\"}\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_lists"]), 1)
+        docker = _parse_host_result("t1", fake)
+        self.assertEqual(len(docker["task_lists"]), 1)
+
+
+class WorkerTaskListTestCase(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", **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))
+            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 _project(self, project_id="p1", name="演示项目",
+                 repo_url="git@github.com:openqj/demo.git"):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT OR REPLACE INTO projects(id, name, owner_id,"
+                " created_at, repo_url, auto_pr)"
+                " VALUES(?, ?, 'tester', ?, ?, 0)",
+                (project_id, name, time.time(), repo_url))
+            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_lists_all_with_counts(self):
+        self._insert("E1", status="failed", prompt="修 bug")
+        self._insert("E2", status="running", prompt="做功能")
+        self._insert("X1", status="failed", project_id="p2",
+                     prompt="别项目")
+        worker = self._worker()
+        worker._deliver_marker_tools(worker.get_task("PM"), {
+            "status": "completed",
+            "task_lists": [{"status": None, "q": None, "limit": 20}]})
+        bodies = self._followup_bodies()
+        self.assertEqual(len(bodies), 1)
+        # 全部状态 + 计数在，跨项目任务不在，PM 自己在（本项目）。
+        self.assertIn("failed=1", bodies[0])
+        self.assertIn("修 bug", bodies[0])
+        self.assertIn("做功能", bodies[0])
+        self.assertNotIn("别项目", bodies[0])
+
+    def test_deliver_status_filter_and_limit(self):
+        self._insert("E1", status="failed", prompt="修 bug")
+        self._insert("E2", status="failed", prompt="修另一 bug")
+        self._insert("E3", status="completed", prompt="已做完")
+        worker = self._worker()
+        worker._deliver_marker_tools(worker.get_task("PM"), {
+            "status": "completed",
+            "task_lists": [{"status": "failed", "q": None, "limit": 1}]})
+        bodies = self._followup_bodies()
+        self.assertIn("共 2 条", bodies[0])
+        self.assertIn("还有 1 条未列出", bodies[0])
+        self.assertNotIn("已做完", bodies[0])
+
+    def test_deliver_denied_without_read_permission(self):
+        self._insert("PM2", status="running", created_by="outsider")
+        worker = self._worker()
+        worker._deliver_marker_tools(worker.get_task("PM2"), {
+            "status": "completed",
+            "task_lists": [{"status": None, "q": None, "limit": 20}]})
+        bodies = self._followup_bodies("PM2")
+        self.assertIn("无 read 权限", 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_lists": [{"status": None, "q": None, "limit": 20}]})
+        self.assertEqual(self._followup_bodies("G1"), [])
+
+    def test_injection_project_and_env(self):
+        self._project()
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO environments(id, project_id, name, branch,"
+                " executor, status, error_text, repo_url, created_by,"
+                " created_at, updated_at)"
+                " VALUES('env1', 'p1', '主环境', 'main', 'docker',"
+                " 'ready', NULL, 'git@github.com:openqj/demo.git',"
+                " 'tester', ?, ?)", (time.time(), time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+        worker = self._worker()
+        task = {"id": "PM", "project_id": "p1", "prompt": "做事",
+                "created_by": "tester", "environment_id": "env1"}
+        out = worker._apply_project_injection(task)
+        self.assertIn("演示项目", out["prompt"])
+        self.assertIn("git@github.com:openqj/demo.git", out["prompt"])
+        self.assertIn("主环境", out["prompt"])
+        self.assertIn("task-list", out["prompt"])
+        self.assertEqual(task["prompt"], "做事")
+
+    def test_injection_unbound_repo_and_goal(self):
+        self._project(repo_url=None)
+        worker = self._worker()
+        goal = {"id": "G1", "project_id": "p1", "prompt": "做事",
+                "goal": "做目标", "created_by": "tester",
+                "goal_mode": 1}
+        out = worker._apply_project_injection(goal)
+        self.assertIn("未绑定", out["prompt"])
+        # goal 任务同时改 goal（执行器取 goal 优先于 prompt）。
+        self.assertIn("未绑定", out["goal"])
+        self.assertIn("演示项目", out["goal"])
+
+    def test_injection_skips_coordinator_resume_missing(self):
+        worker = self._worker()
+        coord = {"id": "C1", "project_id": "p1", "prompt": "x",
+                 "coordinator_parts": "a,b"}
+        self.assertIs(worker._apply_project_injection(coord), coord)
+        resume = {"id": "PM", "project_id": "p1", "prompt": "x",
+                  "_resume": True}
+        self.assertIs(worker._apply_project_injection(resume), resume)
+        missing = {"id": "PM", "project_id": "nope", "prompt": "x"}
+        self.assertIs(worker._apply_project_injection(missing), missing)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -154,6 +154,7 @@ 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_task as _format_task
 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
@@ -1741,6 +1742,80 @@ class Worker:
                     task["id"])
         return new_task
 
+    def _apply_project_injection(self, task):
+        """首轮注入项目上下文：只改内存副本，原 prompt/goal 不落库。
+
+        跳过：协调员任务（不跑模型回合）、续跑与断点续跑（已有
+        会话历史）。goal 任务不跳过——项目上下文是数据不是目标，
+        与自动记忆同款同时改内存里的 goal（执行器取 goal 优先
+        于 prompt，只改 prompt 的话 goal 任务看不到）。
+        贴三行数据：项目名、仓库（未绑定时明写未绑定，不编一个
+        地址）、环境（有 environment_id 时写环境名/分支/状态，
+        否则写无）；再加一行 task-list 格式示例，让模型在不知道
+        任务 id 时也能查本项目任务。项目行查不到、查询失败时
+        原样返回，不贴半截上下文、不炸认领。"""
+        if task.get("coordinator_parts"):
+            return task
+        if task.get("_resume") or task.get("_goal_resume_breakpoint"):
+            return task
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                row = conn.execute(
+                    "SELECT name, repo_url FROM projects WHERE id = ?",
+                    (task["project_id"],)).fetchone()
+            finally:
+                conn.close()
+        except Exception as exc:
+            try:
+                self.logger.warning("project injection failed: %s", exc)
+            except Exception:
+                pass
+            return task
+        if row is None:
+            return task
+        try:
+            project_name = row["name"] or task["project_id"]
+        except (KeyError, IndexError, TypeError):
+            project_name = task["project_id"]
+        try:
+            repo_url = row["repo_url"] or None
+        except (KeyError, IndexError, TypeError):
+            repo_url = None
+        env_line = "无（使用默认仓库）"
+        env_id = task.get("environment_id")
+        if env_id:
+            env = self._fetch_environment(env_id)
+            if env is not None:
+                env_line = "%s（分支 %s，状态 %s）" % (
+                    env.get("name") or env_id,
+                    env.get("branch") or "未指定",
+                    env.get("status") or "未知")
+                if env.get("repo_url"):
+                    repo_url = env["repo_url"]
+            else:
+                env_line = "%s（查不到该环境）" % env_id
+        lines = ["[项目上下文]",
+                 "项目：%s（id=%s）" % (project_name,
+                                       task["project_id"]),
+                 "仓库：%s" % (repo_url or "未绑定"),
+                 "环境：%s" % env_line,
+                 "查本项目任务用 task-list 标记块，例如：",
+                 "```task-list",
+                 '{"status": "failed"}',
+                 "```",
+                 "（不传 status 时查全部状态并附各状态计数。）"]
+        prefix = "\n".join(lines)
+        new_task = dict(task)
+        new_task["prompt"] = prefix + "\n\n" + (task.get("prompt") or "")
+        if task.get("goal_mode") or task.get("verify_command"):
+            new_task["goal"] = prefix + "\n\n" + (
+                task.get("goal") or task.get("prompt") or "")
+        new_task["_project_injected"] = True
+        self._audit(task["project_id"], "worker_project_injected",
+                    task["id"])
+        return new_task
+
     def process_auto_memories(self, now=None, force=False,
                               owner_id=None):
         """扫 completed 任务并提取记忆（run_loop 轮次边界调用）。
@@ -3297,12 +3372,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"]):
@@ -3319,8 +3395,10 @@ class Worker:
         todo_lists = result.get("todo_lists") or []
         todo_creates = result.get("todo_creates") or []
         todo_updates = result.get("todo_updates") or []
+        task_lists = result.get("task_lists") or []
         if not (creates or polls or waits or cancels or retries
-                or todo_lists or todo_creates or todo_updates):
+                or todo_lists or todo_creates or todo_updates
+                or task_lists):
             return
         reports = []
         created = 0
@@ -3503,7 +3581,7 @@ class Worker:
         # 一律拒绝，不静默放宽。跨项目待办按“不存在”拒绝；
         # 关联任务 id 跨项目时同样拒绝，避免把别的项目的任务
         # 挂到本项目待办上。
-        if todo_lists or todo_creates or todo_updates:
+        if todo_lists or todo_creates or todo_updates or task_lists:
             read_ok, _read_reason = self._marker_auth(task, "read")
             write_ok, _write_reason = self._marker_auth(
                 task, "create_task")
@@ -3623,6 +3701,52 @@ class Worker:
                         lines.append("（还有 %d 条未列出。）"
                                      % (len(todos) - 20))
                     reports.append("\n".join(lines))
+            for block in task_lists:
+                if read_ok is not True:
+                    reports.append("查任务列表：拒绝（发送方创建"
+                                   "人无 read 权限）。")
+                    rejected += 1
+                    continue
+                try:
+                    conn = _db_connect(self.db_path)
+                    try:
+                        rows = conn.execute(
+                            "SELECT * FROM tasks"
+                            " WHERE project_id = ?"
+                            " ORDER BY created_at DESC",
+                            (task["project_id"],)).fetchall()
+                    finally:
+                        conn.close()
+                except Exception as exc:
+                    reports.append("查任务列表失败：%s" % exc)
+                    rejected += 1
+                    continue
+                tasks = [_task_from_row(row) if not isinstance(
+                    row, dict) else row for row in rows]
+                counts = {}
+                for t in tasks:
+                    counts[t.get("status")] = counts.get(
+                        t.get("status"), 0) + 1
+                if block.get("status"):
+                    tasks = [t for t in tasks
+                             if t.get("status") == block["status"]]
+                if block.get("q"):
+                    ql = block["q"].lower()
+                    tasks = [t for t in tasks
+                             if ql in (t.get("prompt") or "").lower()]
+                if not tasks:
+                    reports.append("任务列表：无匹配项。")
+                else:
+                    lines = ["任务列表（共 %d 条，状态计数：%s）："
+                             % (len(tasks), ", ".join(
+                                 "%s=%d" % kv for kv in sorted(
+                                     counts.items())))]
+                    for t in tasks[:block["limit"]]:
+                        lines.append(_format_task(t))
+                    if len(tasks) > block["limit"]:
+                        lines.append("（还有 %d 条未列出。）"
+                                     % (len(tasks) - block["limit"]))
+                    reports.append("\n".join(lines))
         for poll in polls:
             try:
                 reports.append(self._poll_task_snapshot(task, poll))
@@ -3639,11 +3763,11 @@ 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"
+                  "todo_creates=%d todo_updates=%d task_lists=%d"
                   % (task["id"], created, rejected, len(polls),
                      len(waits), len(cancels), len(retries),
                      len(todo_lists), len(todo_creates),
-                     len(todo_updates)))
+                     len(todo_updates), len(task_lists)))
         self._audit(task["project_id"], "worker_marker_tools", detail)
 
     def _deliver_peer_messages(self, task, result):
@@ -4896,6 +5020,10 @@ class Worker:
         # 待办注入：首轮才注入未完成待办，见
         # _apply_todo_injection 的跳过规则；只改内存副本。
         task = self._apply_todo_injection(task)
+        # 项目注入：首轮才注入项目上下文，见
+        # _apply_project_injection 的跳过规则；只改内存副本；
+        # 最后贴，排在待办与记忆之前（最通用上下文在最前）。
+        task = self._apply_project_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

@@ -266,3 +266,37 @@ Docker 执行器都经 sandbox 的结果解析提取标记块（与 task-message
   不重复贴，协调员与 goal 任务跳过，无未完成待办时不贴空
   章节，注入查询失败时静默跳过、不炸认领。
 - 注入贴的是数据，不是完整工具指南；完整格式只在本文件里。
+
+## 任务列表标记（task-list）与项目注入
+
+### task-list：查本项目任务
+
+```text
+```task-list
+{"status": "failed", "q": "登录", "limit": 20}
+```
+```
+
+- 不传 status 时查全部状态（与 todo-list 默认只查未完成
+  不同——分诊首先要看到失败与运行中的任务），报告附各状态
+  计数；q 是 prompt 子串；limit 钳制 1-50（默认 20），
+  超出写明还有几条；列表按新建倒序。
+- 每轮最多 5 块；status 非法、q 非字符串或超长、limit 非
+  整数时整块跳过；只读，发送方创建人须有项目 read 权限，
+  否则报告里写清拒绝；跨项目任务不出现。
+- 只读不改状态：捡到失败任务后改状态走 task-retry /
+  task-cancel，合并与审批仍由人操作。
+
+### 首轮项目注入（worker _apply_project_injection）
+
+- 普通任务与 goal 任务被认领的首轮，worker 在内存副本前
+  贴三行项目上下文：项目名、仓库（未绑定时明写未绑定，
+  不编地址）、环境（有 environment_id 时写环境名/分支/
+  状态，否则写无）；再加一行 task-list 格式示例。
+- goal 任务同时改内存里的 goal（执行器取 goal 优先于
+  prompt，只改 prompt 的话 goal 任务看不到）；原 prompt /
+  goal 不落库，协调员、续跑与断点续跑跳过，项目行查不到
+  或查询失败时原样返回、不贴半截上下文、不炸认领。
+- 贴的是数据加一行格式示例，不是完整工具指南；完整格式
+  只在本文件里。注入顺序：项目上下文在最前，其次待办、
+  记忆，最后是原 prompt。