框架：补齐冻结角色提示词、压缩策略与追踪附件能力

agent/agent/__init__.py

@@ -26,7 +26,8 @@ from agent.trace.models import (
     MessageContent,
 )
 from agent.trace.goal_models import Goal, GoalTree, GoalStatus
-from agent.trace.protocols import TraceStore
+from agent.trace.attachments import AttachmentRef
+from agent.trace.protocols import TraceAttachmentStore, TraceStore
 from agent.trace.store import FileSystemTraceStore
 
 # 技能系统
@@ -60,6 +61,8 @@ from agent.orchestration import (
     OrchestrationConfig,
     RoleRunConfigOverrides,
     RoleRunConfigResolver,
+    RoleSystemPromptOverride,
+    RoleSystemPromptResolver,
 )
 from agent.orchestration.wiring import wire_orchestration
 
@@ -89,6 +92,8 @@ __all__ = [
     "GoalTree",
     "GoalStatus",
     "TraceStore",
+    "TraceAttachmentStore",
+    "AttachmentRef",
     "FileSystemTraceStore",
     # Skill
     "Skill",
@@ -122,6 +127,8 @@ __all__ = [
     "OrchestrationConfig",
     "RoleRunConfigOverrides",
     "RoleRunConfigResolver",
+    "RoleSystemPromptOverride",
+    "RoleSystemPromptResolver",
     "wire_orchestration",
     # SDK
     "invoke_agent",

agent/agent/core/prompts/orchestration.py

@@ -3,15 +3,19 @@
 PLANNER_ROLE_CONTRACT = """
 ## Framework role contract: Planner
 
-You are the only planner. Create tasks with task_plan and dispatch them with
-the authorized task-dispatch tool exposed by your preset. Worker completion
-is only an attempt submission. A task is
-complete only after an independent passed validation and your task_decide
-action=accept. For failed, inconclusive, or validation errors, explicitly
-repair, retry, revise, split, revalidate, block, cancel, or supersede.
-Start or resume by calling task_plan operation=inspect. The returned Root Task
-is the mission: top-level tasks are its children, and the run is complete only
-after the Root itself has passed validation and you accept it.
+You are the only planner. Create and dispatch tasks with the authorized
+planning and task-dispatch tools exposed by your preset. Tool
+names are preset-specific: use only the plan, inspection, validation, decision,
+and dispatch capabilities actually available in the current preset. Worker
+completion is only an attempt submission. A task is complete only after an
+independent passed validation and an accept decision through the preset's
+authorized decision tool. For failed, inconclusive, or validation errors,
+explicitly repair, retry, revise, split, revalidate, block, cancel, or
+supersede. Start or resume by using the preset's authorized plan-inspection
+capability. The returned Root Task is the mission: top-level tasks are its
+children. Finish the Root only through the preset's authorized decision tool
+and the active host policy: ordinary missions require Root validation and
+acceptance, while an explicit host capability boundary may require Root BLOCK.
 """.strip()
 
 WORKER_ROLE_CONTRACT = """

agent/agent/orchestration/__init__.py

@@ -48,7 +48,12 @@ from .models import (
     ValidationVerdict,
 )
 from .protocols import AgentExecutor, ArtifactStore, EventSink, TaskStore
-from .run_config import RoleRunConfigOverrides, RoleRunConfigResolver
+from .run_config import (
+    RoleRunConfigOverrides,
+    RoleRunConfigResolver,
+    RoleSystemPromptOverride,
+    RoleSystemPromptResolver,
+)
 from .state_machine import InvalidTaskTransition
 from .store import (
     ArtifactConflict,
@@ -131,6 +136,8 @@ __all__ = [
     "RevisionConflict",
     "RoleRunConfigOverrides",
     "RoleRunConfigResolver",
+    "RoleSystemPromptOverride",
+    "RoleSystemPromptResolver",
     "RootCompletionView",
     "RuleBasedDeterministicValidator",
     "TaskAttempt",

agent/agent/orchestration/executor.py

@@ -15,7 +15,12 @@ from agent.tools.builtin.knowledge import KnowledgeConfig
 
 from .models import AgentRole, CompletionPolicy, ExecutionStats, FailureCode
 from .protocols import ValidatorRunResult, WorkerRunResult
-from .run_config import RoleRunConfigOverrides, RoleRunConfigResolver
+from .run_config import (
+    RoleRunConfigOverrides,
+    RoleRunConfigResolver,
+    RoleSystemPromptOverride,
+    RoleSystemPromptResolver,
+)
 
 
 class LocalAgentExecutor:
@@ -25,9 +30,11 @@ class LocalAgentExecutor:
         self,
         runner: Any,
         role_run_config_resolver: Optional[RoleRunConfigResolver] = None,
+        role_system_prompt_resolver: Optional[RoleSystemPromptResolver] = None,
     ) -> None:
         self.runner = runner
         self.role_run_config_resolver = role_run_config_resolver
+        self.role_system_prompt_resolver = role_system_prompt_resolver
 
     async def stop(self, trace_id: str) -> bool:
         """Request cooperative cancellation of a local sub-trace."""
@@ -56,11 +63,6 @@ class LocalAgentExecutor:
             )
             trace = await self.runner.trace_store.get_trace(continue_trace)
             usage_baseline = _trace_usage(trace)
-            trace_context = dict(trace.context or {})
-            trace_context.update(protected)
-            await self.runner.trace_store.update_trace(
-                continue_trace, context=trace_context
-            )
 
         prompt = {
             "task_spec": context["task_spec"],
@@ -150,6 +152,28 @@ class LocalAgentExecutor:
             name=name,
         )
         try:
+            prompt_override = await self._resolve_system_prompt(
+                role=role,
+                preset=preset,
+                context=context,
+                continue_trace_id=continue_trace_id,
+            )
+            if prompt_override is not None:
+                protected["role_prompt_identity"] = prompt_override.prompt_identity
+            if continue_trace_id:
+                trace = await self.runner.trace_store.get_trace(continue_trace_id)
+                trace_context = dict(trace.context or {})
+                trace_context.update(protected)
+                await self.runner.trace_store.update_trace(
+                    continue_trace_id, context=trace_context
+                )
+            elif prompt_override is not None:
+                # Existing traces keep their originally persisted system message.
+                config = replace(
+                    config,
+                    context=protected,
+                    system_prompt=prompt_override.content,
+                )
             config = await self._apply_run_config_overrides(
                 config,
                 role=role,
@@ -226,6 +250,47 @@ class LocalAgentExecutor:
             _role_run_config_override_fields=frozenset(values),
         )
 
+    async def _resolve_system_prompt(
+        self,
+        *,
+        role: AgentRole,
+        preset: str,
+        context: Dict[str, Any],
+        continue_trace_id: Optional[str],
+    ) -> Optional[RoleSystemPromptOverride]:
+        resolver = self.role_system_prompt_resolver
+        override: Optional[RoleSystemPromptOverride] = None
+        if resolver is not None:
+            override = await resolver.resolve(
+                role=role,
+                preset=preset,
+                context=_deep_readonly(deepcopy(context)),
+            )
+            if override is not None and not isinstance(
+                override, RoleSystemPromptOverride
+            ):
+                raise TypeError(
+                    "RoleSystemPromptResolver.resolve() must return "
+                    "RoleSystemPromptOverride or None"
+                )
+
+        if continue_trace_id:
+            trace = await self.runner.trace_store.get_trace(continue_trace_id)
+            if trace is None:
+                raise ValueError(f"Trace not found: {continue_trace_id}")
+            stored_identity = (trace.context or {}).get("role_prompt_identity")
+            resolved_identity = (
+                override.prompt_identity if override is not None else None
+            )
+            if stored_identity != resolved_identity and (
+                stored_identity is not None or resolver is not None
+            ):
+                raise ValueError(
+                    "ROLE_SYSTEM_PROMPT_MISMATCH: repair continuation prompt "
+                    "identity does not match the original trace"
+                )
+        return override
+
     async def _get_trace(self, trace_id: str) -> Any:
         trace_store = getattr(self.runner, "trace_store", None)
         if trace_store is None:

agent/agent/orchestration/run_config.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
+from hashlib import sha256
 from math import isfinite
 from typing import Any, Mapping, Optional, Protocol
 
@@ -53,4 +55,45 @@ class RoleRunConfigResolver(Protocol):
         ...
 
 
-__all__ = ["RoleRunConfigOverrides", "RoleRunConfigResolver"]
+_SHA256_IDENTITY = re.compile(r"^sha256:[0-9a-f]{64}$")
+
+
+@dataclass(frozen=True, slots=True)
+class RoleSystemPromptOverride:
+    """A frozen business prompt and its content-derived identity."""
+
+    content: str
+    prompt_identity: str
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.content, str) or not self.content.strip():
+            raise ValueError("content must be a non-empty string")
+        if not isinstance(self.prompt_identity, str) or not _SHA256_IDENTITY.fullmatch(
+            self.prompt_identity
+        ):
+            raise ValueError("prompt_identity must be sha256:<64 lowercase hex>")
+        actual = f"sha256:{sha256(self.content.encode('utf-8')).hexdigest()}"
+        if actual != self.prompt_identity:
+            raise ValueError("prompt_identity does not match content")
+
+
+class RoleSystemPromptResolver(Protocol):
+    """Resolve the immutable system prompt for one isolated role run."""
+
+    async def resolve(
+        self,
+        *,
+        role: AgentRole,
+        preset: str,
+        context: Mapping[str, Any],
+    ) -> Optional[RoleSystemPromptOverride]:
+        """Return a frozen prompt, or ``None`` to keep preset behavior."""
+        ...
+
+
+__all__ = [
+    "RoleRunConfigOverrides",
+    "RoleRunConfigResolver",
+    "RoleSystemPromptOverride",
+    "RoleSystemPromptResolver",
+]

agent/agent/orchestration/wiring.py

@@ -9,7 +9,7 @@ from .coordinator import TaskCoordinator
 from .evidence import EvidenceProvider
 from .executor import LocalAgentExecutor
 from .protocols import ArtifactStore, EventSink, TaskStore
-from .run_config import RoleRunConfigResolver
+from .run_config import RoleRunConfigResolver, RoleSystemPromptResolver
 from .validation_policy import DeterministicValidator, ValidationPolicy
 
 
@@ -24,12 +24,13 @@ def wire_orchestration(
     deterministic_validator: Optional[DeterministicValidator] = None,
     evidence_provider: Optional[EvidenceProvider] = None,
     role_run_config_resolver: Optional[RoleRunConfigResolver] = None,
+    role_system_prompt_resolver: Optional[RoleSystemPromptResolver] = None,
 ) -> TaskCoordinator:
     """Wire generic orchestration ports without importing a business project.
 
-    Validation, evidence, and role run configuration dependencies are optional
-    keyword-only extension points. Existing positional calls keep the default
-    Coordinator behavior.
+    Validation, evidence, role run configuration, and role prompt dependencies
+    are optional keyword-only extension points. Existing positional calls keep
+    the default Coordinator behavior.
     """
     coordinator = TaskCoordinator(
         task_store=task_store,
@@ -44,6 +45,7 @@ def wire_orchestration(
     executor = LocalAgentExecutor(
         runner,
         role_run_config_resolver=role_run_config_resolver,
+        role_system_prompt_resolver=role_system_prompt_resolver,
     )
     coordinator.set_executor(executor)
     runner.task_coordinator = coordinator

agent/agent/trace/__init__.py

@@ -11,7 +11,8 @@ Trace module - 执行追踪与计划管理
 
 from .models import Trace, Message
 from .goal_models import Goal, GoalTree, GoalStatus, GoalType, GoalStats
-from .protocols import TraceStore
+from .attachments import AttachmentRef
+from .protocols import TraceAttachmentStore, TraceStore
 from .store import FileSystemTraceStore
 from .trace_id import generate_trace_id, generate_sub_trace_id, parse_parent_trace_id
 
@@ -26,6 +27,8 @@ __all__ = [
     "GoalStats",
     # Store
     "TraceStore",
+    "TraceAttachmentStore",
+    "AttachmentRef",
     "FileSystemTraceStore",
     # Utils
     "generate_trace_id",

agent/agent/trace/attachments.py

@@ -0,0 +1,55 @@
+"""Durable attachment references for trace messages."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+_SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$")
+
+
+@dataclass(frozen=True, slots=True)
+class AttachmentRef:
+    """Content-addressed reference to one persisted message attachment."""
+
+    trace_id: str
+    message_id: str
+    media_type: str
+    sha256: str
+    size_bytes: int
+
+    def __post_init__(self) -> None:
+        for field_name in ("trace_id", "message_id", "media_type"):
+            value = getattr(self, field_name)
+            if not isinstance(value, str) or not value.strip():
+                raise ValueError(f"{field_name} must be a non-empty string")
+        if not isinstance(self.sha256, str) or not _SHA256.fullmatch(self.sha256):
+            raise ValueError("sha256 must be sha256:<64 lowercase hex>")
+        if (
+            isinstance(self.size_bytes, bool)
+            or not isinstance(self.size_bytes, int)
+            or self.size_bytes < 0
+        ):
+            raise ValueError("size_bytes must be a non-negative integer")
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "trace_id": self.trace_id,
+            "message_id": self.message_id,
+            "media_type": self.media_type,
+            "sha256": self.sha256,
+            "size_bytes": self.size_bytes,
+        }
+
+    @classmethod
+    def from_dict(cls, value: dict[str, object]) -> "AttachmentRef":
+        return cls(
+            trace_id=str(value["trace_id"]),
+            message_id=str(value["message_id"]),
+            media_type=str(value["media_type"]),
+            sha256=str(value["sha256"]),
+            size_bytes=int(value["size_bytes"]),
+        )
+
+
+__all__ = ["AttachmentRef"]

