Agent/cyber_agent/core/task_protocol.py

"""Recursive revision 3 的父子任务协议与持久化状态。

Sub-Agent 使用 TaskBrief 下发任务，子 Agent 通过 TaskReport 上报，父 Agent 使用
TaskReview 审核。Runner 和两个协议工具共同把待审、后续动作和回退状态保存在 Trace context。
"""

from __future__ import annotations

from copy import deepcopy
from datetime import datetime
from hashlib import sha256
import json
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

from cyber_agent.core.artifacts import ArtifactRef
from cyber_agent.core.agent_mode import policy_from_context
from cyber_agent.core.task_contract import TaskContractRef
from cyber_agent.core.validation import RetryFrom
from cyber_agent.application.candidate import CandidateRef, CandidateReviewAction


TaskOutcome = Literal["satisfied", "partial", "failed", "protocol_error"]
NextStepDirection = Literal["DESCEND", "REFINE", "ASCEND", "NONE"]
ReviewDecision = Literal[
    "ACCEPT_REFINE",
    "REJECT_REFINE",
    "REVISE_CHILD",
    "DESCEND_AGAIN",
    "REPLAN_CURRENT",
    "ASCEND",
    "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):
    """协议模型的统一严格基类，拒绝未定义字段并清理字符串空白。"""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)


class RootTaskAnchor(StrictProtocolModel):
    """Recursive 根任务不可变的目标锚点。

    调用方创建根 Trace 时提供；Runner 固化后由每层孩子和 Validator读取，
    中间 Agent不能重新概括或覆盖它。
    """

    objective: str = Field(min_length=1)
    completion_criteria: list[str] = Field(min_length=1)
    constraints: list[str] = Field(default_factory=list)

    @field_validator("completion_criteria", "constraints")
    @classmethod
    def validate_non_empty_items(cls, items: list[str]) -> list[str]:
        if any(not item.strip() for item in items):
            raise ValueError("items must be non-empty strings")
        return items


class ContextRef(StrictProtocolModel):
    """模型可在 TaskBrief 中转交的最小上下文引用句柄。

    来源、类型、摘要和内容由框架保管；模型只能选择已获授权的 ID与版本。
    """

    ref_id: str = Field(min_length=1)
    version: str = Field(pattern=r"^[0-9a-f]{64}$")


class TaskBrief(StrictProtocolModel):
    """父 Agent 下发给直接孩子的结构化任务说明。

    `agent()` 在创建/续跑子 Trace 前规范化，Runner 再将格式化内容作为子 Agent 的任务上下文。
    """

    objective: str = Field(min_length=1)
    reason: str = Field(min_length=1)
    completion_criteria: list[str] = Field(min_length=1)
    expected_outputs: list[str] = Field(min_length=1)
    parent_findings: list[str] = Field(default_factory=list)
    context: dict[str, Any] = Field(default_factory=dict)
    constraints: list[str] = Field(default_factory=list)
    context_refs: list[ContextRef] = Field(default_factory=list)
    validation_scopes: list[TaskValidationScope] = Field(default_factory=list)

    @field_validator(
        "completion_criteria",
        "expected_outputs",
        "parent_findings",
        "constraints",
    )
    @classmethod
    def validate_non_empty_items(cls, items: list[str]) -> list[str]:
        if any(not item.strip() for item in items):
            raise ValueError("items must be non-empty strings")
        return items

    @field_validator("validation_scopes")
    @classmethod
    def normalize_validation_scopes(
        cls,
        scopes: list[TaskValidationScope],
    ) -> list[TaskValidationScope]:
        selected = set(scopes)
        return [scope for scope in _VALIDATION_SCOPE_ORDER if scope in selected]


class Validation(StrictProtocolModel):
    """子 Agent 在 TaskReport 中提交的执行者自检。

    `submit_task_report` 校验并持久化它；它不等于框架独立 Validator 产生的权威 ValidationResult。
    """

    hard_passed: bool
    open_issues: list[str] = Field(default_factory=list)

    @field_validator("open_issues")
    @classmethod
    def validate_open_issues(cls, items: list[str]) -> list[str]:
        if any(not item.strip() for item in items):
            raise ValueError("open_issues must contain non-empty strings")
        return items


