feat: agent definition v2 (registry, persistence, developerInstructions channel)

Registry: POST /tasks accepts agent_name, loading <name>.json
from repo-root agents/ (traversal-guarded, skill-pack re-validated);
agent and agent_name together is a 400. Persistence: tasks gains
agent_name/agent_version/agent_instructions (migrate + PG schema),
returned by GET /tasks; stored prompt stays clean. Channel: local
and docker dispatch pass --developer-instructions to host.py
dispatch -> create_thread(developerInstructions=...); run/fork do
not re-send. Cloud executor has no such channel, so the worker
falls back to the v1 marked header in the in-memory dispatch
prompt only (DB prompt stays clean). Goal/coordinator tasks
still reject instructions.

Tests: test_agent_def_v2.py (7 tests) + updated v1 suite.

cloud/agent_def.py

@@ -39,11 +39,33 @@ today, with every boundary written down instead of implied:
   is rejected, so a typo or a v2 idea cannot masquerade as v1 behaviour.
 * Unknown top-level or defaults keys are rejected for the same
   reason: a declarative file must fail on typos, not silently ignore
-  them. Definition identity (name/version/skills) is echoed in the
-  creation response and, via the prompt header, in the stored prompt;
-  v1 adds NO new tasks-table column, so ``GET /tasks`` shows the
-  resolved fields and the header, not a separate ``agent`` column.
-  That column is v2, not something v1 pretends to have.
+  them.
+
+v2 (this file, plus server/worker/sandbox/host changes) closes the
+three v1 boundaries that were explicitly left open:
+
+* Registry: definitions can live as files under repo-root ``agents/``
+  (``default_agents_dir()``) and be referenced by name
+  (``agent_name``) instead of only inline. The server never loads a
+  definition from a caller-supplied path; it only loads
+  ``<name>.json`` under the registry dir, with the traversal guard
+  in ``load_agent_definition``.
+* Persistence: ``tasks`` gains ``agent_name`` / ``agent_version`` /
+  ``agent_instructions`` columns (see ``migrate_tasks_table``), so
+  ``GET /tasks`` shows the definition identity and instructions, and
+  the worker no longer has to reconstruct them from a prompt header.
+* Real channel: for local/docker executors, instructions travel as
+  ``--developer-instructions`` on the host.py top-level command,
+  through ``sandbox`` into ``host.py dispatch`` ->
+  ``create_thread(developerInstructions=...)``. The stored prompt
+  stays clean. For the generic cloud executor, which has no such
+  channel, the worker falls back to the v1 marked header in the
+  in-memory dispatch prompt only (DB prompt stays clean) -- that
+  fallback is the documented cloud boundary, not a claim that the
+  cloud runner honours developerInstructions. Goal and coordinator
+  tasks still reject instructions in v2: the goal loop and the
+  coordinator parent do not pass instructions through, and silently
+  dropping them there would be worse than rejecting them.
 """
 
 import json
@@ -62,6 +84,12 @@ _TOP_LEVEL_KEYS = ("name", "version", "description", "instructions",
                    "skills", "defaults")
 _DEFAULTS_KEYS = ("model", "effort", "approval", "executor", "plan")
 SKILLS_DIR = Path(__file__).resolve().parent / "skills"
+AGENTS_DIR = Path(__file__).resolve().parent.parent / "agents"
+
+
+def default_agents_dir():
+    """Repo-root ``agents/`` directory (definitions live with code)."""
+    return AGENTS_DIR
 
 
 def list_known_skills(skills_dir=None):

cloud/db.py

@@ -880,6 +880,9 @@ POSTGRES_SCHEMA = (
     "parent_task_id TEXT, "
     "coordinator_part TEXT, "
     "goal_resume_pending INTEGER DEFAULT 0, "
+    "agent_name TEXT, "
+    "agent_version TEXT, "
+    "agent_instructions TEXT, "
     "UNIQUE(project_id, idempotency_key))",
     "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
     # 附件包与任务附件：与 server._ensure_uploads_tables 同款，

cloud/sandbox.py

@@ -351,7 +351,8 @@ class LocalExecutor(Executor):
         self.model = model
 
     def _base_command(self, model=None, effort=None, approval=None,
-                      sandbox=None, model_provider=None):
+                      sandbox=None, model_provider=None,
+                      developer_instructions=None):
         cmd = [sys.executable, self.host_py_path]
         if self.config_path is not None:
             cmd += ["--config", self.config_path]
@@ -367,6 +368,13 @@ class LocalExecutor(Executor):
         if effort:
             cmd += ["--effort", effort]
         cmd += _approval_cli_args(approval)
+        # agent 定义 v2：仅 dispatch_task 传，run/fork/goal
+        # 不传（线程已建好指令，续跑不重复发；fork 靠线程
+        # 分叉继承，goal 不接受指令）。空值不传参，旧命令
+        # 逐字节不变。
+        if developer_instructions:
+            cmd += ["--developer-instructions",
+                    developer_instructions]
         # 显式沙箱放最后：argparse 里后出现的 --sandbox 覆盖前面
         # 的（approval=full 自带 danger-full-access）；server 侧
         # 已拒绝 plan+full 同存，这里是第二道防线。
@@ -545,7 +553,9 @@ class LocalExecutor(Executor):
                                   approval=task.get("approval"),
                                   sandbox=("read-only"
                                            if task.get("plan") else None),
-                                  model_provider=model_provider)
+                                  model_provider=model_provider,
+                                  developer_instructions=task.get(
+                                      "agent_instructions"))
         cmd += ["dispatch", "--name", self._host_name(task),
                 "--task", task["prompt"]]
         repo = task.get("repo") or self.repo
@@ -1162,7 +1172,8 @@ class DockerExecutor(Executor):
 
     def _inner_host_command(self, *args, model=None, effort=None,
                             approval=None, sandbox=None,
-                            model_provider=None):
+                            model_provider=None,
+                            developer_instructions=None):
         """容器内 host.py 命令：python3 /app/host.py [args...]。
 
         config_path 若已给，容器内用 /app/<basename> 指代它——