agent/agent/trace/protocols.py

@@ -8,6 +8,7 @@ from typing import Protocol, List, Optional, Dict, Any, runtime_checkable
 
 from .models import Trace, Message
 from .goal_models import GoalTree, Goal
+from .attachments import AttachmentRef
 
 
 @runtime_checkable
@@ -55,7 +56,7 @@ class TraceStore(Protocol):
         attempt_id: Optional[str] = None,
         validation_id: Optional[str] = None,
         operation_id: Optional[str] = None,
-        limit: int = 50
+        limit: int = 50,
     ) -> List[Trace]:
         """列出 Traces"""
         ...
@@ -151,9 +152,7 @@ class TraceStore(Protocol):
         ...
 
     async def get_main_path_messages(
-        self,
-        trace_id: str,
-        head_sequence: int
+        self, trace_id: str, head_sequence: int
     ) -> List[Message]:
         """
         获取主路径上的消息（从 head_sequence 沿 parent_sequence 链回溯到 root）
@@ -167,11 +166,7 @@ class TraceStore(Protocol):
         """
         ...
 
-    async def get_messages_by_goal(
-        self,
-        trace_id: str,
-        goal_id: str
-    ) -> List[Message]:
+    async def get_messages_by_goal(self, trace_id: str, goal_id: str) -> List[Message]:
         """
         获取指定 Goal 关联的所有 Messages
 
@@ -194,7 +189,9 @@ class TraceStore(Protocol):
         """
         ...
 
-    async def abandon_messages_after(self, trace_id: str, cutoff_sequence: int) -> List[str]:
+    async def abandon_messages_after(
+        self, trace_id: str, cutoff_sequence: int
+    ) -> List[str]:
         """
         将 cutoff_sequence 之后的所有 active 消息标记为 abandoned（回溯专用）
 
@@ -210,9 +207,7 @@ class TraceStore(Protocol):
     # ===== 事件流操作（用于 WebSocket 断线续传）=====
 
     async def get_events(
-        self,
-        trace_id: str,
-        since_event_id: int = 0
+        self, trace_id: str, since_event_id: int = 0
     ) -> List[Dict[str, Any]]:
         """
         获取事件流（用于 WS 断线续传）
@@ -227,10 +222,7 @@ class TraceStore(Protocol):
         ...
 
     async def append_event(
-        self,
-        trace_id: str,
-        event_type: str,
-        payload: Dict[str, Any]
+        self, trace_id: str, event_type: str, payload: Dict[str, Any]
     ) -> int:
         """
         追加事件，返回 event_id
@@ -244,3 +236,24 @@ class TraceStore(Protocol):
             event_id: 新事件的 ID
         """
         ...
+
+
+@runtime_checkable
+class TraceAttachmentStore(Protocol):
+    """Optional port for content-addressed message attachments."""
+
+    async def store_message_attachment(
+        self,
+        *,
+        trace_id: str,
+        message_id: str,
+        media_type: str,
+        content: bytes,
+        sha256: str,
+    ) -> AttachmentRef:
+        """Persist bytes after verifying their caller-supplied digest."""
+        ...
+
+    async def read_message_attachment(self, ref: AttachmentRef) -> bytes:
+        """Read bytes and verify the durable reference."""
+        ...

agent/agent/trace/store.py

