feat(task-progress): 增加版本化任务推进模型和统一协议命令服务

新增严格的 TaskProgress、TaskProgressRevision 与任务合同引用模型，落实阶段、问题、阻塞、发现、假设和工作项的容量及字段约束。

以 append-only revision、父版本链和 expected_revision 实现可分叉的乐观并发更新，并通过 TaskProtocolService 的 per-trace 异步锁统一读校验写持久化边界。

新增 update_task_progress 生命周期独占工具，禁止 Validator、SideBranch、Legacy、revision 2、已提交报告及待审核状态修改推进记录；TaskReport 同时固化精确的 Progress revision。

cyber_agent/core/task_protocol.py

@@ -1,4 +1,4 @@
-"""Recursive revision 2 的父子任务协议与持久化状态。
+"""Recursive revision 3 的父子任务协议与持久化状态。
 
 Sub-Agent 使用 TaskBrief 下发任务，子 Agent 通过 TaskReport 上报，父 Agent 使用
 TaskReview 审核。Runner 和两个协议工具共同把待审、后续动作和回退状态保存在 Trace context。
@@ -8,6 +8,7 @@ from __future__ import annotations
 
 from copy import deepcopy
 from datetime import datetime
+from hashlib import sha256
 import json
 from typing import Any, Literal
 
@@ -29,8 +30,26 @@ ReviewDecision = Literal[
     "FAIL",
 ]
 TaskValidationScope = Literal["evidence", "hypothesis", "output", "task"]
+TaskProgressPhase = Literal[
+    "planning",
+    "investigating",
+    "forming_hypothesis",
+    "executing",
+    "producing_output",
+    "ready_to_submit",
+    "blocked",
+]
 _VALIDATION_SCOPE_ORDER = ("evidence", "hypothesis", "output", "task")
 
+TASK_PROTOCOL_SCHEMA_VERSION = 1
+MAX_TASK_PROGRESS_ITEMS_PER_KIND = 20
+MAX_TASK_PROGRESS_ITEMS = 60
+MAX_TASK_PROGRESS_CHARS = 24_000
+MAX_TASK_PROGRESS_REVISIONS = 64
+MAX_TASK_PROGRESS_HISTORY_CHARS = 512_000
+MAX_TASK_PROGRESS_CONTEXT_REFS = 8
+MAX_TASK_PROGRESS_ARTIFACT_REFS = 20
+
 
 class StrictProtocolModel(BaseModel):
     """协议模型的统一严格基类，拒绝未定义字段并清理字符串空白。"""
@@ -232,6 +251,386 @@ class TaskReview(StrictProtocolModel):
         return self
 
 