@@ -1191,6 +1202,10 @@ class DockerExecutor(Executor):
         cmd += _approval_cli_args(approval)
         if sandbox:
             cmd += ["--sandbox", sandbox]
+        # agent 定义 v2：仅 dispatch_task 传，语义同本地版。
+        if developer_instructions:
+            cmd += ["--developer-instructions",
+                    developer_instructions]
         cmd += list(args)
         return cmd
 
@@ -1421,7 +1436,8 @@ class DockerExecutor(Executor):
             "--task", task["prompt"], model=task.get("model"),
             effort=task.get("effort"), approval=task.get("approval"),
             sandbox=("read-only" if task.get("plan") else None),
-            model_provider=model_provider)
+            model_provider=model_provider,
+            developer_instructions=task.get("agent_instructions"))
         repo = task.get("repo")
         if repo:
             inner += ["--repo", repo]

cloud/server.py

@@ -57,8 +57,9 @@ from decimal import Decimal
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import parse_qs, quote, urlparse
 
-from cloud.agent_def import format_instructions_header as _agent_format_header
+from cloud.agent_def import default_agents_dir as _agent_default_dir
 from cloud.agent_def import list_known_skills as _agent_known_skills
+from cloud.agent_def import load_agent_definition as _load_agent_definition
 from cloud.agent_def import parse_agent_definition as _parse_agent_definition
 from cloud.auth import ROLES as _MEMBER_ROLES
 from cloud.auth import AuthStore
@@ -1236,6 +1237,12 @@ def migrate_tasks_table(conn):
         "auto_resume_stall_count": "INTEGER DEFAULT 0",
         "checkpoint_round_at_last_resume": "INTEGER",
         "last_error_signature": "TEXT",