@@ -20,12 +20,16 @@ Sub-Trace 是完全独立的 Trace，有自己的目录：
 └── events.jsonl
 """
 
+import hashlib
 import json
 import logging
+import os
+import uuid
+from datetime import datetime
 from pathlib import Path
 from typing import Dict, List, Optional, Any
-from datetime import datetime
 
+from .attachments import AttachmentRef
 from .models import Trace, Message
 from .goal_models import GoalTree, Goal, GoalStats
 
@@ -67,6 +71,10 @@ class FileSystemTraceStore:
         """获取 model_usage.json 文件路径"""
         return self._get_trace_dir(trace_id) / "model_usage.json"
 
+    def _get_attachments_dir(self, trace_id: str) -> Path:
+        """Return the durable attachment directory for one trace."""
+        return self._get_trace_dir(trace_id) / "attachments"
+
     # ===== Trace 操作 =====
 
     async def create_trace(self, trace: Trace) -> str:
@@ -78,9 +86,13 @@ class FileSystemTraceStore:
         messages_dir = self._get_messages_dir(trace.trace_id)
         messages_dir.mkdir(exist_ok=True)
 
+        self._get_attachments_dir(trace.trace_id).mkdir(exist_ok=True)
+
         # 写入 meta.json
         meta_file = self._get_meta_file(trace.trace_id)
-        meta_file.write_text(json.dumps(trace.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
+        meta_file.write_text(
+            json.dumps(trace.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8"
+        )
 
         # 创建空的 events.jsonl
         events_file = self._get_events_file(trace.trace_id)
@@ -88,6 +100,103 @@ class FileSystemTraceStore:
 
         return trace.trace_id
 
+    async def store_message_attachment(
+        self,
+        *,
+        trace_id: str,
+        message_id: str,
+        media_type: str,
+        content: bytes,
+        sha256: str,
+    ) -> AttachmentRef:
+        """Persist verified bytes without exposing filesystem paths."""
+        _validate_path_segment(trace_id, "trace_id")
+        _validate_path_segment(message_id, "message_id")
+        if not isinstance(content, bytes):
+            raise TypeError("content must be bytes")
+        actual = f"sha256:{hashlib.sha256(content).hexdigest()}"
+        if sha256 != actual:
+            raise ValueError("attachment digest mismatch")
+        if not self._get_meta_file(trace_id).exists():
+            raise ValueError(f"Trace not found: {trace_id}")
+        message = await self.get_message(message_id)
+        if message is None or message.trace_id != trace_id:
+            raise ValueError("attachment message does not belong to the Trace")
+
+        reference = AttachmentRef(
+            trace_id=trace_id,
+            message_id=message_id,
+            media_type=media_type,
+            sha256=actual,
+            size_bytes=len(content),
+        )
+
+        digest_hex = actual.removeprefix("sha256:")
+        attachment_dir = self._get_attachments_dir(trace_id) / message_id
+        attachment_dir.mkdir(parents=True, exist_ok=True)
+        target = attachment_dir / f"{digest_hex}.bin"
+        metadata = attachment_dir / f"{digest_hex}.json"
+        if target.exists():
+            if target.read_bytes() != content:
+                raise ValueError("attachment content conflict")
+        else:
+            temporary = attachment_dir / f".{digest_hex}.{uuid.uuid4().hex}.tmp"
+            try:
+                with temporary.open("xb") as output:
+                    output.write(content)
+                    output.flush()
+                    os.fsync(output.fileno())
+                os.replace(temporary, target)
+                _fsync_directory(attachment_dir)
+            finally:
+                temporary.unlink(missing_ok=True)
+
+        if metadata.exists():
+            try:
+                stored_reference = AttachmentRef.from_dict(
+                    json.loads(metadata.read_text(encoding="utf-8"))
+                )
+            except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+                raise ValueError("attachment metadata is invalid") from exc
+            if stored_reference != reference:
+                raise ValueError("attachment reference conflict")
+        else:
+            _write_atomic_json(metadata, reference.to_dict())
+            _fsync_directory(attachment_dir)
+
+        return reference
+
+    async def read_message_attachment(self, ref: AttachmentRef) -> bytes:
+        """Read and verify one content-addressed attachment."""
+        if not isinstance(ref, AttachmentRef):
+            raise TypeError("ref must be an AttachmentRef")
+        _validate_path_segment(ref.trace_id, "trace_id")
+        _validate_path_segment(ref.message_id, "message_id")
+        digest_hex = ref.sha256.removeprefix("sha256:")
+        target = (
+            self._get_attachments_dir(ref.trace_id)
+            / ref.message_id
+            / f"{digest_hex}.bin"
+        )
+        metadata = target.with_suffix(".json")
+        if not target.is_file():
+            raise FileNotFoundError("trace attachment not found")
+        if not metadata.is_file():
+            raise ValueError("trace attachment metadata not found")
+        try:
+            stored_reference = AttachmentRef.from_dict(
+                json.loads(metadata.read_text(encoding="utf-8"))
+            )
+        except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+            raise ValueError("trace attachment metadata is invalid") from exc
+        if stored_reference != ref:
+            raise ValueError("trace attachment reference does not match metadata")
+        content = target.read_bytes()
+        actual = f"sha256:{hashlib.sha256(content).hexdigest()}"
+        if actual != ref.sha256 or len(content) != ref.size_bytes:
+            raise ValueError("trace attachment integrity check failed")
+        return content
+
     async def get_trace(self, trace_id: str) -> Optional[Trace]:
         """获取 Trace"""
         meta_file = self._get_meta_file(trace_id)
@@ -117,7 +226,9 @@ class FileSystemTraceStore:
 
         # 写回文件
         meta_file = self._get_meta_file(trace_id)
-        meta_file.write_text(json.dumps(trace.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
+        meta_file.write_text(
+            json.dumps(trace.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8"
+        )
 
     async def list_traces(
         self,
@@ -132,7 +243,7 @@ class FileSystemTraceStore:
         attempt_id: Optional[str] = None,
         validation_id: Optional[str] = None,
         operation_id: Optional[str] = None,
-        limit: int = 50
+        limit: int = 50,
     ) -> List[Trace]:
         """列出 Traces"""
         traces = []
@@ -213,7 +324,9 @@ class FileSystemTraceStore:
     async def update_goal_tree(self, trace_id: str, tree: GoalTree) -> None:
         """更新完整 GoalTree"""
         goal_file = self._get_goal_file(trace_id)
-        goal_file.write_text(json.dumps(tree.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
+        goal_file.write_text(
+            json.dumps(tree.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8"
+        )
 
     async def add_goal(self, trace_id: str, goal: Goal) -> None:
         """添加 Goal 到 GoalTree"""
@@ -225,18 +338,21 @@ class FileSystemTraceStore:
         await self.update_goal_tree(trace_id, tree)
 
         # 推送 goal_added 事件
-        event_data = {
-            "goal": goal.to_dict(),
-            "parent_id": goal.parent_id
-        }
+        event_data = {"goal": goal.to_dict(), "parent_id": goal.parent_id}
         await self.append_event(trace_id, "goal_added", event_data)
 
         # 打印详细的 goal 信息
-        desc_preview = goal.description[:80] + "..." if len(goal.description) > 80 else goal.description
+        desc_preview = (
+            goal.description[:80] + "..."
+            if len(goal.description) > 80
+            else goal.description
+        )
         print(f"[Goal Added] ID={goal.id}, Parent={goal.parent_id or 'root'}")
         print(f"  📝 {desc_preview}")
         if goal.reason:
-            reason_preview = goal.reason[:60] + "..." if len(goal.reason) > 60 else goal.reason
+            reason_preview = (
+                goal.reason[:60] + "..." if len(goal.reason) > 60 else goal.reason
+            )
             print(f"  💡 {reason_preview}")
 
     async def update_goal(
@@ -259,7 +375,9 @@ class FileSystemTraceStore:
         for key, value in updates.items():
             if hasattr(goal, key):
                 # 特殊处理 stats 字段（可能是 dict）
-                if key in ["self_stats", "cumulative_stats"] and isinstance(value, dict):
+                if key in ["self_stats", "cumulative_stats"] and isinstance(
+                    value, dict
+                ):
                     value = GoalStats.from_dict(value)
                 setattr(goal, key, value)
 
@@ -274,12 +392,14 @@ class FileSystemTraceStore:
             cascade_completed = await self._check_cascade_completion(trace_id, goal)
             affected_goals.extend(cascade_completed)
 
-        await self.append_event(trace_id, "goal_updated", {
-            "goal_id": goal_id,
-            "updates": updates,
-            "affected_goals": affected_goals
-        })
-        print(f"[DEBUG] Pushed goal_updated event: goal_id={goal_id}, updates={updates}, affected={len(affected_goals)}")
+        await self.append_event(
+            trace_id,
+            "goal_updated",
+            {"goal_id": goal_id, "updates": updates, "affected_goals": affected_goals},
+        )
+        print(
+            f"[DEBUG] Pushed goal_updated event: goal_id={goal_id}, updates={updates}, affected={len(affected_goals)}"
+        )
 
         # Goal 完成时触发知识评估
         if updates.get("status") in ["completed", "abandoned"]:
@@ -293,12 +413,12 @@ class FileSystemTraceStore:
                     trace.context["pending_knowledge_eval"] = True
                     trace.context["knowledge_eval_trigger"] = "goal_completion"
                     await self.update_trace(trace_id, context=trace.context)
-                    logger.info(f"[Knowledge Eval] Goal {goal_id} 完成，设置评估标志，待评估知识: {len(pending)} 条")
+                    logger.info(
+                        f"[Knowledge Eval] Goal {goal_id} 完成，设置评估标志，待评估知识: {len(pending)} 条"
+                    )
 
     async def _check_cascade_completion(
-        self,
-        trace_id: str,
-        completed_goal: Goal
+        self, trace_id: str, completed_goal: Goal
     ) -> List[Dict[str, Any]]:
         """
         检查级联完成：如果一个 Goal 的所有子 Goal 都完成，则自动完成父 Goal
@@ -328,8 +448,7 @@ class FileSystemTraceStore:
 
         # 检查是否所有子 Goal 都已完成（排除 abandoned）
         all_completed = all(
-            child.status in ["completed", "abandoned"]
-            for child in children
+            child.status in ["completed", "abandoned"] for child in children
         )
 
         if all_completed and parent.status != "completed":
@@ -342,15 +461,19 @@ class FileSystemTraceStore:
 
             await self.update_goal_tree(trace_id, tree)
 