class NextStepSuggestion(StrictProtocolModel):
    """子 Agent 对直接父级的下一步建议，本身不会创建同级任务。

    父 Agent 在 `review_task_result` 中结合独立验收决定是否采纳、修订或上行。
    """

    direction: NextStepDirection
    reason: str = Field(min_length=1)
    suggested_next_task: TaskBrief | None = None

    @model_validator(mode="after")
    def validate_suggested_task(self) -> "NextStepSuggestion":
        if self.direction in {"DESCEND", "REFINE"} and not self.suggested_next_task:
            raise ValueError("DESCEND/REFINE requires suggested_next_task")
        if self.direction in {"ASCEND", "NONE"} and self.suggested_next_task:
            raise ValueError("ASCEND/NONE does not accept suggested_next_task")
        return self


class TaskReportSubmission(StrictProtocolModel):
    """子 Agent 可提交的 TaskReport 负载，不允许模型填写 Trace ID。

    `submit_task_report` 接收该结构，检查结果与自检是否一致后由框架注入子 Trace ID。
    """

    summary: str = Field(min_length=1)
    outcome: TaskOutcome
    validation: Validation
    next_step_suggestion: NextStepSuggestion
    outputs: list[dict[str, Any]] = Field(default_factory=list)
    evidence: list[dict[str, Any]] = Field(default_factory=list)
    artifact_refs: list[ArtifactRef] = Field(default_factory=list, max_length=20)
    candidate_refs: list[CandidateRef] = Field(default_factory=list, max_length=20)
    source_urls: list[str] = Field(default_factory=list, max_length=20)
    remaining_issues: list[str] = Field(default_factory=list)

    @field_validator("remaining_issues")
    @classmethod
    def validate_remaining_issues(cls, items: list[str]) -> list[str]:
        if any(not item.strip() for item in items):
            raise ValueError("remaining_issues must contain non-empty strings")
        return items

    @field_validator("source_urls")
    @classmethod
    def validate_source_urls(cls, items: list[str]) -> list[str]:
        normalized: list[str] = []
        for item in items:
            value = item.strip()
            if not value:
                raise ValueError("source_urls must contain non-empty strings")
            if not value.lower().startswith(("http://", "https://")):
                raise ValueError("source_urls only accepts HTTP(S) URLs")
            if value not in normalized:
                normalized.append(value)
        return normalized

    @model_validator(mode="after")
    def validate_outcome_consistency(self) -> "TaskReportSubmission":
        if self.outcome == "satisfied" and (
            not self.validation.hard_passed
            or self.validation.open_issues
            or self.remaining_issues
        ):
            raise ValueError(
                "satisfied requires hard_passed with no open or remaining issues"
            )
        if self.outcome in {"failed", "protocol_error"} and self.validation.hard_passed:
            raise ValueError(f"{self.outcome} cannot have hard_passed=true")
        return self


class TaskReport(TaskReportSubmission):
    """框架补全 `child_trace_id` 后的权威子任务报告。

    Sub-Agent 汇合孩子结果后将其与 ValidationResult 一起写入父 Trace 的待审队列。
    """

    child_trace_id: str = Field(min_length=1)