+        # agent 定义 v2：定义身份与指令落库（NULL = 未用定义）。
+        # 老库行缺列时按 NULL 计，worker/server 读时用 _col /
+        # _row_get 容错，不炸。
+        "agent_name": "TEXT",
+        "agent_version": "TEXT",
+        "agent_instructions": "TEXT",
     }
     for name, coltype in new_cols.items():
         if name not in cols:
@@ -1833,6 +1840,11 @@ def _task_to_dict(row, attachments=None, total_tokens=None):
         "parent_task_id": _row_get(row, "parent_task_id"),
         "coordinator_part": _row_get(row, "coordinator_part"),
         "goal_resume_pending": _row_get(row, "goal_resume_pending") or 0,
+        # agent 定义 v2：身份与指令落库后直接回读（老行缺列
+        # 时为 None；指令是定义文本、非密钥，任务详情里可见）。
+        "agent_name": _row_get(row, "agent_name"),
+        "agent_version": _row_get(row, "agent_version"),
+        "agent_instructions": _row_get(row, "agent_instructions"),
         "paused_by_user": _row_get(row, "paused_by_user") or 0,
         "interrupted_reason": _row_get(row, "interrupted_reason"),
         "auto_resume_count": _row_get(row, "auto_resume_count") or 0,
@@ -4301,12 +4313,18 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             return
         if prompt_goal_target:
             prompt = prompt_goal_target
-        # 声明式 agent 定义 v1（见 cloud/agent_def.py）：inline
-        # agent 对象，逐字段填充任务选项；显式 body 字段优先
-        # 于定义，定义本身不新开任务表列，解析出的字段落进
-        # 已有列，instructions 以带标记的头拼进 prompt（见后），
-        # 身份在创建响应里回显。
+        # 声明式 agent 定义 v1/v2（见 cloud/agent_def.py）：
+        # inline agent 对象，或 agent_name（注册表 agents/ 下
+        # 的 <name>.json）；两者不能同时给，歧义不猜。显式
+        # body 字段逐项优先于定义；解析出的字段落进已有列，
+        # 定义身份与指令落进 agent_name/agent_version/
+        # agent_instructions 列（v2），prompt 不再带头。
         definition = None
+        if body.get("agent") is not None \
+                and body.get("agent_name") is not None:
+            self._send_json(400, {
+                "error": "agent and agent_name cannot both be used"})
+            return
         if body.get("agent") is not None:
             if not isinstance(body.get("agent"), dict):
                 self._send_json(400, {"error": "agent must be an object"})
@@ -4318,6 +4336,36 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             except ValueError as exc:
                 self._send_json(400, {"error": str(exc)})
                 return
+        if body.get("agent_name") is not None:
+            agent_name = body.get("agent_name")
+            if not isinstance(agent_name, str) or not agent_name.strip():
+                self._send_json(400, {
+                    "error": "agent_name must be a non-empty string"})
+                return
+            try:
+                definition = _load_agent_definition(
+                    agent_name.strip() + ".json",
+                    agents_dir=_agent_default_dir())
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+            # 注册表定义同样须过技能包校验：load 只校验了
+            # 格式（known_skills=None 时不查存在性），这里
+            # 用同一解析器补存在性校验，校验不过即 400。
+            try:
+                definition = _parse_agent_definition(
+                    {
+                        "name": definition.name,
+                        "version": definition.version,
+                        "description": definition.description,
+                        "instructions": definition.instructions,
+                        "skills": definition.skills,
+                        "defaults": definition.defaults,
+                    },
+                    known_skills=_agent_known_skills())
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
         depends_on = body.get("depends_on") or []
         if isinstance(depends_on, str):
             depends_on = [depends_on]
@@ -4615,18 +4663,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             if not bundle_files:
                 self._send_json(400, {"error": "附件包为空"})
                 return