+class TaskContractRef(StrictProtocolModel):
+    """TaskProgress 绑定的不可变根目标或 TaskBrief 合同。"""
+
+    kind: Literal["root_task_anchor", "task_brief"]
+    root_task_anchor_hash: str | None = Field(
+        default=None,
+        pattern=r"^[0-9a-f]{64}$",
+    )
+    task_brief_version: int | None = Field(default=None, ge=1)
+    task_brief_hash: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
+
+    @model_validator(mode="after")
+    def validate_kind_fields(self) -> "TaskContractRef":
+        if self.kind == "root_task_anchor":
+            if not self.root_task_anchor_hash:
+                raise ValueError("root_task_anchor requires root_task_anchor_hash")
+            if self.task_brief_version is not None or self.task_brief_hash is not None:
+                raise ValueError("root_task_anchor does not accept TaskBrief fields")
+        else:
+            if self.task_brief_version is None or not self.task_brief_hash:
+                raise ValueError("task_brief requires version and hash")
+            if self.root_task_anchor_hash is not None:
+                raise ValueError("task_brief does not accept root_task_anchor_hash")
+        return self
+
+
+class ProgressRefs(StrictProtocolModel):
+    """任务推进条目可引用的现有上下文与产物句柄。"""
+
+    context_refs: list[ContextRef] = Field(default_factory=list, max_length=8)
+    artifact_refs: list[ArtifactRef] = Field(default_factory=list, max_length=20)
+
+
+class Question(ProgressRefs):
+    item_id: str = Field(min_length=1, max_length=100)
+    text: str = Field(min_length=1, max_length=2_000)
+    state: Literal["open", "answered", "deferred"] = "open"
+    answer: str | None = Field(default=None, min_length=1, max_length=2_000)
+
+    @model_validator(mode="after")
+    def validate_answer(self) -> "Question":
+        if self.state == "answered" and not self.answer:
+            raise ValueError("answered question requires answer")
+        if self.state != "answered" and self.answer is not None:
+            raise ValueError("only answered question accepts answer")
+        return self
+
+
+class Blocker(ProgressRefs):
+    item_id: str = Field(min_length=1, max_length=100)
+    text: str = Field(min_length=1, max_length=2_000)
+    state: Literal["active", "resolved"] = "active"
+    resolution: str | None = Field(default=None, min_length=1, max_length=2_000)
+
+    @model_validator(mode="after")
+    def validate_resolution(self) -> "Blocker":
+        if self.state == "resolved" and not self.resolution:
+            raise ValueError("resolved blocker requires resolution")
+        if self.state != "resolved" and self.resolution is not None:
+            raise ValueError("only resolved blocker accepts resolution")
+        return self
+
+
+class Finding(ProgressRefs):
+    item_id: str = Field(min_length=1, max_length=100)
+    statement: str = Field(min_length=1, max_length=2_000)
+    basis: str = Field(min_length=1, max_length=2_000)
+
+
+class Hypothesis(ProgressRefs):
+    item_id: str = Field(min_length=1, max_length=100)
+    statement: str = Field(min_length=1, max_length=2_000)
+    state: Literal["open", "supported", "rejected"] = "open"
+    rationale: str | None = Field(default=None, min_length=1, max_length=2_000)
+
+    @model_validator(mode="after")
+    def validate_rationale(self) -> "Hypothesis":
+        if self.state in {"supported", "rejected"} and not self.rationale:
+            raise ValueError(f"{self.state} hypothesis requires rationale")
+        if self.state == "open" and self.rationale is not None:
+            raise ValueError("open hypothesis does not accept rationale")
+        return self
+
+
+class WorkItem(ProgressRefs):
+    item_id: str = Field(min_length=1, max_length=100)
+    description: str = Field(min_length=1, max_length=2_000)
+    state: Literal["pending", "in_progress", "done", "abandoned"] = "pending"
+    result_summary: str | None = Field(default=None, min_length=1, max_length=2_000)
+
+    @model_validator(mode="after")
+    def validate_result(self) -> "WorkItem":
+        if self.state == "done" and not self.result_summary:
+            raise ValueError("done work item requires result_summary")
+        if self.state in {"pending", "in_progress"} and self.result_summary is not None:
+            raise ValueError(f"{self.state} work item does not accept result_summary")
+        return self
+
+
+class TaskProgress(StrictProtocolModel):
+    """Agent 提交的当前 Task 完整推进快照。"""
+
+    phase: TaskProgressPhase
+    questions: list[Question] = Field(
+        default_factory=list,
+        max_length=MAX_TASK_PROGRESS_ITEMS_PER_KIND,
+    )
+    blockers: list[Blocker] = Field(
+        default_factory=list,
+        max_length=MAX_TASK_PROGRESS_ITEMS_PER_KIND,
+    )
+    findings: list[Finding] = Field(
+        default_factory=list,
+        max_length=MAX_TASK_PROGRESS_ITEMS_PER_KIND,
+    )
+    hypotheses: list[Hypothesis] = Field(
+        default_factory=list,
+        max_length=MAX_TASK_PROGRESS_ITEMS_PER_KIND,
+    )
+    work_items: list[WorkItem] = Field(
+        default_factory=list,
+        max_length=MAX_TASK_PROGRESS_ITEMS_PER_KIND,
+    )
+    decision_rationale: str = Field(default="", max_length=2_000)
+
+    @model_validator(mode="after")
+    def validate_progress_limits(self) -> "TaskProgress":
+        groups = (
+            self.questions,
+            self.blockers,
+            self.findings,
+            self.hypotheses,
+            self.work_items,
+        )
+        items = [item for group in groups for item in group]
+        if len(items) > MAX_TASK_PROGRESS_ITEMS:
+            raise ValueError(
+                f"TaskProgress accepts at most {MAX_TASK_PROGRESS_ITEMS} total items"
+            )
+        item_ids = [item.item_id for item in items]
+        if len(set(item_ids)) != len(item_ids):
+            raise ValueError("TaskProgress item_id values must be globally unique")
+        context_refs = {
+            (ref.ref_id, ref.version)
+            for item in items
+            for ref in item.context_refs
+        }
+        artifact_refs = {
+            (ref.artifact_id, ref.version, ref.content_hash)
+            for item in items
+            for ref in item.artifact_refs
+        }
+        if len(context_refs) > MAX_TASK_PROGRESS_CONTEXT_REFS:
+            raise ValueError(
+                f"TaskProgress accepts at most {MAX_TASK_PROGRESS_CONTEXT_REFS} ContextRefs"
+            )
+        if len(artifact_refs) > MAX_TASK_PROGRESS_ARTIFACT_REFS:
+            raise ValueError(
+                f"TaskProgress accepts at most {MAX_TASK_PROGRESS_ARTIFACT_REFS} ArtifactRefs"
+            )
+        encoded = _canonical_json(self.model_dump(mode="json"))
+        if len(encoded) > MAX_TASK_PROGRESS_CHARS:
+            raise ValueError(
+                f"TaskProgress exceeds {MAX_TASK_PROGRESS_CHARS} characters"
+            )
+        return self
+
+
+class TaskProgressRevision(TaskProgress):
+    """框架注入版本、时间线与合同引用后的持久化记录。"""
+
+    schema_version: Literal[1] = 1
+    revision: int = Field(ge=1)
+    parent_revision: int | None = Field(default=None, ge=1)
+    effective_at_sequence: int = Field(ge=0)
+    task_contract_ref: TaskContractRef
+
+
+def _canonical_json(value: Any) -> str:
+    try:
+        return json.dumps(
+            value,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"Task protocol data must be JSON serializable: {exc}") from exc
+
+
+def task_brief_hash(task_brief: TaskBrief | dict[str, Any]) -> str:
+    brief = TaskBrief.model_validate(task_brief)
+    return sha256(
+        _canonical_json(brief.model_dump(mode="json")).encode("utf-8")
+    ).hexdigest()
+
+
+def task_contract_ref(
+    state: dict[str, Any],
+    *,
+    root_task_anchor_hash: str | None = None,
+) -> TaskContractRef:
+    brief = state.get("task_brief")
+    if brief is None:
+        if not root_task_anchor_hash:
+            raise ValueError("root TaskProgress requires root_task_anchor_hash")
+        return TaskContractRef(
+            kind="root_task_anchor",
+            root_task_anchor_hash=root_task_anchor_hash,
+        )
+    version = state.get("task_brief_version")
+    return TaskContractRef(
+        kind="task_brief",
+        task_brief_version=version,
+        task_brief_hash=task_brief_hash(brief),
+    )
+
+
+def current_task_progress(state: dict[str, Any]) -> TaskProgressRevision | None:
+    head = state.get("task_progress_head_revision")
+    if head is None:
+        return None
+    matches = [
+        item
+        for item in state.get("task_progress_revisions", [])
+        if item.get("revision") == head
+    ]
+    if len(matches) != 1:
+        raise ValueError("TaskProgress head does not reference exactly one revision")
+    return TaskProgressRevision.model_validate(matches[0])
+
+
+def task_progress_at_revision(
+    state: dict[str, Any],
+    revision: int | None,
+) -> TaskProgressRevision | None:
+    if revision is None:
+        return None
+    matches = [
+        raw
+        for raw in state.get("task_progress_revisions", [])
+        if raw.get("revision") == revision
+    ]
+    if len(matches) != 1:
+        raise ValueError(
+            f"TaskProgress revision {revision} does not reference exactly one record"
+        )
+    return TaskProgressRevision.model_validate(matches[0])
+
+
+def task_progress_artifact_refs(
+    progress: TaskProgress | TaskProgressRevision | None,
+) -> list[ArtifactRef]:
+    if progress is None:
+        return []
+    unique: dict[tuple[str, str, str], ArtifactRef] = {}
+    for item in (
+        *progress.questions,
+        *progress.blockers,
+        *progress.findings,
+        *progress.hypotheses,
+        *progress.work_items,
+    ):
+        for ref in item.artifact_refs:
+            unique[(ref.artifact_id, ref.version, ref.content_hash)] = ref
+    return list(unique.values())
+
+
+def append_task_progress_revision(
+    state: dict[str, Any],
+    progress: TaskProgress | dict[str, Any],
+    *,
+    expected_revision: int,
+    effective_at_sequence: int,
+    contract_ref: TaskContractRef,
+) -> TaskProgressRevision:
+    """校验并追加一个完整快照；校验失败时不修改 state。"""
+    revisions = list(state.get("task_progress_revisions", []))
+    head = state.get("task_progress_head_revision")
+    current_revision = int(head or 0)
+    if expected_revision != current_revision:
+        raise ValueError(
+            f"stale TaskProgress revision: expected={expected_revision}, current={current_revision}"
+        )
+    if len(revisions) >= MAX_TASK_PROGRESS_REVISIONS:
+        raise ValueError(
+            f"TaskProgress accepts at most {MAX_TASK_PROGRESS_REVISIONS} revisions"
+        )
+    parsed = TaskProgress.model_validate(progress)
+    known_revisions = [int(item.get("revision", 0)) for item in revisions]
+    next_revision = max(known_revisions, default=0) + 1
+    revision = TaskProgressRevision(
+        **parsed.model_dump(mode="json"),
+        revision=next_revision,
+        parent_revision=head,
+        effective_at_sequence=max(0, int(effective_at_sequence)),
+        task_contract_ref=contract_ref,
+    )
+    candidate = [*revisions, revision.model_dump(mode="json")]
+    if len(_canonical_json(candidate)) > MAX_TASK_PROGRESS_HISTORY_CHARS:
+        raise ValueError(
+            f"TaskProgress history exceeds {MAX_TASK_PROGRESS_HISTORY_CHARS} characters"
+        )
+    state["task_progress_revisions"] = candidate
+    state["task_progress_head_revision"] = revision.revision
+    return revision
+
+
+def initialize_task_progress(
+    state: dict[str, Any],
+    *,
+    root_task_anchor_hash: str | None = None,
+    effective_at_sequence: int = 0,
+) -> TaskProgressRevision:
+    existing = current_task_progress(state)
+    if existing is not None:
+        return existing
+    return append_task_progress_revision(
+        state,
+        TaskProgress(phase="planning"),
+        expected_revision=0,
+        effective_at_sequence=effective_at_sequence,
+        contract_ref=task_contract_ref(
+            state,
+            root_task_anchor_hash=root_task_anchor_hash,
+        ),
+    )
+
+
+def task_progress_readiness_error(state: dict[str, Any]) -> str | None:
+    progress = current_task_progress(state)
+    if progress is None:
+        return "TaskProgress is required"
+    if progress.phase != "ready_to_submit":
+        return "TaskProgress phase must be ready_to_submit"
+    if any(item.state == "open" for item in progress.questions):
+        return "TaskProgress has open questions"
+    if any(item.state == "active" for item in progress.blockers):
+        return "TaskProgress has active blockers"
+    if any(item.state == "open" for item in progress.hypotheses):
+        return "TaskProgress has open hypotheses"
+    if any(item.state in {"pending", "in_progress"} for item in progress.work_items):
+        return "TaskProgress has unfinished work items"
+    return None
+
+
+def rewind_task_progress(state: dict[str, Any], cutoff: int) -> int | None:
+    """只沿当前 head 的祖先链回溯，不删除其他分支。"""
+    by_revision = {
+        item.revision: item
+        for item in (
+            TaskProgressRevision.model_validate(raw)
+            for raw in state.get("task_progress_revisions", [])
+        )
+    }
+    head = state.get("task_progress_head_revision")
+    visited: set[int] = set()
+    while head is not None:
+        if head in visited or head not in by_revision:
+            raise ValueError("TaskProgress revision ancestry is corrupt")
+        visited.add(head)
+        revision = by_revision[head]
+        if revision.effective_at_sequence <= cutoff:
+            break
+        head = revision.parent_revision
+    state["task_progress_head_revision"] = head
+    retained_ancestry: set[int] = set()
+    cursor = head
+    while cursor is not None:
+        if cursor in retained_ancestry or cursor not in by_revision:
+            raise ValueError("TaskProgress revision ancestry is corrupt")
+        retained_ancestry.add(cursor)
+        cursor = by_revision[cursor].parent_revision
+    report_revision = state.get("task_report_progress_revision")
+    if report_revision is not None and report_revision not in retained_ancestry:
+        state["task_report_progress_revision"] = None
+    return head
+
+
 def new_task_protocol(task_brief: TaskBrief | dict | None = None) -> dict[str, Any]:
     """创建一份完整的 Recursive 任务协议初始状态。
 
@@ -239,6 +638,7 @@ def new_task_protocol(task_brief: TaskBrief | dict | None = None) -> dict[str, A
     """
     brief = TaskBrief.model_validate(task_brief).model_dump() if task_brief else None
     return {
+        "schema_version": TASK_PROTOCOL_SCHEMA_VERSION,
         "task_brief": brief,
         "task_brief_version": 1 if brief else 0,
         "task_brief_effective_at_sequence": 0,
@@ -255,6 +655,9 @@ def new_task_protocol(task_brief: TaskBrief | dict | None = None) -> dict[str, A
         "root_validation_history": [],
         "root_validation_passed": False,
         "protocol_correction_attempts": 0,
+        "task_progress_revisions": [],
+        "task_progress_head_revision": None,
+        "task_report_progress_revision": None,
     }
 
 
@@ -293,6 +696,17 @@ def replace_task_brief(
         0,
         int(effective_at_sequence),
     )
+    append_task_progress_revision(
+        state,
+        TaskProgress(phase="planning"),
+        expected_revision=int(state.get("task_progress_head_revision") or 0),
+        effective_at_sequence=effective_at_sequence,
+        contract_ref=TaskContractRef(
+            kind="task_brief",
+            task_brief_version=next_version,
+            task_brief_hash=task_brief_hash(brief),
+        ),
+    )
     return next_version
 
 
@@ -308,6 +722,8 @@ def ensure_task_protocol(context: dict[str, Any]) -> dict[str, Any]:
     defaults = new_task_protocol()
     for key, value in defaults.items():
         state.setdefault(key, deepcopy(value))
+    if state.get("schema_version") != TASK_PROTOCOL_SCHEMA_VERSION:
+        raise ValueError("Unsupported task_protocol schema_version")
     return state
 
 

cyber_agent/core/task_protocol_service.py

@@ -0,0 +1,114 @@
+"""TaskProtocol 命令的单进程原子执行边界。"""
+
+from __future__ import annotations
+
+import asyncio
+from copy import deepcopy
+from weakref import WeakValueDictionary
+
+from cyber_agent.core.agent_mode import require_mutable_trace_policy
+from cyber_agent.core.context_policy import get_authorized_context_snapshot
+from cyber_agent.core.task_protocol import (
+    TaskProgress,
+    TaskProgressRevision,
+    append_task_progress_revision,
+    ensure_task_protocol,
+    task_contract_ref,
+)
+from cyber_agent.trace.protocols import TraceStore
+
+
+_TRACE_LOCKS: WeakValueDictionary[
+    tuple[asyncio.AbstractEventLoop, str], asyncio.Lock
+] = WeakValueDictionary()
+
+
+class TaskProtocolStateError(RuntimeError):
+    """TaskProtocol 命令不能在当前持久化状态执行。"""
+
+
+class TaskProtocolService:
+    """将协议领域纯函数与 TraceStore 持久化组合起来。"""
+
+    def __init__(self, trace_store: TraceStore) -> None:
+        self.trace_store = trace_store
+
+    def _lock(self, trace_id: str) -> asyncio.Lock:
+        key = (asyncio.get_running_loop(), trace_id)
+        lock = _TRACE_LOCKS.get(key)
+        if lock is None:
+            lock = asyncio.Lock()
+            _TRACE_LOCKS[key] = lock
+        return lock
+
+    async def update_progress(
+        self,
+        trace_id: str,
+        *,
+        expected_revision: int,
+        progress: TaskProgress | dict,
+        effective_at_sequence: int,
+    ) -> TaskProgressRevision:
+        async with self._lock(trace_id):
+            trace = await self.trace_store.get_trace(trace_id)
+            if trace is None:
+                raise TaskProtocolStateError(f"Trace not found: {trace_id}")
+            policy = require_mutable_trace_policy(trace.context)
+            if not policy.requires_task_progress:
+                raise TaskProtocolStateError(
+                    "update_task_progress requires Recursive revision 3"
+                )
+            if (
+                trace.agent_type == "validator"
+                or trace.context.get("created_by_tool") == "validator"
+            ):
+                raise TaskProtocolStateError(
+                    "Validator traces cannot update TaskProgress"
+                )
+
+            parsed = TaskProgress.model_validate(progress)
+            root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
+            for item in (
+                *parsed.questions,
+                *parsed.blockers,
+                *parsed.findings,
+                *parsed.hypotheses,
+                *parsed.work_items,
+            ):
+                for ref in item.context_refs:
+                    get_authorized_context_snapshot(
+                        trace.context,
+                        ref_id=ref.ref_id,
+                        version=ref.version,
+                        root_trace_id=root_trace_id,
+                        uid=trace.uid,
+                    )
+
+            next_context = deepcopy(trace.context)
+            state = ensure_task_protocol(next_context)
+            if state["pending_reviews"]:
+                raise TaskProtocolStateError(
+                    "Review all child TaskReports before updating TaskProgress"
+                )
+            if state["next_actions"]:
+                raise TaskProtocolStateError(
+                    "Execute every approved next action before updating TaskProgress"
+                )
+            if state["task_report"] is not None:
+                raise TaskProtocolStateError(
+                    "TaskProgress cannot change after TaskReport submission"
+                )
+            revision = append_task_progress_revision(
+                state,
+                parsed,
+                expected_revision=expected_revision,
+                effective_at_sequence=effective_at_sequence,
+                contract_ref=task_contract_ref(
+                    state,
+                    root_task_anchor_hash=next_context.get(
+                        "root_task_anchor_hash"
+                    ),
+                ),
+            )
+            await self.trace_store.update_trace(trace_id, context=next_context)
+            return revision

cyber_agent/tools/builtin/__init__.py

@@ -16,7 +16,11 @@ from cyber_agent.tools.builtin.file.grep import grep_content
 from cyber_agent.tools.builtin.bash import bash_command
 from cyber_agent.tools.builtin.skill import skill, list_skills
 from cyber_agent.tools.builtin.subagent import agent, evaluate
-from cyber_agent.tools.builtin.task_protocol import submit_task_report, review_task_result
+from cyber_agent.tools.builtin.task_protocol import (
+    review_task_result,
+    submit_task_report,
+    update_task_progress,
+)
 # sandbox 工具已废弃（2026-04）；search.py / crawler.py 已重构为 content/ 工具族（2026-04）
 from cyber_agent.tools.builtin.knowledge import(knowledge_search,knowledge_save,knowledge_save_pending,knowledge_list,knowledge_update,knowledge_batch_update,knowledge_slim)
 # Memory / Dream（见 cyber_agent/docs/memory.md）
@@ -63,6 +67,7 @@ __all__ = [
     "evaluate",
     "submit_task_report",
     "review_task_result",
+    "update_task_progress",
     # 内容工具族（重构自 search.py + crawler.py）
     "content_platforms",
     "content_search",

cyber_agent/tools/builtin/task_protocol.py

@@ -12,7 +12,7 @@ from typing import Any
 
 from pydantic import ValidationError
 
-from cyber_agent.core.agent_mode import policy_from_context
+from cyber_agent.core.agent_mode import require_mutable_trace_policy
 from cyber_agent.core.context_policy import (
     ContextPolicyError,
     add_context_snapshot,
@@ -29,8 +29,11 @@ from cyber_agent.core.task_protocol import (
     TaskReview,
     ReviewDecision,
     TaskBrief,
+    TaskProgress,
+    task_progress_readiness_error,
     ensure_task_protocol,
 )
+from cyber_agent.core.task_protocol_service import TaskProtocolStateError
 from cyber_agent.core.validation import ValidationResult
 from cyber_agent.tools import tool
 
@@ -54,6 +57,43 @@ def _error(message: str) -> dict[str, Any]:
     return {"status": "failed", "error": message}
 
 
+@tool(
+    description=(
+        "Replace the current Recursive task's structured progress with one "
+        "complete, version-checked snapshot."
+    ),
+    hidden_params=["context"],
+    groups=["core"],
+)
+async def update_task_progress(
+    expected_revision: int,
+    progress: TaskProgress,
+    context: dict | None = None,
+) -> dict[str, Any]:
+    """追加当前 Trace 的下一版 TaskProgress。"""
+    if not context:
+        return _error("context is required")
+    if context.get("side_branch"):
+        return _error("Side branches cannot update TaskProgress")
+    service = context.get("task_protocol_service")
+    trace_id = context.get("trace_id")
+    if service is None or not trace_id:
+        return _error("task_protocol_service and trace_id are required")
+    try:
+        revision = await service.update_progress(
+            trace_id,
+            expected_revision=expected_revision,
+            progress=progress,
+            effective_at_sequence=int(context.get("sequence", 0) or 0),
+        )
+    except (TaskProtocolStateError, ValueError) as exc:
+        return _error(str(exc))
+    return {
+        "status": "completed",
+        "task_progress": revision.model_dump(mode="json"),
+    }
+
+
 @tool(
     description="Submit the current Recursive child Agent's structured task report before finishing.",
     hidden_params=["context"],
@@ -77,7 +117,10 @@ async def submit_task_report(
     trace = await store.get_trace(trace_id)
     if not trace:
         return _error(f"Trace not found: {trace_id}")
-    policy = policy_from_context(trace.context)
+    try:
+        policy = require_mutable_trace_policy(trace.context)
+    except ValueError as exc:
+        return _error(str(exc))
     if not policy.requires_task_protocol:
         return _error("submit_task_report is only available in Recursive revision 2")
     if not trace.parent_trace_id:
@@ -103,20 +146,31 @@ async def submit_task_report(
         return _error("A TaskReport has already been submitted")
 
     try:
-        submission = (
+        parsed_submission = (
             TaskReportSubmission.model_validate_json(task_report)
             if isinstance(task_report, str)
             else TaskReportSubmission.model_validate(task_report)
         )
+    except ValidationError as exc:
+        return _error(f"Invalid TaskReport: {exc}")
+    if policy.requires_task_progress and parsed_submission.outcome == "satisfied":
+        readiness_error = task_progress_readiness_error(state)
+        if readiness_error:
+            return _error(readiness_error)
+
+    try:
         report = TaskReport(
             child_trace_id=trace_id,
-            **submission.model_dump(),
+            **parsed_submission.model_dump(),
         )
     except ValidationError as exc:
         return _error(f"Invalid TaskReport: {exc}")
 
     state["task_report"] = report.model_dump()
     state["task_report_submitted_at_sequence"] = context.get("sequence", 0)
+    state["task_report_progress_revision"] = state.get(
+        "task_progress_head_revision"
+    )
     state["task_report_validation"] = None
     state["protocol_correction_attempts"] = 0
     await store.update_trace(trace_id, context=trace.context)
@@ -169,7 +223,10 @@ async def review_task_result(
     child = await store.get_trace(child_trace_id)
     if not parent or not child:
         return _error("Parent or child Trace not found")
-    policy = policy_from_context(parent.context)
+    try:
+        policy = require_mutable_trace_policy(parent.context)
+    except ValueError as exc:
+        return _error(str(exc))
     if not policy.requires_task_protocol:
         return _error("review_task_result is only available in Recursive revision 2")
     if child.parent_trace_id != parent_trace_id or child.uid != parent.uid:
@@ -388,6 +445,7 @@ async def review_task_result(
             child_state["report_history"].append(old_report)
         child_state["task_report"] = None
         child_state["task_report_submitted_at_sequence"] = None
+        child_state["task_report_progress_revision"] = None
         child_state["task_report_validation"] = None
         child_state["protocol_correction_attempts"] = 0
         replace_context_access(