-            affected.append({
-                "goal_id": parent.id,
-                "status": "completed",
-                "summary": parent.summary,
-                "cumulative_stats": parent.cumulative_stats.to_dict()
-            })
+            affected.append(
+                {
+                    "goal_id": parent.id,
+                    "status": "completed",
+                    "summary": parent.summary,
+                    "cumulative_stats": parent.cumulative_stats.to_dict(),
+                }
+            )
 
             # 递归检查祖父 Goal
-            grandparent_affected = await self._check_cascade_completion(trace_id, parent)
+            grandparent_affected = await self._check_cascade_completion(
+                trace_id, parent
+            )
             affected.extend(grandparent_affected)
 
         return affected
@@ -368,7 +491,10 @@ class FileSystemTraceStore:
         # 1. 写入 message 文件
         messages_dir = self._get_messages_dir(trace_id)
         message_file = messages_dir / f"{message.message_id}.json"
-        message_file.write_text(json.dumps(message.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
+        message_file.write_text(
+            json.dumps(message.to_dict(), indent=2, ensure_ascii=False),
+            encoding="utf-8",
+        )
 
         # 2. 更新 trace 统计
         trace = await self.get_trace(trace_id)
@@ -392,7 +518,9 @@ class FileSystemTraceStore:
             if message.tokens:
                 trace.total_tokens += message.tokens
             elif message.prompt_tokens or message.completion_tokens:
-                trace.total_tokens += (message.prompt_tokens or 0) + (message.completion_tokens or 0)
+                trace.total_tokens += (message.prompt_tokens or 0) + (
+                    message.completion_tokens or 0
+                )
 
             if message.cost:
                 trace.total_cost += message.cost
@@ -411,7 +539,7 @@ class FileSystemTraceStore:
                 total_cache_creation_tokens=trace.total_cache_creation_tokens,
                 total_cache_read_tokens=trace.total_cache_read_tokens,
                 total_cost=trace.total_cost,
-                total_duration_ms=trace.total_duration_ms
+                total_duration_ms=trace.total_duration_ms,
             )
 
         # 3. 更新 Goal stats
@@ -419,10 +547,11 @@ class FileSystemTraceStore:
 
         # 4. 追加 message_added 事件
         affected_goals = await self._get_affected_goals(trace_id, message)
-        event_id = await self.append_event(trace_id, "message_added", {
-            "message": message.to_dict(),
-            "affected_goals": affected_goals
-        })
+        event_id = await self.append_event(
+            trace_id,
+            "message_added",
+            {"message": message.to_dict(), "affected_goals": affected_goals},
+        )
         if event_id:
             try:
                 from . import websocket as trace_ws
@@ -434,7 +563,11 @@ class FileSystemTraceStore:
                     affected_goals=affected_goals,
                 )
             except Exception:
-                logger.exception("Failed to broadcast message_added (trace_id=%s, event_id=%s)", trace_id, event_id)
+                logger.exception(
+                    "Failed to broadcast message_added (trace_id=%s, event_id=%s)",
+                    trace_id,
+                    event_id,
+                )
 
         return message.message_id
 
@@ -483,7 +616,9 @@ class FileSystemTraceStore:
         # 保存更新后的 tree
         await self.update_goal_tree(trace_id, tree)
 
-    async def _get_affected_goals(self, trace_id: str, message: Message) -> List[Dict[str, Any]]:
+    async def _get_affected_goals(
+        self, trace_id: str, message: Message
+    ) -> List[Dict[str, Any]]:
         """获取受影响的 Goals（自身 + 所有祖先）"""
         tree = await self.get_goal_tree(trace_id)
 
@@ -497,11 +632,13 @@ class FileSystemTraceStore:
         affected = []
 
         # 添加自身（包含 self_stats 和 cumulative_stats）
-        affected.append({
-            "goal_id": goal.id,
-            "self_stats": goal.self_stats.to_dict(),
-            "cumulative_stats": goal.cumulative_stats.to_dict()
-        })
+        affected.append(
+            {
+                "goal_id": goal.id,
+                "self_stats": goal.self_stats.to_dict(),
+                "cumulative_stats": goal.cumulative_stats.to_dict(),
+            }
+        )
 
         # 添加所有祖先（仅 cumulative_stats）
         current_goal = goal
@@ -510,10 +647,12 @@ class FileSystemTraceStore:
             if not parent:
                 break
 
-            affected.append({
-                "goal_id": parent.id,
-                "cumulative_stats": parent.cumulative_stats.to_dict()
-            })
+            affected.append(
+                {
+                    "goal_id": parent.id,
+                    "cumulative_stats": parent.cumulative_stats.to_dict(),
+                }
+            )
 
             current_goal = parent
 
@@ -560,9 +699,7 @@ class FileSystemTraceStore:
         return messages
 
     async def get_main_path_messages(
-        self,
-        trace_id: str,
-        head_sequence: int
+        self, trace_id: str, head_sequence: int
     ) -> List[Message]:
         """
         获取从 head_sequence 沿 parent_sequence 链回溯到 root 的完整路径
@@ -592,11 +729,7 @@ class FileSystemTraceStore:
         path.reverse()
         return path
 
-    async def get_messages_by_goal(
-        self,
-        trace_id: str,
-        goal_id: str
-    ) -> List[Message]:
+    async def get_messages_by_goal(self, trace_id: str, goal_id: str) -> List[Message]:
         """获取指定 Goal 关联的所有 Messages"""
         all_messages = await self.get_trace_messages(trace_id)
         return [m for m in all_messages if m.goal_id == goal_id]
@@ -615,9 +748,14 @@ class FileSystemTraceStore:
         # 确定文件路径
         messages_dir = self._get_messages_dir(message.trace_id)
         message_file = messages_dir / f"{message_id}.json"
-        message_file.write_text(json.dumps(message.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
+        message_file.write_text(
+            json.dumps(message.to_dict(), indent=2, ensure_ascii=False),
+            encoding="utf-8",
+        )
 
-    async def abandon_messages_after(self, trace_id: str, cutoff_sequence: int) -> List[str]:
+    async def abandon_messages_after(
+        self, trace_id: str, cutoff_sequence: int
+    ) -> List[str]:
         """
         将 sequence > cutoff_sequence 的 active messages 标记为 abandoned。
         返回被 abandon 的 message_id 列表。
@@ -631,10 +769,12 @@ class FileSystemTraceStore:
                 msg.status = "abandoned"
                 msg.abandoned_at = now
                 # 直接写回文件
-                message_file = self._get_messages_dir(trace_id) / f"{msg.message_id}.json"
+                message_file = (
+                    self._get_messages_dir(trace_id) / f"{msg.message_id}.json"
+                )
                 message_file.write_text(
                     json.dumps(msg.to_dict(), indent=2, ensure_ascii=False),
-                    encoding="utf-8"
+                    encoding="utf-8",
                 )
                 abandoned_ids.append(msg.message_id)
 
@@ -710,15 +850,17 @@ class FileSystemTraceStore:
             model_entry["cache_read_tokens"] += cache_read_tokens
             model_entry["call_count"] += 1
         else:
-            data["models"].append({
-                "model": model,
-                "source": source,
-                "prompt_tokens": prompt_tokens,
-                "completion_tokens": completion_tokens,
-                "total_tokens": total_tokens,
-                "cache_read_tokens": cache_read_tokens,
-                "call_count": 1,
-            })
+            data["models"].append(
+                {
+                    "model": model,
+                    "source": source,
+                    "prompt_tokens": prompt_tokens,
+                    "completion_tokens": completion_tokens,
+                    "total_tokens": total_tokens,
+                    "cache_read_tokens": cache_read_tokens,
+                    "call_count": 1,
+                }
+            )
             data["summary"]["total_models"] = len(data["models"])
 
         # 添加到timeline
@@ -736,14 +878,14 @@ class FileSystemTraceStore:
         data["timeline"].append(timeline_entry)
 
         # 写回文件
-        usage_file.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+        usage_file.write_text(
+            json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8"
+        )
 
     # ===== 事件流操作（用于 WebSocket 断线续传）=====
 
     async def get_events(
-        self,
-        trace_id: str,
-        since_event_id: int = 0
+        self, trace_id: str, since_event_id: int = 0
     ) -> List[Dict[str, Any]]:
         """获取事件流"""
         events_file = self._get_events_file(trace_id)
@@ -751,7 +893,7 @@ class FileSystemTraceStore:
             return []
 
         events = []
-        with events_file.open('r', encoding='utf-8') as f:
+        with events_file.open("r", encoding="utf-8") as f:
             for line in f:
                 try:
                     event = json.loads(line.strip())
@@ -763,10 +905,7 @@ class FileSystemTraceStore:
         return events
 
     async def append_event(
-        self,
-        trace_id: str,
-        event_type: str,
-        payload: Dict[str, Any]
+        self, trace_id: str, event_type: str, payload: Dict[str, Any]
     ) -> int:
         """追加事件，返回 event_id"""
         # 获取 trace 并递增 event_id
@@ -785,13 +924,13 @@ class FileSystemTraceStore:
             "event_id": event_id,
             "event": event_type,
             "ts": datetime.now().isoformat(),
-            **payload
+            **payload,
         }
 
         # 追加到 events.jsonl
         events_file = self._get_events_file(trace_id)
-        with events_file.open('a', encoding='utf-8') as f:
-            f.write(json.dumps(event, ensure_ascii=False) + '\n')
+        with events_file.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(event, ensure_ascii=False) + "\n")
 
         return event_id
 