class TaskReview(StrictProtocolModel):
    """父 Agent 对一份直接孩子报告的结构化审核结果。

    `review_task_result` 根据 ValidationResult 的决策矩阵构建它，然后持久化修订、新任务或回退动作。
    """

    parent_trace_id: str = Field(min_length=1)
    child_trace_id: str = Field(min_length=1)
    decision: ReviewDecision
    reason: str = Field(min_length=1)
    approved_next_task: TaskBrief | None = None
    retry_from: RetryFrom | None = None
    candidate_actions: list[CandidateReviewAction] = Field(
        default_factory=list,
        max_length=20,
    )

    @model_validator(mode="after")
    def validate_approved_next_task(self) -> "TaskReview":
        action_types = {item.action for item in self.candidate_actions}
        if "revise" in action_types and self.decision != "REVISE_CHILD":
            raise ValueError("candidate revise requires REVISE_CHILD")
        if self.decision == "REVISE_CHILD" and action_types - {"revise", "discard"}:
            raise ValueError("REVISE_CHILD cannot adopt a candidate")
        identities = [
            (item.candidate_ref.candidate_id, item.candidate_ref.revision)
            for item in self.candidate_actions
        ]
        if len(identities) != len(set(identities)):
            raise ValueError("candidate_actions must reference unique candidates")
        if self.decision in {"ACCEPT_REFINE", "DESCEND_AGAIN"}:
            if not self.approved_next_task:
                raise ValueError(f"{self.decision} requires approved_next_task")
        elif self.decision != "REVISE_CHILD" and self.approved_next_task:
            raise ValueError(
                f"{self.decision} does not accept approved_next_task"
            )
        if self.decision == "REPLAN_CURRENT":
            if self.retry_from is None:
                raise ValueError("REPLAN_CURRENT requires retry_from")
        elif self.retry_from is not None:
            raise ValueError(f"{self.decision} does not accept retry_from")
        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 任务协议初始状态。

    Runner 初始化根 Trace，Sub-Agent 创建子 Trace 时调用，状态随 Trace context 持久化。
    """
    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,
        "task_brief_history": [],
        "task_report": None,
        "task_report_submitted_at_sequence": None,
        "task_report_validation": None,
        "report_history": [],
        "pending_reviews": {},
        "reviews": [],
        "next_actions": [],
        "pending_replans": [],
        "root_validation_attempts": 0,
        "root_validation_history": [],
        "root_validation_passed": False,
        "protocol_correction_attempts": 0,
        "task_progress_revisions": [],
        "task_progress_head_revision": None,
        "task_report_progress_revision": None,
    }


def replace_task_brief(
    state: dict[str, Any],
    task_brief: TaskBrief | dict[str, Any],
    *,
    effective_at_sequence: int,
) -> int:
    """以唯一入口替换当前 TaskBrief并保留旧版本。

    `review_task_result`批准修订后，`agent(continue_from=...)`真正恢复孩子时调用，
    避免审核阶段提前改写孩子，也避免覆盖后丢失任务合同的审计版本。
    """
    brief = TaskBrief.model_validate(task_brief).model_dump(mode="json")
    current = state.get("task_brief")
    current_version = state.get("task_brief_version", 0)
    if not isinstance(current_version, int) or current_version < 0:
        current_version = 0
    if current is not None:
        history_entry = {
            "version": current_version or 1,
            "task_brief": deepcopy(current),
            "effective_at_sequence": state.get(
                "task_brief_effective_at_sequence",
                0,
            ),
        }
        history = state.setdefault("task_brief_history", [])
        if history_entry not in history:
            history.append(history_entry)
    next_version = max(current_version, 0) + 1 if current is not None else 1
    state["task_brief"] = brief
    state["task_brief_version"] = next_version
    state["task_brief_effective_at_sequence"] = max(
        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


def ensure_task_protocol(context: dict[str, Any]) -> dict[str, Any]:
    """获取 Trace context 中的协议状态，并幂等补齐缺失字段。

    Runner、Sub-Agent 和报告/审核工具在读写任务状态前统一调用此入口。
    """
    state = context.get("task_protocol")
    policy = policy_from_context(context)
    strict = policy.requires_task_progress
    if not isinstance(state, dict):
        if strict:
            raise ValueError(
                "Recursive revision 3 task_protocol is missing or invalid"
            )
        state = new_task_protocol()
        context["task_protocol"] = state
    defaults = new_task_protocol()
    if strict:
        missing = sorted(set(defaults).difference(state))
        if missing:
            raise ValueError(
                "Recursive revision 3 task_protocol is missing required fields: "
                + ", ".join(missing)
            )
    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


def rebuild_pending_replans(state: dict[str, Any]) -> list[dict[str, Any]]:
    """在 Trace 回溯截断审核历史后，重建尚未解决的重规划请求。

    `AgentRunner._rewind` 清理截断点之后的协议记录时调用，避免丢失同批次的 `REPLAN_CURRENT`。
    """
    pending_batches = {
        entry.get("received_at_sequence")
        for entry in state.get("pending_reviews", {}).values()
    }
    pending_batches.discard(None)
    if not pending_batches:
        return []
    replans = []
    for review in state.get("reviews", []):
        pending_entry = review.get("pending_review", {})
        if (
            review.get("decision") == "REPLAN_CURRENT"
            and pending_entry.get("received_at_sequence") in pending_batches
        ):
            replans.append({
                "child_trace_id": review.get("child_trace_id"),
                "goal_id": pending_entry.get("goal_id"),
                "retry_from": review.get("retry_from"),
                "requested_at_sequence": review.get("reviewed_at_sequence", 0),
                "received_at_sequence": pending_entry.get(
                    "received_at_sequence"
                ),
            })
    return replans


def protocol_error_report(child_trace_id: str, reason: str) -> TaskReport:
    """在子 Agent 未能提交合法报告时生成框架级失败报告。

    Runner 的完成门禁和 Sub-Agent 结果汇合阶段调用，使父级仍能审核并恢复。
    """
    return TaskReport(
        child_trace_id=child_trace_id,
        summary="Task protocol failed before a valid report was submitted.",
        outcome="protocol_error",
        validation=Validation(hard_passed=False, open_issues=[reason]),
        next_step_suggestion=NextStepSuggestion(
            direction="ASCEND",
            reason="Parent review is required to revise or fail this task.",
        ),
        remaining_issues=[reason],
    )


def stopped_task_report(child_trace_id: str, reason: str) -> TaskReport:
    """为被协作式停止的子 Agent 生成 `failed` TaskReport。

    Sub-Agent 的停止汇合路径调用，然后交由父 Agent 选择修订孩子或失败结束。
    """
    return TaskReport(
        child_trace_id=child_trace_id,
        summary="Child Agent execution was stopped before completion.",
        outcome="failed",
        validation=Validation(hard_passed=False, open_issues=[reason]),
        next_step_suggestion=NextStepSuggestion(
            direction="NONE",
            reason="Parent review must revise the stopped child or fail the task.",
        ),
        remaining_issues=[reason],
    )


def pending_review_entry(
    *,
    goal_id: str | None,
    report: TaskReport,
    validation_result: dict[str, Any],
    received_at_sequence: int,
) -> dict[str, Any]:
    """组装父 Trace `pending_reviews` 中的一份待审记录。

    Sub-Agent 完成子 Trace 验收后调用，将 TaskReport、ValidationResult、原 Goal 和序列绑定。
    """
    return {
        "goal_id": goal_id,
        "task_report": report.model_dump(),
        "validation_result": validation_result,
        "received_at_sequence": received_at_sequence,
        "received_at": datetime.now().isoformat(),
    }


def format_task_brief(
    task_brief: TaskBrief,
    *,
    available_context_refs: list[dict[str, Any]] | None = None,
) -> str:
    """将结构化 TaskBrief 格式化为子 Agent 可执行的任务文本。

    ``agent`` 完成规范化后调用，生成 Runner 作为 user message 传给子 Agent 的任务说明。
    """
    criteria = "\n".join(
        f"{index}. {item}"
        for index, item in enumerate(task_brief.completion_criteria, start=1)
    )
    constraints = "\n".join(f"- {item}" for item in task_brief.constraints)
    expected_outputs = "\n".join(
        f"- {item}" for item in task_brief.expected_outputs
    )
    parent_findings = "\n".join(
        f"- {item}" for item in task_brief.parent_findings
    )
    parts = [
        "# Task Brief",
        f"\n## Objective\n{task_brief.objective}",
        f"\n## Reason\n{task_brief.reason}",
        f"\n## Completion Criteria\n{criteria}",
        f"\n## Expected Outputs\n{expected_outputs}",
    ]
    if parent_findings:
        parts.append(f"\n## Parent Findings\n{parent_findings}")
    if task_brief.context:
        parts.append(
            "\n## Context\n"
            + json.dumps(
                task_brief.context,
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
            )
        )
    if constraints:
        parts.append(f"\n## Constraints\n{constraints}")
    effective_scopes = [*task_brief.validation_scopes]
    if "task" not in effective_scopes:
        effective_scopes.append("task")
    parts.append(
        "\n## Validation Scopes\n"
        + ", ".join(effective_scopes)
    )
    if available_context_refs:
        parts.append(
            "\n## Available Context References\n"
            + json.dumps(
                available_context_refs,
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
            )
        )
    parts.append(
        "\nSubmit a validated TaskReport with submit_task_report before finishing."
    )
    return "\n".join(parts)