-        # agent instructions：goal 与协调员任务不接受（见
-        # agent_def.py 的边界说明）。非这两种时以带标记的头
-        # 拼进 prompt，存库的 prompt 即带头的 prompt，worker
-        # 派发、回读看到的都是同一份；子任务派生默认值解析
-        # 在前面已用原始 prompt 跑过，不受头影响。
+        # agent instructions v2：goal 与协调员任务仍不接受
+        # （goal 循环与协调员父任务不传 instructions，见
+        # agent_def.py 的 v2 边界）。非这两种时指令不拼进
+        # prompt，存进 agent_instructions 列，由 worker 按
+        # executor 选通道派发（local/docker 走
+        # --developer-instructions，cloud 走内存里的头回落）。
         if definition is not None and definition.instructions:
             if goal_mode or verify_command or coord_parts:
                 self._send_json(400, {
                     "error": "agent instructions cannot be used with"
                              " goal/coordinator tasks"})
                 return
-            prompt = _agent_format_header(definition, prompt)
+        agent_name_val = definition.name if definition else None
+        agent_version_val = definition.version if definition else None
+        agent_instructions_val = (
+            definition.instructions if definition else None)
         now = time.time()
         conn = self._db()
         try:
@@ -4651,16 +4703,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     " model, effort, approval, plan, goal_mode,"
                     " environment_id, coordinator_parts,"
                     " parent_task_id, coordinator_part,"
-                    " child_model, child_effort)"
+                    " child_model, child_effort,"
+                    " agent_name, agent_version, agent_instructions)"
                     " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?,"
-                    " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                     (task_id, project_id, prompt, idempotency_key,
                      json.dumps(depends_on), priority, user_id, now, now,
                      executor, goal, verify_command, goal_max_rounds,
                      model, effort, approval, plan, goal_mode,
                      environment_id,
                      json.dumps(coord_parts) if coord_parts else None,
-                     None, None, child_model, child_effort),
+                     None, None, child_model, child_effort,
+                     agent_name_val, agent_version_val,
+                     agent_instructions_val),
                 )
                 # 长任务不限时：API 字段 no_time_limit=1 时写标记，
                 # worker 认领后按不限时跑（单轮 host 600 秒仍在）；
@@ -4714,8 +4769,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 len(coord_parts)
                 * COORDINATOR_RESERVE_PER_PART_USD_MICROS)
         if definition is not None:
-            # 仅新建任务的 201 响应回显定义身份；幂等返回的
-            # 已有任务不补这个字段（那是另一条创建记录）。
+            # 201 响应回显定义身份与技能（v1 兼容）；幂等返回
+            # 的已有任务靠 _task_to_dict 里的 agent_name /
+            # agent_version / agent_instructions（v2 落库后
+            # 回读即有），不补 agent 汇总字段。
             task["agent"] = definition.to_summary()
         self._send_json(201, task)
 

cloud/tests/test_agent_def.py

@@ -1,4 +1,6 @@
-"""声明式 agent 定义 v1 单测：解析、校验、解析到任务字段、建任务入口。
+"""声明式 agent 定义 v1/v2 单测：解析、校验、解析到任务字段、建任务入口。
+
+v2 改动：prompt 不再拼头，指令落库（见 test_agent_def_v2.py）。
 
 运行（在仓库根目录下）：
     python3 -m unittest discover -s cloud/tests -p "test_agent_def.py" -v
@@ -168,7 +170,9 @@ class ServerAgentTestCase(unittest.TestCase):
         self.assertEqual(status, 200)
         return data["tasks"]
 
-    def test_agent_fills_fields_and_header(self):
+    def test_agent_fills_fields_and_stores_instructions(self):
+        # v2：prompt 保持干净，指令落库，不再拼头（v1 行为已
+        # 被 v2 替换，见 agent_def.py 的 v2 边界说明）。
         status, task = self._req("POST", "/tasks", {
             "project_id": self.project["id"], "prompt": "do research",
             "agent": _agent()}, self.owner_key)
