Agent/cyber_agent/core/artifacts.py

"""Recursive Validator 的最小只读产物协议。

框架只保存不可变 ``ArtifactRef``，并通过应用注入的 ``ArtifactResolver``
读取当前任务获授权的真实材料；上传、ACL、索引和生命周期不属于本模块。
"""

from __future__ import annotations

from collections.abc import Mapping, Sequence
from hashlib import sha256
import json
from typing import Any, Literal, Protocol, runtime_checkable

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


class _StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)


class ArtifactRef(_StrictModel):
    """任务报告或标准 Tool Result 中的不可变产物句柄。"""

    artifact_id: str = Field(min_length=1, max_length=500)
    version: str = Field(min_length=1, max_length=200)
    content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    kind: str = Field(min_length=1, max_length=100)
    mime_type: str | None = Field(default=None, max_length=200)


class ValidationMaterial(_StrictModel):
    """Resolver 返回、可直接放入 Validator packet 的只读材料。"""

    artifact_id: str = Field(min_length=1, max_length=500)
    version: str = Field(min_length=1, max_length=200)
    content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    kind: str = Field(min_length=1, max_length=100)
    mime_type: str | None = Field(default=None, max_length=200)
    root_trace_id: str = Field(min_length=1)
    uid: str | None = None
    content: Any

    @field_validator("content")
    @classmethod
    def validate_content(cls, value: Any) -> Any:
        canonical_material_content(value)
        return value

    def to_ref(self) -> ArtifactRef:
        return ArtifactRef(
            artifact_id=self.artifact_id,
            version=self.version,
            content_hash=self.content_hash,
            kind=self.kind,
            mime_type=self.mime_type,
        )


@runtime_checkable
class ArtifactResolver(Protocol):
    """由业务应用实现的可信、只读产物解析端口。"""

    async def resolve(
        self,
        ref: ArtifactRef,
        root_trace_id: str,
        uid: str | None,
    ) -> ValidationMaterial:
        ...


class ArtifactResolutionError(RuntimeError):
    """带确定性语义的产物解析错误。"""

    def __init__(
        self,
        outcome: Literal["failed", "unknown", "error"],
        message: str,
    ) -> None:
        super().__init__(message)
        self.outcome = outcome


class ArtifactNotFound(ArtifactResolutionError):
    def __init__(self, message: str = "Artifact does not exist") -> None:
        super().__init__("failed", message)


class ArtifactTemporarilyUnavailable(ArtifactResolutionError):
    def __init__(self, message: str = "Artifact resolver is unavailable") -> None:
        super().__init__("unknown", message)


class ArtifactAccessDenied(ArtifactResolutionError):
    def __init__(self, message: str = "Artifact is not authorized") -> None:
        super().__init__("error", message)


class MaterialIssue(_StrictModel):
    """框架在 LLM 调用前产生的材料确定性检查结果。"""

    artifact_id: str
    outcome: Literal["failed", "unknown", "error"]
    reason: str = Field(min_length=1)
    scopes: list[Literal["evidence", "hypothesis", "output", "task", "root"]] = (
        Field(default_factory=lambda: [
            "evidence", "hypothesis", "output", "task", "root",
        ], min_length=1)
    )

    @field_validator("scopes")
    @classmethod
    def normalize_scopes(cls, value: list[str]) -> list[str]:
        order = ("evidence", "hypothesis", "output", "task", "root")
        selected = set(value)
        return [scope for scope in order if scope in selected]


def canonical_material_content(content: Any) -> str:
    """稳定序列化材料内容，供字符计数和 SHA-256 校验共用。"""
    if isinstance(content, str):
        return content
    try:
        return json.dumps(
            content,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        )
    except (TypeError, ValueError) as exc:
        raise ValueError(f"Artifact content must be JSON serializable: {exc}") from exc


def material_content_hash(content: Any) -> str:
    return sha256(canonical_material_content(content).encode("utf-8")).hexdigest()


def material_chars(material: ValidationMaterial) -> int:
    return len(canonical_material_content(material.content))


def extract_artifact_refs(value: Any) -> list[ArtifactRef]:
    """只读取标准对象顶层 ``artifact_refs``，非法元数据直接失败关闭。"""
    if isinstance(value, BaseModel):
        value = value.model_dump(mode="json")
    if not isinstance(value, Mapping) or "artifact_refs" not in value:
        return []
    raw_refs = value.get("artifact_refs")
    if not isinstance(raw_refs, Sequence) or isinstance(
        raw_refs, (str, bytes, bytearray)
    ):
        raise ValueError("artifact_refs must be an array")
    found = [ArtifactRef.model_validate(raw) for raw in raw_refs]
    unique: dict[tuple[str, str, str], ArtifactRef] = {}
    for ref in found:
        unique[(ref.artifact_id, ref.version, ref.content_hash)] = ref
    return list(unique.values())


async def resolve_artifact_refs(
    refs: Sequence[ArtifactRef],
    *,
    resolver: ArtifactResolver | None,
    root_trace_id: str,
    uid: str | None,
) -> tuple[list[ValidationMaterial], list[MaterialIssue]]:
    """逐个解析引用并把缺失、暂不可用和越权映射为稳定结果。"""
    materials: list[ValidationMaterial] = []
    issues: list[MaterialIssue] = []
    for ref in refs:
        issue_scopes = _artifact_validation_scopes(ref.kind)
        if resolver is None:
            issues.append(MaterialIssue(
                artifact_id=ref.artifact_id,
                outcome="unknown",
                reason="No ArtifactResolver is configured for this artifact",
                scopes=issue_scopes,
            ))
            continue
        try:
            material = ValidationMaterial.model_validate(
                await resolver.resolve(ref, root_trace_id, uid)
            )
            if material.root_trace_id != root_trace_id or material.uid != uid:
                raise ArtifactAccessDenied(
                    "Resolved artifact belongs to another task tree or owner"
                )
            if material.to_ref() != ref:
                raise ArtifactAccessDenied(
                    "Resolved artifact metadata does not match the authorized reference"
                )
            if material_content_hash(material.content) != ref.content_hash:
                raise ArtifactNotFound("Resolved artifact content hash does not match")
            materials.append(material)
        except ArtifactResolutionError as exc:
            issues.append(MaterialIssue(
                artifact_id=ref.artifact_id,
                outcome=exc.outcome,
                reason=str(exc),
                scopes=issue_scopes,
            ))
        except (ValidationError, TypeError, ValueError) as exc:
            issues.append(MaterialIssue(
                artifact_id=ref.artifact_id,
                outcome="error",
                reason=f"Artifact resolver returned invalid data: {exc}",
                scopes=issue_scopes,
            ))
        except Exception as exc:
            issues.append(MaterialIssue(
                artifact_id=ref.artifact_id,
                outcome="unknown",
                reason=f"Artifact resolver failed: {exc}",
                scopes=issue_scopes,
            ))
    return materials, issues


def _artifact_validation_scopes(
    kind: str,
) -> list[Literal["evidence", "hypothesis", "output", "task", "root"]]:
    """按产物类型限制确定性失败的影响面，未受影响 Scope 仍继续验收。"""
    normalized = kind.strip().lower()
    if any(marker in normalized for marker in ("evidence", "source", "reference")):
        return ["evidence", "hypothesis", "task", "root"]
    return ["output", "task", "root"]