diff --git a/cloud/marker_tools.py b/cloud/marker_tools.py index 0b1afd9..72ff897 100644 --- a/cloud/marker_tools.py +++ b/cloud/marker_tools.py @@ -1,6 +1,6 @@ """任务内标记工具（marker tools）：task-create / task-poll / task-read / wait-threads / task-cancel / task-retry / todo-list / todo-create / -todo-update。 +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 每轮最多 - 2 块，每块最多 8 个目标。8 这个数是刻意与服务端 + todo-create、todo-update、task-list 各最多 5 块；wait + 每轮最多 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 diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 0aaa0eb..0157522 100644 --- a/cloud/sandbox.py +++ b/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, diff --git a/cloud/tests/test_task_list_markers.py b/cloud/tests/test_task_list_markers.py new file mode 100644 index 0000000..e156667 --- /dev/null +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index 4603375..e990838 100644 --- a/cloud/worker.py +++ b/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 四个 diff --git a/docs/marker-tools.md b/docs/marker-tools.md index a1ffa4c..ad9b121 100644 --- a/docs/marker-tools.md +++ b/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。