@@ -179,10 +183,14 @@ class ServerAgentTestCase(unittest.TestCase):
         self.assertEqual(task["executor"], "docker")
         self.assertEqual(task["agent"]["name"], "research-brief")
         self.assertEqual(task["agent"]["skills"], ["live-info"])
+        self.assertEqual(task["agent_name"], "research-brief")
+        self.assertEqual(task["agent_version"], "v1")
+        self.assertEqual(task["agent_instructions"],
+                         "Cite sources and dates.")
         stored = self._list_tasks()[0]
-        self.assertTrue(stored["prompt"].startswith(
-            "[Agent: research-brief v1]\nCite sources and dates.\n\n"))
-        self.assertTrue(stored["prompt"].endswith("do research"))
+        self.assertEqual(stored["prompt"], "do research")
+        self.assertEqual(stored["agent_instructions"],
+                         "Cite sources and dates.")
 
     def test_explicit_body_wins_over_agent(self):
         status, task = self._req("POST", "/tasks", {

cloud/tests/test_agent_def_v2.py

@@ -0,0 +1,184 @@
+"""声明式 agent 定义 v2 单测：注册表、落库、developerInstructions 通道。
+
+运行（在仓库根目录下）：
+    python3 -m unittest discover -s cloud/tests -p "test_agent_def_v2.py" -v
+
+覆盖：agent_name 查注册表、agent 与 agent_name 同给 400、
+未知 agent_name 400、本地/docker 派发命令含
+--developer-instructions、无指令时命令不变、host dispatch 把
+指令传给 create_thread 的 instructions（即 developerInstructions）。
+不跑真实模型、不联网；host 部分用假 server，只断言传参。
+"""
+
+import http.client
+import importlib.util
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.agent_def import default_agents_dir
+    from cloud.auth import AuthStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.agent_def import default_agents_dir
+    from cloud.auth import AuthStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server
+
+
+def _load_host():
+    spec = importlib.util.spec_from_file_location(
+        "hostmod_v2", str(Path(__file__).resolve().parents[2] / "host.py"))
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class ServerAgentNameTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        headers = {"Content-Type": "application/json"}
+        if key:
+            headers["X-API-Key"] = key
+        conn.request(method, path,
+                     json.dumps(body) if body is not None else None, headers)
+        resp = conn.getresponse()
+        raw = resp.read()
+        conn.close()
+        try:
+            data = json.loads(raw)
+        except ValueError:
+            data = None
+        return resp.status, data
+
+    def _list_tasks(self):
+        status, data = self._req(
+            "GET", "/tasks?project_id=" + self.project["id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        return data["tasks"]
+
+    def test_agent_name_from_registry(self):
+        # 注册表里有 agents/research-brief.json（仓库自带示例）。
+        self.assertTrue((default_agents_dir()
+                         / "research-brief.json").is_file())
+        status, task = self._req("POST", "/tasks", {
+            "project_id": self.project["id"], "prompt": "do research",
+            "agent_name": "research-brief"}, self.owner_key)
+        self.assertEqual(status, 201, task)
+        self.assertEqual(task["agent_name"], "research-brief")
+        self.assertEqual(task["agent_version"], "v1")
+        self.assertEqual(task["effort"], "medium")
+        self.assertEqual(task["executor"], "docker")
+        self.assertEqual(task["prompt"], "do research")
+        stored = self._list_tasks()[0]
+        self.assertEqual(stored["agent_instructions"][:20],
+                         "Confirm the question")
+
+    def test_agent_and_agent_name_conflict(self):
+        status, _data = self._req("POST", "/tasks", {
+            "project_id": self.project["id"], "prompt": "do research",
+            "agent": {"name": "research-brief"},
+            "agent_name": "research-brief"}, self.owner_key)
+        self.assertEqual(status, 400)
+        self.assertEqual(self._list_tasks(), [])
+
+    def test_unknown_agent_name(self):
+        status, _data = self._req("POST", "/tasks", {
+            "project_id": self.project["id"], "prompt": "do research",
+            "agent_name": "no-such-agent"}, self.owner_key)
+        self.assertEqual(status, 400)
+        self.assertEqual(self._list_tasks(), [])
+
+
+class InstructionsChannelTestCase(unittest.TestCase):
+    def test_local_dispatch_includes_instructions(self):
+        captured = []
+
+        def runner(cmd, env=None):
+            captured.append(cmd)
+            return {"returncode": 0, "stdout": "ok", "stderr": ""}
+
+        ex = LocalExecutor(host_py_path="host.py", runner=runner,
+                           repo="/tmp/repo")
+        ex.dispatch_task({"id": "t1", "prompt": "hi",
+                          "agent_instructions": "be careful"})
+        self.assertIn("--developer-instructions", captured[0])
+        idx = captured[0].index("--developer-instructions")
+        self.assertEqual(captured[0][idx + 1], "be careful")
+        self.assertLess(idx, captured[0].index("dispatch"))
+
+    def test_local_dispatch_without_instructions_unchanged(self):
+        captured = []
+
+        def runner(cmd, env=None):
+            captured.append(cmd)
+            return {"returncode": 0, "stdout": "ok", "stderr": ""}
+
+        ex = LocalExecutor(host_py_path="host.py", runner=runner,
+                           repo="/tmp/repo")
+        ex.dispatch_task({"id": "t1", "prompt": "hi"})
+        self.assertNotIn("--developer-instructions", captured[0])
+
+    def test_docker_inner_command_includes_instructions(self):
+        ex = DockerExecutor.__new__(DockerExecutor)
+        ex.config_path = None
+        ex.model = None
+        ex.ignore_host_budget = False
+        cmd = ex._inner_host_command(
+            "dispatch", model="m",
+            developer_instructions="be careful")
+        self.assertIn("--developer-instructions", cmd)
+        self.assertEqual(cmd[cmd.index("--developer-instructions") + 1],
+                         "be careful")
+
+    def test_host_dispatch_passes_instructions(self):
+        host = _load_host()
+
+        class FakeServer:
+            def __init__(self):
+                self.calls = []
+
+            def create_thread(self, cwd, model=None, sandbox=None,
+                              approval_policy=None, instructions=None):
+                self.calls.append(instructions)
+                return "thr1"
+
+        srv = FakeServer()
+        state = {"tasks": {}}
+        with tempfile.TemporaryDirectory() as tmpdir:
+            host.dispatch(srv, state, "t1", "hi", tmpdir, "m", "low",
+                          None, None, use_worktree=False,
+                          developer_instructions="be careful")
+        self.assertEqual(srv.calls, ["be careful"])

cloud/worker.py

@@ -134,6 +134,8 @@ import time
 import uuid
 from types import SimpleNamespace
 
+from cloud.agent_def import AgentDefinition as _AgentDefinition
+from cloud.agent_def import format_instructions_header as _agent_header
 from cloud.auth import AuthStore
 from cloud.coordinator import build_report as _coord_build_report
 from cloud.coordinator import DEFAULT_POLL_INTERVAL as _COORD_POLL_INTERVAL
@@ -412,6 +414,10 @@ def _task_from_row(row):
         "coordinator_parts": _coord_parse_parts(_col("coordinator_parts")),
         "parent_task_id": _col("parent_task_id"),
         "coordinator_part": _col("coordinator_part"),
+        # agent 定义 v2：老行缺列时 None = 未用定义，不炸。
+        "agent_name": _col("agent_name"),
+        "agent_version": _col("agent_version"),
+        "agent_instructions": _col("agent_instructions"),
     }
 
 
@@ -5864,6 +5870,21 @@ class Worker:
                                 task_id, decision.tier, decision.source,
                                 decision.applied))
             task["model"] = decision.chosen_model or self.model
+        # agent 指令通道（v2）：local/docker 由 sandbox 的
+        # --developer-instructions 派发，prompt 保持干净；通用
+        # cloud 执行器没有这个通道，只能在这里把 v1 的带标记
+        # 头拼进内存里的派发 prompt（只改本轮 task 副本，不写
+        # 库，库 prompt 保持干净），否则 cloud 任务会比 v1
+        # 倒退成指令全丢。已带头的 prompt 不重复拼。
+        if task.get("executor") == "cloud" \
+                and task.get("agent_instructions") \
+                and not (task.get("prompt") or "").startswith("[Agent:"):
+            task["prompt"] = _agent_header(
+                _AgentDefinition(
+                    name=task.get("agent_name") or "agent",
+                    version=task.get("agent_version"),
+                    instructions=task["agent_instructions"]),
+                task.get("prompt") or "")
         actual_model = task["model"]
         # 有限重试与真实 fallback 的优先级：取消不重试；一次尝试失败
         # 且可 fallback、且显式配了 fallback_models 时，优先真换模型

host.py

@@ -1803,7 +1803,8 @@ def _seed_worktree(cwd, seed_dir):
 
 def dispatch(server, state, name, prompt, repo, model, effort, sandbox,
              approval_policy, use_worktree=True, parent_task=None,
-             depends_on=None, priority=0, max_retries=0, seed_dir=None):
+             depends_on=None, priority=0, max_retries=0, seed_dir=None,
+             developer_instructions=None):
     if name in state["tasks"]:
         raise RuntimeError(f"task 已存在: {name}，先换名或 discard 它")
     if repo and not os.path.isabs(repo) and not os.path.exists(repo):
