fix: M1-M4 修复 Protocol 签名与设计文档对齐 + §14 去重
CI / lint (pull_request) Failing after 7s
CI / test (pull_request) Has been skipped
CI / notify-on-failure (pull_request) Successful in 3s

This commit is contained in:
cfdaily
2026-06-10 15:40:12 +08:00
parent 86504faf1f
commit 0e4d12898d
2 changed files with 21 additions and 454 deletions
+10 -45
View File
@@ -9,7 +9,10 @@ from __future__ import annotations
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Protocol, runtime_checkable
if TYPE_CHECKING:
from src.daemon.prompt_composer import PromptContext
logger = logging.getLogger("moziplus-v2.registry")
@@ -26,35 +29,12 @@ class TaskTypeHandler(Protocol):
task_type: str # 类型标识:'task' | 'mail' | 'toolchain'
virtual_project: Optional[str] # 虚拟项目 ID,如 '_mail'、'_toolchain'。普通任务为 None
def build_prompt(
self,
task_id: str,
title: str,
description: str,
must_haves: str,
project_id: str,
agent_id: str,
task: Optional[Dict] = None,
spawn_type: str = "executor",
spawner: Any = None,
) -> str:
"""构建 Agent prompt。"""
def build_prompt(self, context: "PromptContext") -> str:
"""构建 Agent prompt(通过 PromptComposer 拼 section)。"""
...
def build_api_section(
self, project_id: str, task_id: str, agent_id: str
) -> str:
"""构建 API 操作指令(success_status 等)。"""
...
def skip_guardrail(self, project_id: str) -> bool:
"""是否跳过 guardrail 检查。"""
...
def pre_spawn(
self, task_id: str, db_path: Path, dispatcher: Any
) -> Optional[Any]:
"""spawn 前回调,返回 on_checks_passed 回调或 None。"""
def pre_spawn(self, task_id: str, db_path: Path) -> bool:
"""spawn 前业务准备。默认 Truemail/toolchain override 为 auto_working。"""
...
def post_complete(
@@ -63,27 +43,12 @@ class TaskTypeHandler(Protocol):
agent_id: str,
outcome: str,
db_path: Path,
must_haves: str,
dispatcher: Any,
) -> None:
"""spawn 完成后回调"""
...
def build_retry_prompt(
self,
task_id: str,
agent_id: str,
retry_count: int,
max_retries: int,
retry_field: str,
task_info: Dict,
spawner: Any,
) -> str:
"""构建重试 prompt。"""
"""spawn 完成后的业务处理。统一流程:crash→verify→mark→notify"""
...
def check_completion(self, task_id: str, db_path: Path) -> bool:
"""检查任务是否已完成(如 mail 的回复检查"""
"""ticker 级别的完成检查。"""
...
def get_sections(self) -> list: