feat(task-progress): 接入上下文、报告、验证、恢复与回溯

新建 Recursive revision 3 根与子 Trace 时初始化 TaskProgress，并仅把当前 head 注入模型上下文；新增只读 Progress 投影接口，避免业务直接读取 task_protocol 内部结构。

子报告与根验收绑定精确 Progress revision 和合同 hash，Validator packet 合并授权 ContextRef 与 ArtifactRef，plan hash 纳入结构化 Progress 且 Validator 不反写推进状态。

根完成与 satisfied 报告增加 ready_to_submit 门禁；恢复按消息落盘序列回退未完成工具写入，回溯仅沿当前父版本链移动 head，报告引用脱离有效祖先链时失败关闭。

cyber_agent/core/context_policy.py

@@ -13,7 +13,12 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
 
-from cyber_agent.core.task_protocol import ContextRef, RootTaskAnchor, TaskBrief
+from cyber_agent.core.task_protocol import (
+    ContextRef,
+    RootTaskAnchor,
+    TaskBrief,
+    current_task_progress,
+)
 
 
 MAX_ROOT_TASK_ANCHOR_CHARS = 4_000
@@ -514,6 +519,11 @@ def render_recursive_context(context: dict[str, Any]) -> str:
     payload = {
         "root_task_anchor": anchor.model_dump(mode="json"),
         "task_brief_version": state.get("task_brief_version", 0),
+        "task_progress": (
+            current_task_progress(state).model_dump(mode="json")
+            if state.get("task_progress_head_revision") is not None
+            else None
+        ),
         "available_context_refs": context_ref_descriptors(context),
         "context_usage": access.get("metrics", {}),
     }

cyber_agent/core/runner.py

@@ -76,9 +76,15 @@ from cyber_agent.core.agent_mode import (
 from cyber_agent.core.task_protocol import (
     RootTaskAnchor,
     ensure_task_protocol,
+    initialize_task_progress,
+    task_progress_artifact_refs,
+    task_progress_at_revision,
+    task_progress_readiness_error,
+    rewind_task_progress,
     protocol_error_report,
     rebuild_pending_replans,
 )
+from cyber_agent.core.task_protocol_service import TaskProtocolService
 from cyber_agent.core.context_policy import (
     canonical_json,
     ContextPolicyError,
@@ -86,6 +92,7 @@ from cyber_agent.core.context_policy import (
     persist_root_task_anchor,
     prune_context_access,
     render_recursive_context,
+    get_authorized_context_snapshot,
     require_matching_root_task_anchor,
     require_root_task_anchor,
     replace_context_access,
@@ -101,6 +108,7 @@ from cyber_agent.core.artifacts import (
     ArtifactRef,
     ArtifactResolver,
     MaterialIssue,
+    ValidationMaterial,
     extract_artifact_refs,
     material_chars,
     resolve_artifact_refs,
@@ -362,6 +370,9 @@ class AgentRunner:
         self.resource_budget = (
             ResourceBudgetController(trace_store) if trace_store else None
         )
+        self.task_protocol_service = (
+            TaskProtocolService(trace_store) if trace_store else None
+        )
 
     # ===== 核心公开方法 =====
 
@@ -517,22 +528,59 @@ class AgentRunner:
         )
         policy, settings = require_validation_policy(root.context)
         state = ensure_task_protocol(evaluated.context)
+        runtime_policy = policy_from_context(evaluated.context)
         authoritative_brief = state.get("task_brief")
         if authoritative_brief is not None:
             task_brief = authoritative_brief
         task_brief_version = int(state.get("task_brief_version", 0) or 0)
+        progress_revision = (
+            state.get("task_progress_head_revision")
+            if root_validator or task_report is None
+            else state.get("task_report_progress_revision")
+        )
+        task_progress = task_progress_at_revision(state, progress_revision)
+        if runtime_policy.requires_task_progress and task_progress is None:
+            raise ValueError("Recursive revision 3 validation requires TaskProgress")
         trajectory = await self.trace_store.get_main_path_messages(
             evaluated_trace_id,
             evaluated.head_sequence or evaluated.last_sequence,
         )
 
         refs: list[ArtifactRef] = []
+        materials: list[ValidationMaterial] = []
         source_urls: list[str] = []
         material_issues: list[MaterialIssue] = []
         try:
             if isinstance(task_report, dict):
                 refs.extend(extract_artifact_refs(task_report))
                 source_urls = list(task_report.get("source_urls") or [])
+            refs.extend(task_progress_artifact_refs(task_progress))
+            if task_progress is not None:
+                for item in (
+                    *task_progress.questions,
+                    *task_progress.blockers,
+                    *task_progress.findings,
+                    *task_progress.hypotheses,
+                    *task_progress.work_items,
+                ):
+                    for ref in item.context_refs:
+                        snapshot = get_authorized_context_snapshot(
+                            evaluated.context,
+                            ref_id=ref.ref_id,
+                            version=ref.version,
+                            root_trace_id=root_trace_id,
+                            uid=evaluated.uid,
+                        )
+                        materials.append(ValidationMaterial(
+                            artifact_id=ref.ref_id,
+                            version=snapshot.version,
+                            content_hash=snapshot.version,
+                            kind=f"context.{snapshot.kind}",
+                            mime_type="application/json",
+                            root_trace_id=root_trace_id,
+                            uid=evaluated.uid,
+                            content=snapshot.content,
+                        ))
             if root_validator:
                 for message in trajectory:
                     if message.role == "tool" and isinstance(message.content, dict):
@@ -546,12 +594,13 @@ class AgentRunner:
         unique_refs = {
             (item.artifact_id, item.version, item.content_hash): item for item in refs
         }
-        materials, resolved_issues = await resolve_artifact_refs(
+        resolved_materials, resolved_issues = await resolve_artifact_refs(
             list(unique_refs.values()),
             resolver=self.artifact_resolver,
             root_trace_id=root_trace_id,
             uid=evaluated.uid,
         )
+        materials.extend(resolved_materials)
         material_issues.extend(resolved_issues)
         total_material_chars = sum(material_chars(item) for item in materials)
         if deterministic_failure:
@@ -586,6 +635,7 @@ class AgentRunner:
             material_issues=material_issues,
             model_by_scope=model_by_scope,
             root=root_validator,
+            task_progress=task_progress,
         )
 
         cached = state.get("task_report_validation")
@@ -761,6 +811,7 @@ class AgentRunner:
                 root_task_anchor=root_anchor,
                 task_brief=task_brief,
                 task_report=task_report,
+                task_progress=task_progress,
                 candidate_output=candidate_output,
                 materials=materials,
                 material_issues=material_issues,
@@ -1271,6 +1322,13 @@ class AgentRunner:
             )
             trace_context[RESOURCE_BUDGET_CONTEXT_KEY] = budget.to_dict()
             state = ensure_task_protocol(trace_context)
+            if policy.requires_task_progress:
+                initialize_task_progress(
+                    state,
+                    root_task_anchor_hash=trace_context.get(
+                        "root_task_anchor_hash"
+                    ),
+                )
             replace_context_access(
                 trace_context,
                 [],
@@ -1445,6 +1503,31 @@ class AgentRunner:
                                 "Root task already used its two independent "
                                 "validation attempts; create a new trace"
                             )
+                    if policy.requires_task_progress:
+                        state = ensure_task_protocol(trace_obj.context)
+                        had_report = state.get("task_report") is not None
+                        report_progress_revision = state.get(
+                            "task_report_progress_revision"
+                        )
+                        if had_report and report_progress_revision is None:
+                            raise ValueError(
+                                "TaskReport is missing its TaskProgress revision binding"
+                            )
+                        previous_head = state.get("task_progress_head_revision")
+                        rewind_task_progress(state, trace_obj.last_sequence)
+                        if (
+                            had_report
+                            and state.get("task_report_progress_revision") is None
+                        ):
+                            raise ValueError(
+                                "TaskReport references TaskProgress outside the "
+                                "current revision ancestry"
+                            )
+                        if state.get("task_progress_head_revision") != previous_head:
+                            await self.trace_store.update_trace(
+                                trace_obj.trace_id,
+                                context=trace_obj.context,
+                            )
                     assert self.resource_budget is not None
                     await self.resource_budget.get_usage(root.trace_id)
 
@@ -2551,7 +2634,12 @@ class AgentRunner:
                 tool_calls = None
                 finish_reason = "budget_exhausted"
 
-            lifecycle_tools = {"agent", "submit_task_report", "review_task_result"}
+            lifecycle_tools = {
+                "agent",
+                "submit_task_report",
+                "review_task_result",
+                "update_task_progress",
+            }
             lifecycle_call_count = sum(
                 1 for tc in (tool_calls or [])
                 if tc.get("function", {}).get("name") in lifecycle_tools
@@ -3446,7 +3534,18 @@ class AgentRunner:
                     )
                     pending_reviews = bool(state["pending_reviews"])
                     pending_actions = bool(state["next_actions"])
-                    if missing_report or pending_reviews or pending_actions:
+                    progress_error = (
+                        task_progress_readiness_error(state)
+                        if policy.requires_task_progress
+                        and not trace.parent_trace_id
+                        else None
+                    )
+                    if (
+                        missing_report
+                        or pending_reviews
+                        or pending_actions
+                        or progress_error
+                    ):
                         attempts = state["protocol_correction_attempts"]
                         if attempts < 2:
                             state["protocol_correction_attempts"] = attempts + 1
@@ -3463,10 +3562,15 @@ class AgentRunner:
                                 required_action = (
                                     "execute the approved next action with agent"
                                 )
-                            else:
+                            elif missing_report:
                                 required_action = (
                                     "submit a valid TaskReport with submit_task_report"
                                 )
+                            else:
+                                required_action = (
+                                    "update TaskProgress to a ready_to_submit snapshot "
+                                    f"({progress_error})"
+                                )
                             history.append({
                                 "role": "user",
                                 "content": (
@@ -3483,6 +3587,9 @@ class AgentRunner:
                             )
                             state["task_report"] = report.model_dump()
                             state["task_report_submitted_at_sequence"] = sequence
+                            state["task_report_progress_revision"] = state.get(
+                                "task_progress_head_revision"
+                            )
                         completion_status = "failed"
                         await self.trace_store.update_trace(
                             trace_id,
@@ -3629,6 +3736,9 @@ class AgentRunner:
                     )
                     state["task_report"] = report.model_dump()
                     state["task_report_submitted_at_sequence"] = sequence
+                    state["task_report_progress_revision"] = state.get(
+                        "task_progress_head_revision"
+                    )
                     completion_status = "failed"
                     await self.trace_store.update_trace(
                         trace_id,
@@ -3775,6 +3885,7 @@ class AgentRunner:
                     state["report_history"].append(state["task_report"])
                 state["task_report"] = None
                 state["task_report_submitted_at_sequence"] = None
+                state["task_report_progress_revision"] = None
                 state["task_report_validation"] = None
             validation_cache = state.get("task_report_validation")
             if (
@@ -3821,6 +3932,8 @@ class AgentRunner:
                     "effective_at_sequence",
                     0,
                 )
+            if policy_from_context(trace.context).requires_task_progress:
+                rewind_task_progress(state, cutoff)
             state["pending_replans"] = rebuild_pending_replans(state)
             root_history = [
                 item
@@ -4816,6 +4929,7 @@ class AgentRunner:
         protocol_tools = {
             "submit_task_report",
             "review_task_result",
+            "update_task_progress",
             "read_context_ref",
         }
 
@@ -4833,6 +4947,8 @@ class AgentRunner:
             return tool_names & {"agent", "read_context_ref"}
 
         tool_names.discard("review_task_result")
+        if not policy.requires_task_progress or state.get("task_report") is not None:
+            tool_names.discard("update_task_progress")
         if not trace.parent_trace_id or state.get("task_report") is not None:
             tool_names.discard("submit_task_report")
         return tool_names
@@ -4905,6 +5021,7 @@ class AgentRunner:
         """
         framework_context = {
             "store": self.trace_store,
+            "task_protocol_service": self.task_protocol_service,
             "trace_id": trace_id,
             "goal_id": goal_id,
             "runner": self,

cyber_agent/core/validation.py

@@ -153,6 +153,11 @@ class ValidationPlan(_StrictModel):
     policy_version: str = Field(min_length=1)
     policy_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
     task_brief_version: int = Field(ge=0)
+    task_progress_revision: int = Field(default=0, ge=0)
+    task_progress_hash: str | None = Field(
+        default=None,
+        pattern=r"^[0-9a-f]{64}$",
+    )
     evaluated_head_sequence: int = Field(ge=0)
     effective_scopes: list[ValidationScope] = Field(min_length=1)
     checks: list[ValidationCheckSpec] = Field(min_length=1)
@@ -363,6 +368,7 @@ class ValidationPolicy(_StrictModel):
         material_issues: Sequence[MaterialIssue],
         model_by_scope: Mapping[str, str],
         root: bool,
+        task_progress: Mapping[str, Any] | BaseModel | None = None,
     ) -> ValidationPlan:
         """从任务合同和权威材料确定性编译完整检查计划。"""
         brief = _jsonable(task_brief) or {}
@@ -445,10 +451,19 @@ class ValidationPolicy(_StrictModel):
             }
             for issue in material_issues
         ]
+        progress = _jsonable(task_progress)
+        progress_revision = (
+            int(progress.get("revision", 0))
+            if isinstance(progress, dict)
+            else 0
+        )
+        progress_hash = _sha(progress) if progress is not None else None
         base = {
             "policy_version": self.policy_version,
             "policy_hash": self.policy_hash,
             "task_brief_version": task_brief_version,
+            "task_progress_revision": progress_revision,
+            "task_progress_hash": progress_hash,
             "evaluated_head_sequence": evaluated_head_sequence,
             "effective_scopes": scopes,
             "checks": [item.model_dump(mode="json") for item in checks],
@@ -456,6 +471,7 @@ class ValidationPolicy(_StrictModel):
             "task_brief": brief,
             "root_task_anchor": anchor,
             "task_report": _jsonable(task_report),
+            "task_progress": progress,
             "candidate_output": candidate_output,
             "model_by_scope": dict(model_by_scope),
             "tool_policy_version": self.tool_policy_version,
@@ -464,6 +480,8 @@ class ValidationPolicy(_StrictModel):
             policy_version=self.policy_version,
             policy_hash=self.policy_hash,
             task_brief_version=task_brief_version,
+            task_progress_revision=progress_revision,
+            task_progress_hash=progress_hash,
             evaluated_head_sequence=evaluated_head_sequence,
             effective_scopes=scopes,
             checks=checks,
@@ -596,6 +614,7 @@ def build_validation_packet(
     root_task_anchor: Mapping[str, Any] | BaseModel | None = None,
     task_brief: Mapping[str, Any] | BaseModel | None = None,
     task_report: Mapping[str, Any] | BaseModel | None = None,
+    task_progress: Mapping[str, Any] | BaseModel | None = None,
     completion_criteria: Sequence[str] | None = None,
     expected_outputs: Sequence[str] | None = None,
     candidate_output: str | None = None,
@@ -618,6 +637,7 @@ def build_validation_packet(
         "completion_criteria": _jsonable(completion_criteria or []),
         "expected_outputs": _jsonable(expected_outputs or []),
         "task_report": _jsonable(task_report),
+        "task_progress": _jsonable(task_progress),
         "candidate_output": candidate_output,
         "materials": [_jsonable(item) for item in materials],
         "trajectory": [],
@@ -941,6 +961,7 @@ class LLMValidator:
         materials: Sequence[ValidationMaterial],
         material_issues: Sequence[MaterialIssue],
         model_by_scope: Mapping[str, str],
+        task_progress: Mapping[str, Any] | BaseModel | None = None,
         source_urls: Sequence[str] = (),
         resume_scope_results: Sequence[ScopeValidationResult] = (),
         on_scope_result: ScopeResultCallback | None = None,
@@ -990,6 +1011,7 @@ class LLMValidator:
                         root_task_anchor=root_task_anchor,
                         task_brief=task_brief,
                         task_report=task_report,
+                        task_progress=task_progress,
                         candidate_output=candidate_output,
                         materials=materials,
                         model=model_by_scope.get(scope, ""),
@@ -1056,6 +1078,7 @@ class LLMValidator:
         model: str,
         source_urls: Sequence[str],
         validator_trace_id: str,
+        task_progress: Mapping[str, Any] | BaseModel | None = None,
     ) -> _ScopeRun:
         """创建一个Scope Trace并运行严格白名单的模型/工具循环。"""
         session = (
@@ -1081,6 +1104,7 @@ class LLMValidator:
                 root_task_anchor=root_task_anchor,
                 task_brief=task_brief,
                 task_report=task_report,
+                task_progress=task_progress,
                 candidate_output=candidate_output,
                 validation_plan=plan,
                 materials=materials,

cyber_agent/tools/builtin/subagent.py

@@ -24,12 +24,14 @@ from cyber_agent.core.agent_mode import (
     apply_policy_to_context,
     assert_removed_config_absent,
     policy_from_context,
+    require_mutable_trace_policy,
     validate_recursive_child_execution,
 )
 from cyber_agent.core.task_protocol import (
     TaskBrief,
     TaskReport,
     ensure_task_protocol,
+    initialize_task_progress,
     format_task_brief,
     new_task_protocol,
     pending_review_entry,
@@ -294,6 +296,9 @@ async def _load_or_create_task_report(
     )
     state["task_report"] = report.model_dump()
     state["task_report_submitted_at_sequence"] = generated_at_sequence
+    state["task_report_progress_revision"] = state.get(
+        "task_progress_head_revision"
+    )
     state["task_report_validation"] = None
     await store.update_trace(child_trace_id, context=child.context)
     return report
@@ -490,8 +495,11 @@ def _get_allowed_tools(
             blocked_tools.update({
                 "submit_task_report",
                 "review_task_result",
+                "update_task_progress",
                 "read_context_ref",
             })
+        elif not policy.requires_task_progress:
+            blocked_tools.add("update_task_progress")
         if policy.mode is AgentMode.LEGACY or agent_depth >= policy.max_depth:
             blocked_tools.add("agent")
         return sorted(allowed_tools - blocked_tools)
@@ -768,7 +776,7 @@ async def _run_agents(
     if not parent_trace:
         return {"status": "failed", "error": f"Parent trace not found: {trace_id}"}
     try:
-        policy = policy_from_context(parent_trace.context)
+        policy = require_mutable_trace_policy(parent_trace.context)
     except ValueError as exc:
         return {"status": "failed", "error": str(exc)}
     root_task_anchor = None
@@ -1063,6 +1071,8 @@ async def _run_agents(
                     )
                     child_context[CONTEXT_ACCESS_KEY] = prepared_context_accesses[i]
                     persist_root_task_anchor(child_context, root_task_anchor)
+                    if policy.requires_task_progress:
+                        initialize_task_progress(child_context["task_protocol"])
                 sub_trace = Trace(
                     trace_id=stid,
                     mode="agent",

cyber_agent/trace/api.py

@@ -16,6 +16,10 @@ from cyber_agent.core.agent_mode import (
     RECURSIVE_REVISION_READ_ONLY_ERROR,
     require_mutable_trace_policy,
 )
+from cyber_agent.core.task_protocol import (
+    current_task_progress,
+    ensure_task_protocol,
+)
 
 from .protocols import TraceStore
 
@@ -43,6 +47,15 @@ class MessagesResponse(BaseModel):
     messages: List[Dict[str, Any]]
 
 
+class TaskProgressResponse(BaseModel):
+    trace_id: str
+    protocol_revision: int
+    current: Optional[Dict[str, Any]] = None
+    task_report_progress_revision: Optional[int] = None
+    revision_count: int
+    revisions: Optional[List[Dict[str, Any]]] = None
+
+
 # ===== 全局 TraceStore（由 api_server.py 注入）=====
 
 
@@ -191,6 +204,29 @@ async def get_messages(
     )
 
 
+@router.get("/{trace_id}/task-progress", response_model=TaskProgressResponse)
+async def get_task_progress(
+    trace_id: str,
+    include_history: bool = Query(False),
+):
+    """返回 TaskProgress 的只读投影，不暴露其他协议内部状态。"""
+    store = get_trace_store()
+    trace = await store.get_trace(trace_id)
+    if not trace:
+        raise HTTPException(status_code=404, detail="Trace not found")
+    state = ensure_task_protocol(trace.context)
+    current = current_task_progress(state)
+    revisions = list(state.get("task_progress_revisions", []))
+    return TaskProgressResponse(
+        trace_id=trace_id,
+        protocol_revision=int(trace.context.get("agent_mode_revision", 1)),
+        current=(current.model_dump(mode="json") if current else None),
+        task_report_progress_revision=state.get("task_report_progress_revision"),
+        revision_count=len(revisions),
+        revisions=revisions if include_history else None,
+    )
+
+
 # ===== 知识反馈 =====