@@ -1823,8 +1824,10 @@ def dispatch(server, state, name, prompt, repo, model, effort, sandbox,
             raise RuntimeError(
                 "seed-dir 需要 worktree，当前派发没有建 worktree")
         _seed_worktree(cwd, seed_dir)
-    thread_id = server.create_thread(cwd, model=model, sandbox=sandbox,
-                                     approval_policy=approval_policy)
+    thread_id = server.create_thread(
+        cwd, model=model, sandbox=sandbox,
+        approval_policy=approval_policy,
+        instructions=developer_instructions)
     task = {"name": name, "thread_id": thread_id, "cwd": cwd,
             "repo": os.path.abspath(repo) if repo else None,
             "branch": branch, "sandbox": sandbox, "status": "running",
@@ -2247,6 +2250,10 @@ def main():
     ap.add_argument("--sandbox", default=None)
     ap.add_argument("--approval-policy", default=None)
     ap.add_argument("--approval", default=None, choices=["auto", "deny"])
+    ap.add_argument("--developer-instructions", default=None,
+                    help="agent 定义 v2：派发时传给 codex 的"
+                         " developerInstructions（只在 dispatch 建"
+                         " 线程时用，run/goal/fork 不用它）")
     ap.add_argument("--turn-timeout", type=int, default=None,
                     help="单轮等待上限（秒）；0 表示不限时、对齐 Codex App "
                          "等 turn/completed 的语义；不传时用配置 "
@@ -2709,7 +2716,9 @@ def main():
                             use_worktree=not args.no_worktree,
                             depends_on=deps, priority=args.priority,
                             max_retries=args.max_retries,
-                            seed_dir=args.seed_dir)
+                            seed_dir=args.seed_dir,
+                            developer_instructions=(
+                                args.developer_instructions))
             # 生产教训：先落盘任务登记再跑 turn，中途被杀/超时后
             # 任务还在 host-state 里，可用 run 续跑；否则 turn 中断
             # 就等于任务丢了（2026-10-25 委派评审 300 秒超时实测）。