@@ -872,7 +1011,9 @@ class FileSystemTraceStore:
         event["timestamp"] = datetime.now().isoformat()
         log["events"].append(event)
         log_file = self._get_cognition_log_file(trace_id)
-        log_file.write_text(json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8")
+        log_file.write_text(
+            json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8"
+        )
 
     async def append_knowledge_entry(
         self,
@@ -881,7 +1022,7 @@ class FileSystemTraceStore:
         goal_id: str,
         injected_at_sequence: int,
         task: str,
-        content: str
+        content: str,
     ) -> None:
         """兼容旧接口：追加知识注入记录（转换为 query 事件）"""
         await self.append_cognition_event(
@@ -893,8 +1034,10 @@ class FileSystemTraceStore:
                 "query": task,
                 "response": "",
                 "source_ids": [knowledge_id],
-                "sources": [{"id": knowledge_id, "task": task, "content": content[:500]}],
-            }
+                "sources": [
+                    {"id": knowledge_id, "task": task, "content": content[:500]}
+                ],
+            },
         )
 
     async def update_knowledge_evaluation(
@@ -902,7 +1045,7 @@ class FileSystemTraceStore:
         trace_id: str,
         knowledge_id: str,
         eval_result: Dict[str, Any],
-        trigger_event: str
+        trigger_event: str,
     ) -> None:
         """更新知识评估结果（兼容旧格式 + 新 cognition_log 格式）
 
@@ -915,23 +1058,30 @@ class FileSystemTraceStore:
         # 旧格式兼容：直接更新 entries 中的 eval_result 字段
         if "entries" in log:
             matching = [
-                (i, e) for i, e in enumerate(log["entries"])
-                if e.get("knowledge_id") == knowledge_id and e.get("eval_result") is None
+                (i, e)
+                for i, e in enumerate(log["entries"])
+                if e.get("knowledge_id") == knowledge_id
+                and e.get("eval_result") is None
             ]
             if matching:
-                matching.sort(key=lambda x: x[1].get("injected_at_sequence", 0), reverse=True)
+                matching.sort(
+                    key=lambda x: x[1].get("injected_at_sequence", 0), reverse=True
+                )
                 _, entry = matching[0]
                 entry["eval_result"] = eval_result
                 entry["evaluated_at"] = datetime.now().isoformat()
                 entry["evaluated_at_trigger"] = trigger_event
                 log_file = self._get_knowledge_log_file(trace_id)
-                log_file.write_text(json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8")
+                log_file.write_text(
+                    json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8"
+                )
                 return
 
         # 新格式：追加 evaluation 事件
         # 找到包含该 knowledge_id 的最近 query 事件
         query_events = [
-            e for e in events
+            e
+            for e in events
             if e.get("type") == "query" and knowledge_id in e.get("source_ids", [])
         ]
         query_sequence = query_events[-1]["sequence"] if query_events else None
@@ -944,12 +1094,18 @@ class FileSystemTraceStore:
                 "query_sequence": query_sequence,
                 "trigger": trigger_event,
                 "assessments": [
-                    {"source_id": knowledge_id, "status": eval_result.get("eval_status", ""), "reason": eval_result.get("reason", "")}
+                    {
+                        "source_id": knowledge_id,
+                        "status": eval_result.get("eval_status", ""),
+                        "reason": eval_result.get("reason", ""),
+                    }
                 ],
-            }
+            },
         )
 
-    async def get_pending_knowledge_entries(self, trace_id: str) -> List[Dict[str, Any]]:
+    async def get_pending_knowledge_entries(
+        self, trace_id: str
+    ) -> List[Dict[str, Any]]:
         """获取所有待评估的知识条目（兼容旧格式 + 新格式）"""
         log = await self.get_cognition_log(trace_id)
 
@@ -970,21 +1126,20 @@ class FileSystemTraceStore:
             if qe.get("sequence") not in evaluated_sequences:
                 # 转为旧格式兼容（runner 中的评估逻辑期望此格式）
                 for source in qe.get("sources", []):
-                    pending.append({
-                        "knowledge_id": source.get("id", ""),
-                        "goal_id": qe.get("goal_id", ""),
-                        "injected_at_sequence": qe.get("sequence", 0),
-                        "task": source.get("task", ""),
-                        "content": source.get("content", ""),
-                        "query_sequence": qe.get("sequence"),
-                    })
+                    pending.append(
+                        {
+                            "knowledge_id": source.get("id", ""),
+                            "goal_id": qe.get("goal_id", ""),
+                            "injected_at_sequence": qe.get("sequence", 0),
+                            "task": source.get("task", ""),
+                            "content": source.get("content", ""),
+                            "query_sequence": qe.get("sequence"),
+                        }
+                    )
         return pending
 
     async def update_user_feedback(
-        self,
-        trace_id: str,
-        knowledge_id: str,
-        user_feedback: Dict[str, Any]
+        self, trace_id: str, knowledge_id: str, user_feedback: Dict[str, Any]
     ) -> None:
         """记录用户对知识的反馈（confirm/override）"""
         log = await self.get_cognition_log(trace_id)
@@ -992,15 +1147,20 @@ class FileSystemTraceStore:
         # 旧格式
         if "entries" in log:
             matching = [
-                (i, e) for i, e in enumerate(log["entries"])
+                (i, e)
+                for i, e in enumerate(log["entries"])
                 if e.get("knowledge_id") == knowledge_id
             ]
             if matching:
-                matching.sort(key=lambda x: x[1].get("injected_at_sequence", 0), reverse=True)
+                matching.sort(
+                    key=lambda x: x[1].get("injected_at_sequence", 0), reverse=True
+                )
                 _, entry = matching[0]
                 entry["user_feedback"] = user_feedback
             log_file = self._get_knowledge_log_file(trace_id)
-            log_file.write_text(json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8")
+            log_file.write_text(
+                json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8"
+            )
             return
 
         # 新格式：追加 user_feedback 事件（或直接记录在 evaluation 上）
@@ -1010,5 +1170,37 @@ class FileSystemTraceStore:
                 "type": "user_feedback",
                 "knowledge_id": knowledge_id,
                 "feedback": user_feedback,
-            }
+            },
         )
+
+
+def _validate_path_segment(value: str, field_name: str) -> None:
+    if (
+        not isinstance(value, str)
+        or not value
+        or value in {".", ".."}
+        or Path(value).name != value
+        or "/" in value
+        or "\\" in value
+    ):
+        raise ValueError(f"{field_name} must be a safe path segment")
+
+
+def _fsync_directory(path: Path) -> None:
+    descriptor = os.open(path, os.O_RDONLY)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def _write_atomic_json(path: Path, value: Dict[str, Any]) -> None:
+    temporary = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp"
+    try:
+        with temporary.open("x", encoding="utf-8") as output:
+            json.dump(value, output, ensure_ascii=False, sort_keys=True)
+            output.flush()
+            os.fsync(output.fileno())
+        os.replace(temporary, path)
+    finally:
+        temporary.unlink(missing_ok=True)

agent/tests/test_compression_policy_persistence.py

@@ -0,0 +1,350 @@
+from __future__ import annotations
+
+import pytest
+
+from agent.core.runner import AgentRunner, RunConfig
+from agent.tools.builtin.knowledge import KnowledgeConfig
+from agent.trace.models import Message, Trace
+from agent.trace.store import FileSystemTraceStore
+
+
+async def _add(store, *, trace_id, role, sequence, parent, content):
+    message = Message.create(
+        trace_id=trace_id,
+        role=role,
+        sequence=sequence,
+        parent_sequence=parent,
+        content=content,
+    )
+    await store.add_message(message)
+    return message
+
+
+@pytest.mark.asyncio
+async def test_level2_compression_carries_active_system_policies_across_reload(
+    tmp_path,
+):
+    store = FileSystemTraceStore(str(tmp_path))
+    await store.create_trace(Trace(trace_id="trace", mode="agent"))
+    initial = await _add(
+        store,
+        trace_id="trace",
+        role="system",
+        sequence=1,
+        parent=None,
+        content="initial framework policy",
+    )
+    first_user = await _add(
+        store,
+        trace_id="trace",
+        role="user",
+        sequence=2,
+        parent=1,
+        content="mission",
+    )
+    await _add(
+        store,
+        trace_id="trace",
+        role="assistant",
+        sequence=3,
+        parent=2,
+        content={"text": "old work"},
+    )
+    await _add(
+        store,
+        trace_id="trace",
+        role="system",
+        sequence=4,
+        parent=3,
+        content="phase two policy",
+    )
+    await _add(
+        store,
+        trace_id="trace",
+        role="user",
+        sequence=5,
+        parent=4,
+        content="continue",
+    )
+    await store.update_trace("trace", head_sequence=5)
+    runner = AgentRunner(trace_store=store)
+
+    summary, next_sequence, carried = await runner._persist_compressed_main_path(
+        trace_id="trace",
+        original_head_sequence=5,
+        next_sequence=10,
+        summary_content="summary one",
+    )
+    await store.update_trace("trace", head_sequence=summary.sequence)
+
+    assert summary.sequence == 11
+    assert next_sequence == 12
+    in_memory = runner._rebuild_history_after_compression(
+        [
+            initial.to_llm_dict(),
+            first_user.to_llm_dict(),
+            {"role": "assistant", "content": "old work"},
+            {"role": "system", "content": "phase two policy"},
+        ],
+        summary.to_llm_dict(),
+        carried_system_messages=carried,
+    )
+    assert [item["role"] for item in in_memory] == [
+        "system",
+        "user",
+        "system",
+        "user",
+    ]
+    assert in_memory[2]["content"] == "phase two policy"
+
+    reloaded = FileSystemTraceStore(str(tmp_path))
+    trace = await reloaded.get_trace("trace")
+    main_path = await reloaded.get_main_path_messages("trace", trace.head_sequence)
+    assert [message.role for message in main_path] == [
+        "system",
+        "user",
+        "system",
+        "user",
+    ]
+    assert [message.content for message in main_path if message.role == "system"] == [
+        "initial framework policy",
+        "phase two policy",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_level2_compression_does_not_duplicate_systems_before_first_user(
+    tmp_path,
+):
+    store = FileSystemTraceStore(str(tmp_path))
+    await store.create_trace(Trace(trace_id="trace", mode="agent"))
+    parent = None
+    for sequence, role, content in (
+        (1, "system", "initial"),
+        (2, "system", "startup policy"),
+        (3, "user", "mission"),
+        (4, "system", "phase two policy"),
+        (5, "user", "continue"),
+    ):
+        await _add(
+            store,
+            trace_id="trace",
+            role=role,
+            sequence=sequence,
+            parent=parent,
+            content=content,
+        )
+        parent = sequence
+
+    runner = AgentRunner(trace_store=store)
+    summary, _, carried = await runner._persist_compressed_main_path(
+        trace_id="trace",
+        original_head_sequence=5,
+        next_sequence=10,
+        summary_content="summary",
+    )
+    await store.update_trace("trace", head_sequence=summary.sequence)
+    path = await store.get_main_path_messages("trace", summary.sequence)
+
+    assert [item["content"] for item in carried] == ["phase two policy"]
+    assert [message.content for message in path if message.role == "system"] == [
+        "initial",
+        "startup policy",
+        "phase two policy",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_repeated_level2_compression_preserves_policy_order_without_growth(
+    tmp_path,
+):
+    store = FileSystemTraceStore(str(tmp_path))
+    await store.create_trace(Trace(trace_id="trace", mode="agent"))
+    parent = None
+    for sequence, role, content in (
+        (1, "system", "initial"),
+        (2, "user", "mission"),
+        (3, "system", "phase one"),
+        (4, "user", "work"),
+    ):
+        await _add(
+            store,
+            trace_id="trace",
+            role=role,
+            sequence=sequence,
+            parent=parent,
+            content=content,
+        )
+        parent = sequence
+    runner = AgentRunner(trace_store=store)
+    summary_one, _, _ = await runner._persist_compressed_main_path(
+        trace_id="trace",
+        original_head_sequence=4,
+        next_sequence=10,
+        summary_content="summary one",
+    )
+    await _add(
+        store,
+        trace_id="trace",
+        role="system",
+        sequence=12,
+        parent=summary_one.sequence,
+        content="phase two",
+    )
+    await _add(
+        store,
+        trace_id="trace",
+        role="user",
+        sequence=13,
+        parent=12,
+        content="more work",
+    )
+
+    summary_two, _, _ = await runner._persist_compressed_main_path(
+        trace_id="trace",
+        original_head_sequence=13,
+        next_sequence=20,
+        summary_content="summary two",
+    )
+    await store.update_trace("trace", head_sequence=summary_two.sequence)
+    path = await store.get_main_path_messages("trace", summary_two.sequence)
+
+    assert [message.content for message in path if message.role == "system"] == [
+        "initial",
+        "phase one",
+        "phase two",
+    ]
+    assert [message.role for message in path] == [
+        "system",
+        "user",
+        "system",
+        "system",
+        "user",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_real_runner_level2_branch_persists_policy_across_store_reload(tmp_path):
+    calls = 0
+
+    async def llm_call(**kwargs):
+        nonlocal calls
+        calls += 1
+        messages = kwargs["messages"]
+        if any(
+            "[[SUMMARY]]" in str(message.get("content", "")) for message in messages
+        ):
+            return {
+                "content": "[[SUMMARY]] durable compressed state",
+                "tool_calls": None,
+                "finish_reason": "stop",
+            }
+        return {"content": "done", "tool_calls": None, "finish_reason": "stop"}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    runner = AgentRunner(trace_store=store, llm_call=llm_call)
+    config = RunConfig(
+        new_trace_id="real-compression-trace",
+        name="compression test",
+        max_iterations=4,
+        side_branch_max_turns=1,
+        force_side_branch=["compression"],
+        goal_compression="none",
+        knowledge=KnowledgeConfig(
+            enable_extraction=False,
+            enable_completion_extraction=False,
+            enable_injection=False,
+        ),
+    )
+
+    events = [
+        event
+        async for event in runner.run(
+            [
+                {"role": "system", "content": "initial framework policy"},
+                {"role": "user", "content": "mission"},
+                {"role": "system", "content": "phase two policy"},
+            ],
+            config,
+        )
+    ]
+
+    assert calls >= 2
+    assert events[-1].status == "completed"
+    reloaded = FileSystemTraceStore(str(tmp_path))
+    trace = await reloaded.get_trace("real-compression-trace")
+    main_path = await reloaded.get_main_path_messages(
+        trace.trace_id, trace.head_sequence
+    )
+    system_contents = [
+        str(message.content) for message in main_path if message.role == "system"
+    ]
+    assert len(system_contents) == 2
+    assert "initial framework policy" in system_contents[0]
+    assert "phase two policy" in system_contents[1]
+    assert any(
+        message.role == "user" and "durable compressed state" in str(message.content)
+        for message in main_path
+    )
+
+
+@pytest.mark.asyncio
+async def test_compression_does_not_reactivate_policy_outside_current_main_path(
+    tmp_path,
+):
+    store = FileSystemTraceStore(str(tmp_path))
+    await store.create_trace(Trace(trace_id="trace", mode="agent"))
+    await _add(
+        store,
+        trace_id="trace",
+        role="system",
+        sequence=1,
+        parent=None,
+        content="initial",
+    )
+    await _add(
+        store,
+        trace_id="trace",
+        role="user",
+        sequence=2,
+        parent=1,
+        content="mission",
+    )
+    await _add(
+        store,
+        trace_id="trace",
+        role="system",
+        sequence=3,
+        parent=2,
+        content="active phase policy",
+    )
+    side_policy = Message.create(
+        trace_id="trace",
+        role="system",
+        sequence=4,
+        parent_sequence=3,
+        branch_type="compression",
+        branch_id="side",
+        content="side branch policy must stay inactive",
+    )
+    await store.add_message(side_policy)
+    await _add(
+        store,
+        trace_id="trace",
+        role="user",
+        sequence=5,
+        parent=3,
+        content="main path continues",
+    )
+    runner = AgentRunner(trace_store=store)
+
+    summary, _, _ = await runner._persist_compressed_main_path(
+        trace_id="trace",
+        original_head_sequence=5,
+        next_sequence=10,
+        summary_content="summary",
+    )
+    path = await store.get_main_path_messages("trace", summary.sequence)
+
+    policies = [str(message.content) for message in path if message.role == "system"]
+    assert policies == ["initial", "active phase policy"]

agent/tests/test_orchestration_run_config_resolver.py

@@ -10,9 +10,9 @@ from agent import (
     AgentRole,
     FileSystemArtifactStore,
     FileSystemTaskStore,
-    RunConfig,
     RoleRunConfigOverrides,
     RoleRunConfigResolver,
+    RunConfig,
     wire_orchestration,
 )
 from agent.core.prompts.orchestration import PLANNER_ROLE_CONTRACT
@@ -278,5 +278,11 @@ async def test_wire_orchestration_injects_role_run_config_resolver(tmp_path):
 
 
 def test_planner_contract_allows_host_specific_dispatch_tool_name():
-    assert "dispatch_tasks" not in PLANNER_ROLE_CONTRACT
-    assert "task-dispatch tool exposed by your preset" in PLANNER_ROLE_CONTRACT
+    for framework_tool_name in ("task_plan", "task_decide", "dispatch_tasks"):
+        assert framework_tool_name not in PLANNER_ROLE_CONTRACT
+    assert "tools exposed by your preset" in PLANNER_ROLE_CONTRACT
+    assert "preset-specific" in PLANNER_ROLE_CONTRACT
+    assert "plan-inspection" in PLANNER_ROLE_CONTRACT
+    assert "explicit host capability boundary may require Root BLOCK" in (
+        PLANNER_ROLE_CONTRACT
+    )

agent/tests/test_orchestration_system_prompt_resolver.py

@@ -0,0 +1,388 @@
+from __future__ import annotations
+
+import asyncio
+from hashlib import sha256
+from types import SimpleNamespace
+from typing import Any, Mapping
+
+import pytest
+
+from agent import (
+    AgentPreset,
+    AgentRole,
+    AgentRunner,
+    FileSystemArtifactStore,
+    FileSystemTaskStore,
+    FileSystemTraceStore,
+    RoleSystemPromptOverride,
+    RoleSystemPromptResolver,
+    wire_orchestration,
+)
+from agent.core.presets import register_preset
+from agent.orchestration.executor import LocalAgentExecutor
+from agent.orchestration.models import FailureCode
+
+
+def _override(content: str) -> RoleSystemPromptOverride:
+    return RoleSystemPromptOverride(
+        content=content,
+        prompt_identity=f"sha256:{sha256(content.encode('utf-8')).hexdigest()}",
+    )
+
+
+def _worker_context(**updates: Any) -> dict[str, Any]:
+    context = {
+        "worker_preset": "script_worker",
+        "worker_trace_id": "worker-trace",
+        "root_trace_id": "root",
+        "task_id": "task",
+        "spec_version": 1,
+        "attempt_id": "attempt",
+        "task_spec": {"objective": "write"},
+        "continue_trace_id": None,
+    }
+    context.update(updates)
+    return context
+
+
+def _validator_context(**updates: Any) -> dict[str, Any]:
+    context = {
+        "validator_preset": "script_validator",
+        "validator_trace_id": "validator-trace",
+        "root_trace_id": "root",
+        "task_id": "task",
+        "spec_version": 1,
+        "attempt_id": "attempt",
+        "snapshot_id": "snapshot",
+        "validation_id": "validation",
+        "task_spec": {"objective": "validate"},
+        "artifact_snapshot": {"artifact_refs": []},
+    }
+    context.update(updates)
+    return context
+
+
+class PromptResolver:
+    def __init__(self, override: RoleSystemPromptOverride | None) -> None:
+        self.override = override
+        self.calls: list[tuple[AgentRole, str, Mapping[str, Any]]] = []
+
+    async def resolve(self, *, role, preset, context):
+        with pytest.raises(TypeError):
+            context["unsafe"] = True
+        self.calls.append((role, preset, context))
+        return self.override
+
+
+class CapturingRunner:
+    trace_store = None
+    task_coordinator = None
+
+    def __init__(self) -> None:
+        self.configs = []
+
+    async def run_result(self, *, messages, config):
+        self.configs.append(config)
+        return {"status": "completed", "summary": "done"}
+
+
+def test_system_prompt_types_are_public_and_validate_content_identity():
+    assert RoleSystemPromptResolver.__name__ == "RoleSystemPromptResolver"
+    value = _override("frozen prompt")
+    assert value.content == "frozen prompt"
+    with pytest.raises(ValueError, match="does not match"):
+        RoleSystemPromptOverride(
+            content="frozen prompt",
+            prompt_identity=f"sha256:{'0' * 64}",
+        )
+    with pytest.raises(ValueError, match="lowercase"):
+        RoleSystemPromptOverride(content="x", prompt_identity=f"sha256:{'A' * 64}")
+
+
+@pytest.mark.asyncio
+async def test_new_role_trace_receives_only_prompt_and_protected_identity():
+    runner = CapturingRunner()
+    resolver = PromptResolver(_override("business policy"))
+
+    result = await LocalAgentExecutor(
+        runner, role_system_prompt_resolver=resolver
+    ).run_worker(_worker_context(untrusted="not protected"))
+
+    assert result.status == "completed"
+    assert resolver.calls[0][0:2] == (AgentRole.WORKER, "script_worker")
+    config = runner.configs[0]
+    assert config.system_prompt == "business policy"
+    assert (
+        config.context["role_prompt_identity"]
+        == _override("business policy").prompt_identity
+    )
+    assert "untrusted" not in config.context
+    assert config.tools is None and config.tool_groups is None
+
+
+@pytest.mark.asyncio
+async def test_new_validator_trace_receives_frozen_prompt_identity():
+    runner = CapturingRunner()
+    expected = _override("validator business policy")
+    resolver = PromptResolver(expected)
+
+    result = await LocalAgentExecutor(
+        runner, role_system_prompt_resolver=resolver
+    ).run_validator(_validator_context(untrusted="not protected"))
+
+    assert result.status == "completed"
+    assert resolver.calls[0][0:2] == (AgentRole.VALIDATOR, "script_validator")
+    config = runner.configs[0]
+    assert config.system_prompt == expected.content
+    assert config.context["role_prompt_identity"] == expected.prompt_identity
+    assert config.context["validation_id"] == "validation"
+    assert "untrusted" not in config.context
+
+
+@pytest.mark.asyncio
+async def test_missing_or_empty_prompt_resolver_keeps_existing_defaults():
+    no_resolver_runner = CapturingRunner()
+    empty_resolver_runner = CapturingRunner()
+
+    no_resolver = await LocalAgentExecutor(no_resolver_runner).run_worker(
+        _worker_context()
+    )
+    empty_resolver = await LocalAgentExecutor(
+        empty_resolver_runner,
+        role_system_prompt_resolver=PromptResolver(None),
+    ).run_worker(_worker_context())
+
+    assert no_resolver.status == empty_resolver.status == "completed"
+    assert no_resolver_runner.configs[0].system_prompt is None
+    assert empty_resolver_runner.configs[0].system_prompt is None
+    assert "role_prompt_identity" not in no_resolver_runner.configs[0].context
+    assert "role_prompt_identity" not in empty_resolver_runner.configs[0].context
+
+
+@pytest.mark.asyncio
+async def test_invalid_prompt_resolver_result_is_an_executor_failure():
+    class InvalidResolver:
+        async def resolve(self, **kwargs):
+            return {"content": "unsafe", "prompt_identity": "forged"}
+
+    runner = CapturingRunner()
+    result = await LocalAgentExecutor(
+        runner, role_system_prompt_resolver=InvalidResolver()
+    ).run_worker(_worker_context())
+
+    assert result.status == "failed"
+    assert "must return RoleSystemPromptOverride or None" in result.error
+    assert runner.configs == []
+
+
+@pytest.mark.asyncio
+async def test_prompt_resolver_exception_is_normalized_as_executor_failure():
+    class BrokenResolver:
+        async def resolve(self, **kwargs):
+            raise RuntimeError("frozen prompt manifest unavailable")
+
+    runner = CapturingRunner()
+    result = await LocalAgentExecutor(
+        runner, role_system_prompt_resolver=BrokenResolver()
+    ).run_validator(_validator_context())
+
+    assert result.status == "failed"
+    assert result.error == "frozen prompt manifest unavailable"
+    assert result.execution_stats.failure_code == FailureCode.EXECUTOR_ERROR
+    assert runner.configs == []
+
+
+@pytest.mark.asyncio
+async def test_concurrent_roles_keep_distinct_frozen_prompts_and_identities():
+    class PerTaskResolver:
+        async def resolve(self, *, role, preset, context):
+            del preset
+            await asyncio.sleep(0)
+            return _override(f"{role.value} policy for {context['task_id']}")
+
+    runner = CapturingRunner()
+    executor = LocalAgentExecutor(runner, role_system_prompt_resolver=PerTaskResolver())
+
+    worker, validator = await asyncio.gather(
+        executor.run_worker(
+            _worker_context(task_id="worker-task", worker_trace_id="worker-trace")
+        ),
+        executor.run_validator(
+            _validator_context(
+                task_id="validator-task", validator_trace_id="validator-trace"
+            )
+        ),
+    )
+
+    assert worker.status == validator.status == "completed"
+    configs = {config.name: config for config in runner.configs}
+    worker_config = configs["Worker worker-task"]
+    validator_config = configs["Validator validator-task"]
+    assert worker_config.system_prompt == "worker policy for worker-task"
+    assert validator_config.system_prompt == "validator policy for validator-task"
+    assert (
+        worker_config.context["role_prompt_identity"]
+        == _override("worker policy for worker-task").prompt_identity
+    )
+    assert (
+        validator_config.context["role_prompt_identity"]
+        == _override("validator policy for validator-task").prompt_identity
+    )
+
+
+@pytest.mark.asyncio
+async def test_real_runner_persists_worker_and_validator_prompt_identity(tmp_path):
+    worker_preset = "test_prompt_identity_worker"
+    validator_preset = "test_prompt_identity_validator"
+    register_preset(
+        worker_preset,
+        AgentPreset(role=AgentRole.WORKER, allowed_tools=[], max_iterations=1),
+    )
+    register_preset(
+        validator_preset,
+        AgentPreset(role=AgentRole.VALIDATOR, allowed_tools=[], max_iterations=1),
+    )
+
+    async def llm_call(**_kwargs):
+        return {"content": "no terminal tool", "tool_calls": None}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    runner = AgentRunner(
+        trace_store=store,
+        llm_call=llm_call,
+        task_coordinator=object(),
+    )
+
+    class PerRoleResolver:
+        async def resolve(self, *, role, preset, context):
+            del preset, context
+            return _override(f"frozen {role.value} prompt")
+
+    executor = LocalAgentExecutor(runner, role_system_prompt_resolver=PerRoleResolver())
+    worker = await executor.run_worker(_worker_context(worker_preset=worker_preset))
+    validator = await executor.run_validator(
+        _validator_context(validator_preset=validator_preset)
+    )
+
+    assert worker.status == validator.status == "failed"
+    worker_trace = await store.get_trace("worker-trace")
+    validator_trace = await store.get_trace("validator-trace")
+    assert (
+        worker_trace.context["role_prompt_identity"]
+        == _override("frozen worker prompt").prompt_identity
+    )
+    assert (
+        validator_trace.context["role_prompt_identity"]
+        == _override("frozen validator prompt").prompt_identity
+    )
+    worker_path = await store.get_main_path_messages(
+        worker_trace.trace_id, worker_trace.head_sequence
+    )
+    validator_path = await store.get_main_path_messages(
+        validator_trace.trace_id, validator_trace.head_sequence
+    )
+    assert "frozen worker prompt" in str(worker_path[0].content)
+    assert "frozen validator prompt" in str(validator_path[0].content)
+
+
+@pytest.mark.asyncio
+async def test_repair_validates_identity_without_resetting_system_prompt():
+    expected = _override("original frozen prompt")
+    trace = SimpleNamespace(
+        context={"role_prompt_identity": expected.prompt_identity},
+        total_tokens=0,
+        total_cost=0.0,
+        model="worker-model",
+    )
+
+    class Coordinator:
+        async def validate_continue_from(self, *args):
+            return "worker-trace"
+
+    class Store:
+        async def get_trace(self, trace_id):
+            return trace
+
+        async def update_trace(self, trace_id, **updates):
+            trace.context = updates["context"]
+
+    class Runner(CapturingRunner):
+        def __init__(self):
+            super().__init__()
+            self.trace_store = Store()
+            self.task_coordinator = Coordinator()
+
+    runner = Runner()
+    result = await LocalAgentExecutor(
+        runner, role_system_prompt_resolver=PromptResolver(expected)
+    ).run_worker(
+        _worker_context(
+            attempt_id="attempt-2",
+            prior_attempt_id="attempt-1",
+            continue_trace_id="worker-trace",
+        )
+    )
+
+    assert result.status == "completed"
+    assert runner.configs[0].system_prompt is None
+    assert trace.context["role_prompt_identity"] == expected.prompt_identity
+    assert trace.context["attempt_id"] == "attempt-2"
+
+
+@pytest.mark.asyncio
+async def test_repair_prompt_mismatch_fails_before_trace_context_mutation():
+    original = _override("original")
+    trace = SimpleNamespace(
+        context={"role_prompt_identity": original.prompt_identity, "attempt_id": "old"},
+        total_tokens=0,
+        total_cost=0.0,
+        model="worker-model",
+    )
+
+    class Coordinator:
+        async def validate_continue_from(self, *args):
+            return "worker-trace"
+
+    class Store:
+        updated = False
+
+        async def get_trace(self, trace_id):
+            return trace
+
+        async def update_trace(self, trace_id, **updates):
+            self.updated = True
+
+    runner = CapturingRunner()
+    runner.trace_store = Store()
+    runner.task_coordinator = Coordinator()
+    result = await LocalAgentExecutor(
+        runner, role_system_prompt_resolver=PromptResolver(_override("changed"))
+    ).run_worker(
+        _worker_context(
+            attempt_id="new",
+            prior_attempt_id="old",
+            continue_trace_id="worker-trace",
+        )
+    )
+
+    assert result.status == "failed"
+    assert "ROLE_SYSTEM_PROMPT_MISMATCH" in result.error
+    assert runner.trace_store.updated is False
+    assert runner.configs == []
+
+
+@pytest.mark.asyncio
+async def test_wire_injects_optional_system_prompt_resolver(tmp_path):
+    runner = CapturingRunner()
+    resolver = PromptResolver(_override("wired prompt"))
+    coordinator = wire_orchestration(
+        runner,
+        FileSystemTaskStore(str(tmp_path)),
+        FileSystemArtifactStore(str(tmp_path)),
+        role_system_prompt_resolver=resolver,
+    )
+
+    result = await coordinator.executor.run_worker(_worker_context())
+
+    assert result.status == "completed"
+    assert runner.configs[0].system_prompt == "wired prompt"

agent/tests/test_trace_attachments.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+import base64
+import inspect
+from hashlib import sha256
+
+import pytest
+
+from agent import AttachmentRef, TraceAttachmentStore
+from agent.core.runner import AgentRunner, RunConfig
+from agent.trace.models import Message, Trace
+from agent.trace.store import FileSystemTraceStore
+
+
+@pytest.mark.asyncio
+async def test_filesystem_trace_attachment_roundtrip_and_digest_verification(tmp_path):
+    store = FileSystemTraceStore(str(tmp_path))
+    await store.create_trace(Trace(trace_id="trace", mode="agent"))
+    await store.add_message(
+        Message.create(
+            trace_id="trace",
+            role="tool",
+            sequence=1,
+            content="attachment source",
+        )
+    )
+    content = b"\x89PNG\r\ncontrolled-image"
+    digest = f"sha256:{sha256(content).hexdigest()}"
+
+    ref = await store.store_message_attachment(
+        trace_id="trace",
+        message_id="trace-0001",
+        media_type="image/png",
+        content=content,
+        sha256=digest,
+    )
+
+    assert isinstance(store, TraceAttachmentStore)
+    assert isinstance(ref, AttachmentRef)
+    assert ref.size_bytes == len(content)
+    assert await store.read_message_attachment(ref) == content
+    reloaded = FileSystemTraceStore(str(tmp_path))
+    assert await reloaded.read_message_attachment(ref) == content
+    assert (
+        await store.store_message_attachment(
+            trace_id="trace",
+            message_id="trace-0001",
+            media_type="image/png",
+            content=content,
+            sha256=digest,
+        )
+        == ref
+    )
+
+    with pytest.raises(ValueError, match="digest mismatch"):
+        await store.store_message_attachment(
+            trace_id="trace",
+            message_id="message-2",
+            media_type="image/png",
+            content=content,
+            sha256=f"sha256:{'0' * 64}",
+        )
+
+    forged_media_type = AttachmentRef(
+        trace_id=ref.trace_id,
+        message_id=ref.message_id,
+        media_type="image/jpeg",
+        sha256=ref.sha256,
+        size_bytes=ref.size_bytes,
+    )
+    with pytest.raises(ValueError, match="does not match metadata"):
+        await reloaded.read_message_attachment(forged_media_type)
+
+    payload_path = (
+        tmp_path
+        / ref.trace_id
+        / "attachments"
+        / ref.message_id
+        / f"{ref.sha256.removeprefix('sha256:')}.bin"
+    )
+    payload_path.write_bytes(b"X" * len(content))
+    with pytest.raises(ValueError, match="integrity check failed"):
+        await reloaded.read_message_attachment(ref)
+
+
+@pytest.mark.asyncio
+async def test_runner_uses_optional_attachment_port_and_old_store_stays_compatible(
+    tmp_path,
+):
+    content = b"image bytes"
+    image = {
+        "type": "base64",
+        "media_type": "image/png",
+        "data": base64.b64encode(content).decode("ascii"),
+    }
+    store = FileSystemTraceStore(str(tmp_path))
+    await store.create_trace(Trace(trace_id="trace", mode="agent"))
+    await store.add_message(
+        Message.create(
+            trace_id="trace",
+            role="tool",
+            sequence=1,
+            content="attachment source",
+        )
+    )
+    runner = AgentRunner(trace_store=store)
+
+    await runner._persist_tool_attachments("trace", "trace-0001", [image])
+
+    ref = AttachmentRef(
+        trace_id="trace",
+        message_id="trace-0001",
+        media_type="image/png",
+        sha256=f"sha256:{sha256(content).hexdigest()}",
+        size_bytes=len(content),
+    )
+    assert await store.read_message_attachment(ref) == content
+
+    class LegacyStore:
+        pass
+
+    legacy_runner = AgentRunner(trace_store=LegacyStore())
+    await legacy_runner._persist_tool_attachments("trace", "message", [image])
+
+    with pytest.raises(ValueError, match="invalid base64"):
+        await runner._persist_tool_attachments(
+            "trace",
+            "bad-message",
+            [{"type": "base64", "media_type": "image/png", "data": "%%%"}],
+        )
+
+
+@pytest.mark.asyncio
+async def test_non_multimodal_run_works_with_trace_store_without_attachment_port(
+    tmp_path,
+):
+    inner = FileSystemTraceStore(str(tmp_path))
+
+    class LegacyTraceStore:
+        def __getattr__(self, name):
+            if name in {"store_message_attachment", "read_message_attachment"}:
+                raise AttributeError(name)
+            return getattr(inner, name)
+
+    async def llm_call(**_kwargs):
+        return {"content": "done", "tool_calls": None, "finish_reason": "stop"}
+
+    legacy = LegacyTraceStore()
+    assert not isinstance(legacy, TraceAttachmentStore)
+    runner = AgentRunner(trace_store=legacy, llm_call=llm_call)
+
+    result = await runner.run_result(
+        [{"role": "user", "content": "plain text task"}],
+        RunConfig(name="legacy text run", max_iterations=1),
+    )
+
+    assert result["status"] == "completed"
+    assert "_get_messages_dir" not in inspect.getsource(AgentRunner)
+
+
+def test_attachment_ref_rejects_unsafe_or_invalid_fields():
+    digest = f"sha256:{'0' * 64}"
+    with pytest.raises(ValueError, match="media_type"):
+        AttachmentRef("trace", "message", "", digest, 1)
+    with pytest.raises(ValueError, match="size_bytes"):
+        AttachmentRef("trace", "message", "image/png", digest, -1)