第 21 章：实现 Installed Agent 与标准轨迹
一次故障诊断 Trial 得到了满分，agent/trajectory.json 也能被 Viewer 打开。团队据此把轨迹送入训练流水线，抽检时才发现三个问题：工具输出挂在错误的调用上，压缩前的对话被重复计入样本，某条 observation 里还保留了认证 Header。格式“像 ATIF”并不等于轨迹真实、可训练或可安全分享。
本章把第 20 章的外部 Agent 改造成 Installed Agent：Agent 的执行程序安装在任务容器中，以 Headless（无人值守）方式运行；Harbor 侧适配器把原生日志转换为 ATIF（Agent Trajectory Interchange Format）轨迹。示例是确定性的契约夹具，不调用模型，也不声称跑过 Docker。它用于验证安装接口、权限、日志同步、转换、脱敏和 ATIF 校验；接入真实 Agent 时，把夹具的原生事件生产器替换为实际 CLI 即可。
读完本章，你应该能够：
- 实现
BaseInstalledAgent的安装、Headless 执行和宿主侧日志回填； - 正确记录 user、agent、tool call、observation 和指标之间的关系；
- 区分 ATIF v1.7 规范与 Harbor 的
raw_content、linear_history生产者选项； - 表达上下文压缩后的真实线性历史，避免重复训练 copied context；
- 在 Viewer、SFT 和 RL 之前执行格式、脱敏和消费端兼容性验收。
21.1 Installed 不是“换一个运行位置”
Harbor 把自定义 Agent 分为两种集成方式：External Agent 通过 BaseEnvironment 从 Harbor 进程控制环境；Installed Agent 则把程序装进任务环境，并在那里以 Headless 模式执行。两者都实现 BaseAgent 的 setup()、run() 和结果上下文契约；BaseInstalledAgent 额外抽象了 install()、root/Agent 用户命令、Prompt 模板、CLI 参数、环境变量和非零退出分类。12
BaseInstalledAgent.setup() 会先以 root 确保 /installed-agent 存在，再调用子类的 install()；若没有显式版本且子类提供版本命令，版本探测只是 best effort，异常会被忽略。exec_as_root() 显式使用 root，exec_as_agent() 不指定用户，因而落到 Environment 当前的 default_user。两者最终都经 _exec() 加上 set -o pipefail，非零退出会抛出异常，而不是被末尾的 tee 掩盖。3
用户身份不是适配器自己猜的。Task 的 [agent].user 可以是用户名、UID 或空值；Trial 在 setup 和 run 周围临时设置 environment.default_user。若为空，才使用容器默认用户。Agent 的 env 也由 Trial 以 scoped overlay 只包围 setup/run，优先级高于命令级 env，不会自然延伸到 Verifier、构建或 Artifact 阶段。456
这带来一条实用的权限规则：系统包、共享 venv 和可执行文件由 exec_as_root() 准备；配置、缓存和 Agent CLI 由 exec_as_agent() 创建。不要在 run() 中临时 chmod -R 777 /workspace，也不要默认 Agent 必须是 root。需要写日志时，固定路径是容器内 /logs/agent；Trial 的输出目录把它映射或同步到宿主侧的 trial/agent/。7
Installed Agent 的最小数据流如下：
宿主 Harbor Task Environment
─────────── ────────────────
setup() ── upload/install ─────────> /installed-agent/
run() ── Headless command ───────> Agent CLI
│
└─ /logs/agent/native-events.json
<──── 同步日志 ───┘
populate_context_post_run()
├─ 解析原生事件
├─ 脱敏并映射 ATIF
├─ Pydantic/Validator 校验
└─ 写 trial/agent/trajectory.json
Trial 会先下载 Agent 日志，再调用 populate_context_post_run()；但只有 AgentContext 仍为空时才调用。由此可以推断：若转换依赖同步回宿主的原生日志，run() 不应先写入 context.metadata 或 token 统计，否则 post-run 回填会被跳过。转换成功后再填上下文。8
注意：
BaseInstalledAgent没有一个通用的headless=True开关。非交互入口、接受确认、标准输入、退出状态和超时都是具体 CLI 适配器的责任。自动化命令至少应使用上游的非交互模式、关闭 stdin，并让非零退出传播。盲目追加--yes并不能让一个交互式 CLI 变得可靠。
21.2 ATIF 是数据契约，不是运行器
本书基线 Harbor v0.18.0 随附的 RFC 当前版本是 ATIF v1.7。根对象必需 schema_version、agent 和非空 steps；session_id 在 v1.7 已是可选的 run 级标识，trajectory_id 才是单个轨迹文档标识。Harbor 的 Pydantic 模型接受 ATIF-v1.0 到 ATIF-v1.7，默认写 v1.7。910
一条 Step 至少有从 1 开始连续递增的 step_id、source 和 message。source 只能是 system、user 或 agent；模型名、reasoning、tool call 和 metrics 只能放在 agent step。工具调用要求 tool_call_id、function_name 和对象类型的 arguments，observation 的 source_call_id 必须引用同一步已有的调用 ID。1112
根级 agent.name 与 agent.version 都是必需字段，model_name、tool_definitions 和 extra 可选。工具定义采用 OpenAI function-calling schema；它描述 Agent 当时可见的工具集合，不是某次实际调用。若 Agent 动态增删工具，不能只填最终集合而不解释变化；可以按真实能力拆段，或在 step 的 extra 中记录不会破坏标准字段语义的版本标识。Harbor 模型把未知字段设为 forbidden，自定义数据应进入各层显式的 extra，不要在根或 step 随意发明并列字段。910
时间戳虽然可选，但一旦填写就要是可解析的 ISO 8601 字符串。它适合排序和定位等待，不适合直接推导模型 latency：一个 agent step 可能包含模型推理、工具运行和重试，而生产者记录的时间点也可能是开始、结束或事件落盘。要比较延迟，应在 extra 中定义清楚单调时钟区间或在外部 Trial timing 中取数，不能只拿相邻 timestamp 相减后命名为“模型耗时”。
ATIF v1.7 还补上了一个容易丢失的事实：一个 step 到底代表多少次模型推理。能取得逐调用边界时，生产者应该每次 LLM inference 写一个 step；无法拆分时，用 llm_call_count 写实际调用数。确定性的调度或规则引擎应写 llm_call_count=0，且不得附带 metrics 或 reasoning_content；这类 step 必须从 SFT 中排除。13
下面的映射表比“把聊天记录转成 JSON”更有用：
| 原生事件 | ATIF 位置 | 必须保留的关系 |
|---|---|---|
| system/task instruction | `Step(source="system" | "user")` |
| 一次模型返回 | Step(source="agent") | 一次调用一个 step；聚合时写 llm_call_count |
| 工具请求 | 同一 agent step 的 tool_calls[] | 调用 ID 唯一，参数保持 JSON 对象 |
| 工具结果 | 同一 step 的 observation.results[] | source_call_id 精确指向调用 ID |
| token、logprob、成本 | metrics / final_metrics | 未观测到就省略，不能用 0 冒充已测 |
| 压缩、裁剪、注入 | system step 的 extra.context_management | 写清边界如何改变后续有效上下文 |
TrajectoryValidator 会用 Pydantic 校验字段、类型和模型级约束；对文件路径输入，它还会检查本地图片引用是否存在。它不检查“这段文本是否真送给模型”、token 数是否来自账单，也不扫描 Key、Token 或 Password。因此，格式校验是必要条件，不是语义审计或脱敏。14
还有一处必须写进发布 lint 的规范/实现差异：RFC 把 schema_version 列为 Required，而 v0.18.0 的 Trajectory 模型为它提供默认值 ATIF-v1.7；因此该版本 Validator 会接受原始 JSON 中缺少 schema_version 的文档。生产者应始终显式写出版本，发布检查还要在调用 Validator 前断言原始对象含此键，不能把默认填充误当作规范兼容。910
21.2.1 Harbor 选项与 ATIF 字段不能混为一谈
raw_content 和 linear_history 定义在 Harbor 的 TrajectoryConfig TypedDict 中，并非 ATIF 根字段。v0.18.0 里只有部分生产者消费这些选项，尤其是 Terminus-2 与 OpenHands；继承 BaseInstalledAgent 不会自动获得相同行为。15
| 概念 | Harbor v0.18.0 约定 | ATIF v1.7 表达 |
|---|---|---|
raw_content=True | Terminus-2 把原始 LLM response 放入 message，并跳过其解析出来的 tool_calls | ATIF 只规定 message 和 tool_calls 的结构，不定义这个布尔开关 |
linear_history=True | Terminus-2 在 summarization 时拆分文件 | 前段用 continued_trajectory_ref 指向后段；复制进新段的 step 标 is_copied_context |
| 上下文压缩 | Agent 自己决定何时摘要、如何恢复 | system step 可在 extra.context_management 写 type 和 boundary |
对调试，结构化 tool_calls 通常比完整原始 response 更易检索；对 SFT，模型实际生成内容通常更重要。二者不是“详细/简略”开关，而是两种记录目标。若上游同时提供原始 response 和结构化事件，稳妥做法是 ATIF 保留训练所需内容，把经过脱敏的原生 payload 放在单独受控 Artifact，或在 extra 中只存不会泄密且消费端明确支持的引用；不要复制一份含认证 Header 的完整响应。
21.2.2 压缩后只有真实历史才叫线性历史
ATIF v1.7 的 context-management 约定允许 system step 在 extra.context_management 中声明 compaction、pruning 或 injection，以及 replace、append 或 truncate 边界。对于 boundary="replace"，后续有效上下文由该边界 step 的 observation 内容和边界后的新 step 组成；边界前内容仍保留用于审计，却不再属于模型输入。16
另一种做法是拆分轨迹：trajectory.json 通过 continued_trajectory_ref 指向 trajectory.cont-1.json。如果新段复制了旧 step 作为上下文，规范要求将它们标为 is_copied_context=True；SFT 消费端必须过滤这些 step，避免同一次交互重复成为训练目标。Harbor 的 Terminus-2 正是在线性模式下先保存当前段，再把恢复后的上下文标记为 copied context。17
选择标准很直接：若一个文件仍能无歧义地重建每轮真实输入，用 context boundary 即可；若摘要后模型上下文被重置，拆成 continuation 更容易审计。不要为了“看起来连续”把压缩前后的 agent 消息拼成一条从未送给模型的对话。
21.3 最小 Installed Agent 契约夹具
示例目录放在配套项目根目录，模块路径为 agents.diag_installed:DiagInstalledAgent：
agents/
├── __init__.py
├── diag_installed.py
└── diag_runner.py
先实现容器内 runner。它只依赖 Python 标准库，写一份教学性诊断文件和一份已做第一层脱敏的原生日志。它没有调用 LLM，所以两个 agent event 都明确写 llm_call_count=0；不能把这份夹具的结果当作模型实测。
# agents/diag_runner.py
import argparse
import base64
import json
import re
from pathlib import Path
def scrub_text(text: str) -> str:
text = re.sub(r"(?i)Bearer\s+[^\s]+", "Bearer [REDACTED]", text)
return re.sub(r"\bsk-[A-Za-z0-9_-]+", "[REDACTED]", text)
parser = argparse.ArgumentParser()
parser.add_argument("--instruction-b64", required=True)
parser.add_argument("--session-id", required=True)
args = parser.parse_args()
instruction = scrub_text(base64.b64decode(args.instruction_b64).decode("utf-8"))
diagnosis = {
"status": "diagnosed",
"service": "api",
"evidence": ["deterministic-contract-fixture"],
}
workspace = Path("/workspace")
workspace.mkdir(parents=True, exist_ok=True)
(workspace / "diagnosis.json").write_text(
json.dumps(diagnosis, ensure_ascii=False, indent=2), encoding="utf-8"
)
events = {
"session_id": args.session_id,
"events": [
{"role": "user", "message": instruction},
{
"role": "agent",
"message": "",
"llm_call_count": 0,
"tool_calls": [{
"id": "call-write-1",
"name": "write_diagnosis",
"arguments": {"path": "/workspace/diagnosis.json"},
}],
"observations": [{
"call_id": "call-write-1",
"content": "diagnosis.json created",
}],
},
{
"role": "agent",
"message": "诊断记录已写入 /workspace/diagnosis.json。",
"llm_call_count": 0,
},
],
}
logs = Path("/logs/agent")
logs.mkdir(parents=True, exist_ok=True)
(logs / "native-events.json").write_text(
json.dumps(events, ensure_ascii=False, indent=2), encoding="utf-8"
)
接着实现宿主侧适配器。run() 只启动容器程序，不提前填充 AgentContext；populate_context_post_run() 等日志同步后才转换。base64 只用于安全传递 shell 参数，不是加密，任务 instruction 不应承载认证秘密。
# agents/diag_installed.py
import base64
import json
import re
import shlex
from pathlib import Path
from typing import Any, override
from harbor.agents.installed.base import BaseInstalledAgent, with_prompt_template
from harbor.environments.base import BaseEnvironment
from harbor.models.agent.context import AgentContext
from harbor.models.trajectories import (
Agent as AtifAgent,
FinalMetrics,
Observation,
ObservationResult,
Step,
ToolCall,
Trajectory,
)
from harbor.utils.trajectory_utils import format_trajectory_json
from harbor.utils.trajectory_validator import TrajectoryValidator
_SENSITIVE_PARTS = {"KEY", "SECRET", "TOKEN", "PASSWORD", "CREDENTIAL", "AUTH"}
def scrub(value: Any) -> Any:
if isinstance(value, dict):
clean = {}
for key, item in value.items():
parts = set(filter(None, re.split(r"[^A-Za-z0-9]+", key.upper())))
clean[key] = "[REDACTED]" if parts & _SENSITIVE_PARTS else scrub(item)
return clean
if isinstance(value, list):
return [scrub(item) for item in value]
if isinstance(value, str):
value = re.sub(r"(?i)Bearer\s+[^\s]+", "Bearer [REDACTED]", value)
return re.sub(r"\bsk-[A-Za-z0-9_-]+", "[REDACTED]", value)
return value
class DiagInstalledAgent(BaseInstalledAgent):
SUPPORTS_ATIF = True
@staticmethod
@override
def name() -> str:
return "diag-installed"
@override
async def install(self, environment: BaseEnvironment) -> None:
runner = Path(__file__).with_name("diag_runner.py")
await environment.upload_file(runner, "/installed-agent/diag_runner.py")
await self.exec_as_root(
environment, "chmod 0755 /installed-agent/diag_runner.py"
)
@with_prompt_template
async def run(
self,
instruction: str,
environment: BaseEnvironment,
context: AgentContext,
) -> None:
del context # post-run 在日志同步后回填
encoded = base64.b64encode(instruction.encode("utf-8")).decode("ascii")
session_id = self.session_id or "missing-session-id"
command = (
"python3 /installed-agent/diag_runner.py "
f"--instruction-b64 {shlex.quote(encoded)} "
f"--session-id {shlex.quote(session_id)} "
"</dev/null 2>&1 | tee /logs/agent/diag-installed.txt"
)
await self.exec_as_agent(environment, command=command)
def _convert(self, raw: dict[str, Any]) -> Trajectory:
raw = scrub(raw)
steps: list[Step] = []
for event in raw["events"]:
role = event["role"]
if role == "user":
steps.append(Step(
step_id=len(steps) + 1,
source="user",
message=event.get("message", ""),
))
continue
if role != "agent":
raise ValueError(f"unsupported native role: {role}")
calls = [
ToolCall(
tool_call_id=item["id"],
function_name=item["name"],
arguments=item.get("arguments", {}),
)
for item in event.get("tool_calls", [])
]
results = [
ObservationResult(
source_call_id=item.get("call_id"),
content=item.get("content"),
)
for item in event.get("observations", [])
]
steps.append(Step(
step_id=len(steps) + 1,
source="agent",
message=event.get("message", ""),
tool_calls=calls or None,
observation=Observation(results=results) if results else None,
llm_call_count=event.get("llm_call_count"),
))
return Trajectory(
schema_version="ATIF-v1.7",
session_id=raw.get("session_id"),
agent=AtifAgent(
name=self.name(),
version=self.version() or "unknown",
model_name=self.model_name,
),
steps=steps,
final_metrics=FinalMetrics(total_steps=len(steps)),
notes="Deterministic contract fixture; no LLM was called.",
)
@override
def populate_context_post_run(self, context: AgentContext) -> None:
native_path = self.logs_dir / "native-events.json"
if not native_path.exists():
self.logger.error("native event log is missing")
return
try:
raw = json.loads(native_path.read_text(encoding="utf-8"))
trajectory = self._convert(raw)
payload = trajectory.to_json_dict()
validator = TrajectoryValidator()
if not validator.validate(payload, validate_images=False):
raise ValueError("; ".join(validator.get_errors()))
target = self.logs_dir / "trajectory.json"
temporary = self.logs_dir / "trajectory.json.tmp"
temporary.write_text(
format_trajectory_json(payload), encoding="utf-8"
)
temporary.replace(target)
context.metadata = {"trajectory_schema_version": trajectory.schema_version}
native_path.unlink() # 成功转换后不保留原生日志副本
except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
self.logger.error("trajectory conversion failed: %s", exc)
真实模型适配器应在原生事件里记录实际的模型请求边界和 usage，再在 _convert() 中构造 Metrics。没有从 Provider 取得 token ID、logprob 或实际成本时就省略字段；不要根据字符数估 token，也不要把“未知”写成 0。若上游一个事件聚合了两次模型调用，写 llm_call_count=2，而不是伪装成单次推理。
21.3.1 用真实事件替换夹具
把规则夹具换成真实 CLI 时，先固定“原生日志契约”，再写转换器。至少为每次模型调用保存稳定的 event ID、请求序号、实际模型名、response 文本、结构化 tool call、usage 以及调用结束状态；对每次工具执行保存 call ID、退出状态和交还模型的 observation。上游只给累计 token 时，转换器要用相邻累计值求每步增量，并为重启、重试和计数回退写测试；不能把同一个累计值复制到每个 step 后再求和。
原始 response 与 ATIF reasoning_content 也要分开。reasoning_content 表示 Agent 显式提供的推理文本，不是让适配器推测模型“心里想了什么”。Provider 只返回加密 thinking、摘要或空字段时，应保留可公开的实际字段并省略 reasoning；为了填满 schema 而让另一个模型补写推理，会把合成内容伪装成运行证据。
一个可靠 recorder 应维护四条不变量：
- 每个有模型调用的 agent step 都能追溯到一个或明确数量的原生请求；
- 每个
source_call_id都能在同一步找到唯一 tool call，未关联输出不伪造 ID； final_metrics只聚合本轨迹有证据的指标，子 Agent 和 continuation 是否计入必须写进 notes；- 轨迹只在完整构造、脱敏和验证后原子替换，异常时保留原生日志供本地诊断，不留下半个 JSON。
如果原生 CLI 在运行中持续写 JSONL，可在 Agent timeout 后读取到已完成事件。转换器应忽略最后一条截断行并在 notes 记录“部分轨迹”，而不是因为尾部损坏丢弃整个历史。相反，普通 JSON 文件只有整体解析成功才可信；对它吞掉 JSONDecodeError 并生成空成功轨迹，会把 Agent 崩溃误包装为一次合法运行。
21.4 确定性验证：先证明契约，再运行容器
从项目根目录先做不需要 Docker 和模型的语义探针。它直接向宿主日志目录写入一份原生夹具，调用 post-run 转换，并检查脱敏、工具关联和 llm_call_count=0 约束：
uv run --frozen python - <<'PY'
import json
import tempfile
from pathlib import Path
from agents.diag_installed import DiagInstalledAgent
from harbor.models.agent.context import AgentContext
from harbor.utils.trajectory_validator import TrajectoryValidator
with tempfile.TemporaryDirectory() as tmp:
logs = Path(tmp)
native = {
"session_id": "contract-1",
"events": [
{"role": "user", "message": "Authorization: Bearer do-not-store"},
{
"role": "agent",
"message": "",
"llm_call_count": 0,
"tool_calls": [{
"id": "call-1", "name": "inspect", "arguments": {"path": "/etc/hosts"}
}],
"observations": [{"call_id": "call-1", "content": "ok"}],
},
],
}
(logs / "native-events.json").write_text(json.dumps(native), encoding="utf-8")
agent = DiagInstalledAgent(logs_dir=logs, version="0.1.0")
context = AgentContext()
agent.populate_context_post_run(context)
trajectory_path = logs / "trajectory.json"
payload = json.loads(trajectory_path.read_text())
assert TrajectoryValidator().validate(trajectory_path)
assert "do-not-store" not in trajectory_path.read_text()
assert payload["steps"][1]["llm_call_count"] == 0
assert payload["steps"][1]["observation"]["results"][0]["source_call_id"] == "call-1"
assert not (logs / "native-events.json").exists()
assert context.metadata == {"trajectory_schema_version": "ATIF-v1.7"}
print("installed-agent trajectory contract: ok")
PY
再做一个故障反例：把 observation 的 call_id 改成 missing-call。Trajectory 构造应拒绝它，因为该 ID 不在同一步的 tool_calls 中。另一个反例是在 llm_call_count=0 的 step 加上 metrics；v0.18.0 的 Step 模型也应拒绝。失败才说明测试覆盖了关键约束，不能只验证 happy path。
21.4.1 三层验收，不用一次昂贵运行代替测试
第一层是纯转换测试：以脱敏后的合成原生事件覆盖单轮、并行工具、未知 role、截断文件、累计 usage 回退和压缩边界。断言完整 payload，而不只断言 validate() is True。这一层应在每次提交运行，既不需要容器，也不需要凭据。
第二层是 mock Environment 契约测试：用 AsyncMock 调用 setup() 与 run()，验证 runner 上传到 /installed-agent、系统操作显式用 root、Headless 命令含关闭 stdin、执行用户没有被硬编码为 root、命令不含合成 Key。还要模拟返回码非零，确认 _exec() 抛错；否则 shell pipeline 很容易把 Agent 失败当成 tee 成功。
第三层才是发布门控的 Docker/Provider 冒烟：以非 root Task user 启动一个无付费模型的契约任务，验证 /workspace 写权限、日志同步顺序、timeout 恢复和清理。真实模型 Trial 属于另一条付费门控，应使用短期凭据并记录 Agent、模型、镜像和 Task 版本。三层的失败归因不同，不能用一次真实 Job 同时证明转换器、权限和模型质量。
静态契约通过后，可在具备 Docker 的环境中运行一个 Task：
harbor run \
-d datasets/system-config-benchmark \
--agent agents.diag_installed:DiagInstalledAgent \
--ak version=0.1.0 \
--agent-include-logs 'native-events.json' \
--agent-include-logs 'diag-installed.txt' \
--yes
这条命令的 import-path 形式与 v0.18.0 CLI 和 Factory 一致。18 但日志过滤有一个顺序陷阱：非挂载 Provider 会先按 include/exclude 下载 /logs/agent，然后才调用 post-run 转换。如果配置 include，必须把 native-events.json 包含进来；只包含尚未生成的 trajectory.json 会让转换器看不到输入。默认不设置过滤则下载整个 Agent 日志目录。19
本章没有执行上面的 Docker 命令。实际验收至少检查：agent/trajectory.json 存在并通过 validator；result.json 的 Agent 名称与版本正确；容器运行日志没有等待输入；原生临时日志在成功转换后删除；故障 Task 的 Verifier 只依据环境结果评分，不依据轨迹自述。
21.5 脱敏不是一个正则表达式
示例的 scrub() 只是最低限度的防回归夹具：按字段名遮蔽常见 secret，并处理 Bearer 与示例 Key 形态。真实系统还可能在 URL query、Cookie、堆栈、Git remote、文件内容、截图、base64、工具参数和模型复述中泄密。应按数据源建立 allowlist，默认不保留未知 Header；对必须保留的值使用不可逆替换，并记录脱敏规则版本。
Harbor 的 AgentConfig.env 序列化会把名称含 KEY、SECRET、TOKEN、PASSWORD、CREDENTIAL 或 AUTH 的宿主匹配值转成 ${VAR}，其他敏感字面量只保留部分掩码。这个机制保护的是持久化配置，不是 trajectory；运行时 Agent 仍能读取完整值。TrajectoryValidator 也不会调用这套 env 脱敏逻辑。20
可以按数据生命周期安排三道互不替代的门：
| 时点 | 目标 | 典型措施 |
|---|---|---|
| 写原生日志前 | 减少秘密首次落盘 | Header allowlist、禁止记录进程环境、query 参数清洗、截图区域遮挡 |
| 原生格式转 ATIF 时 | 统一处理嵌套内容 | 按字段名与值模式递归脱敏、删除无训练价值 payload、校验替换计数 |
| 上传、Viewer 或训练前 | 防止规则遗漏扩散 | Canary 扫描、人工抽检、访问控制、不可变审计清单和发布阻断 |
第一道门最重要，因为删除宿主副本不能收回 Provider 控制面、集中日志或备份中的内容。第二道门要同时扫 message、reasoning_content、tool_calls.arguments、observation、extra、notes 和多模态文件路径；只扫对话文本会漏掉最常见的命令输出。第三道门应 fail closed：扫描器异常或规则版本未知时停止发布，而不是打印 warning 后继续上传。
脱敏还可能破坏可训练语义。例如把所有长十六进制串都删掉，会误删 commit ID、校验和和错误追踪 ID。更好的策略是先识别来源字段，再做类型化替换，例如 <API_TOKEN_1>、<COOKIE_1>；同一轨迹内保持同一占位符，便于看出数据流，但不同 Trial 使用不同映射盐，避免跨样本关联。映射表若必须保留，应与书稿、Job 和训练数据分开加密存储，并设置更短保留期。
警告：不要把 API Key 放进 instruction、CLI 参数、Prompt、
tool_calls.arguments或reasoning_content。Agent 需要认证时，用AgentConfig.env的${HOST_VAR}模板和最小权限、短期凭据；仍须审计原生 CLI 是否把环境或请求 Header写入日志。base64、日志 exclude 和成功后删除都不是擦除已经上传到外部日志系统的秘密。
脱敏后的轨迹还应保留可追踪性。建议生成一份不含内容的审计清单：规则版本、被替换字段计数、被删除文件名、trajectory 的 SHA-256、审核人和时间。不要记录被遮蔽值的前后缀；它们常能辅助撞库或跨数据集关联。
21.6 Viewer、SFT 与 RL：合法 JSON 只是第一道门
Viewer 后端的轨迹接口读取 trial/agent/trajectory.json 并 json.loads() 返回；它不调用 TrajectoryValidator。因此“Viewer 能打开”只能证明文件存在且 JSON 可解析，不能证明 ATIF 有效。发布前仍应单独运行 validator。21
SFT 的边界更微妙。v0.18.0 的 extract_conversations_from_trajectory() 会过滤 is_copied_context=True 的 agent step，并沿 continued_trajectory_ref 收集主轨迹后续段；但高层 export_traces() 会先把结果中的 Agent 名称转成内置 AgentName，再从 Factory 检查 SUPPORTS_ATIF。任意 import-path 自定义名称（如 diag-installed）会在这里被拒绝，即使文件本身是合法 ATIF。2223
更准确地说，提取器不会把 copied agent step 当成新的 episode 终点，但在构造后续 episode 的输入 conversation 时仍可能包含 copied step。是否保留它作为上下文、是否把它计算进 loss，最终取决于训练数据结构和 collator。不能仅看到 exporter “过滤”代码就断言重复训练已经被消除；应检查导出行和 loss mask，而不只是行数。
此外，v0.18.0 的 conversation 提取器按所有未 copied 的 agent step 建立 episode，没有检查 llm_call_count=0；这与 ATIF v1.7 要求 SFT 排除确定性调度 step 的规范存在缺口。由此可以推断，本章夹具不能直接送进 v0.18.0 的高层 SFT 导出器。安全做法是：向 Harbor 上游注册并修复消费端门槛，或者在自有 exporter 中先验证 ATIF、排除 llm_call_count=0 和 copied context，再构造训练样本；不要把自定义 Agent 冒充某个内置名称。24
RL 也不是“有 trajectory 就能训练”。ATIF 的 prompt_token_ids、completion_token_ids 和 logprobs 都是可选字段；Harbor 的 AgentContext.rollout_details 另有按轮保存 token ID 与 logprob 的结构。这里还有一个 v0.18.0 类型差异：AgentContext 的字段描述提到 loss mask，但 RolloutDetail TypedDict 实际只声明 prompt/completion token IDs、logprobs 和 extra，没有独立 loss_masks 字段。训练 Adapter 必须明确从对话角色推导 mask，或在双方约定的 extra 中承载并自行校验，不能假设 Harbor 已提供标准 loss mask。若 Provider 没有返回 token 数据，重新分词还会改变动作边界，合法 ATIF 仍不足以做无损 rollout。2526
因此，把兼容性写成门槛表，而不是一个布尔值：
| 消费端 | 本章夹具 | 生产接入门槛 |
|---|---|---|
| Viewer | JSON 可读取；仍需独立校验 | trajectory.json、有效 ATIF、内容已脱敏 |
| 审计/调试 | 可用 | tool/result 关联正确，压缩边界真实，原生日志受控 |
| Harbor v0.18.0 SFT CLI | 不直接兼容自定义名称 | 内置 Factory 门槛通过；另补 llm_call_count=0 过滤 |
| 自有 SFT exporter | 可作为负向夹具 | raw response 目标明确；过滤 copied 与零调用 step；人工抽检 |
| RL rollout | 不可用 | 实际 token IDs、loss mask/logprob 语义、Reward 与训练 Adapter 均已核验 |
上线前为每个消费端保存一份黄金夹具：同一轨迹在新版本 Viewer、exporter 和 RL Adapter 中得到预期 step 数、episode 数、工具关联和 loss mask。升级 Harbor 时比较结构化结果，不比较截图。只要 consumer 改变了 copied context、零调用 step、multimodal content 或 continuation 的处理方式，就应把它视为训练数据版本变化，而不是无害的 UI 升级。
21.7 失败模式与排查顺序
trajectory.json 缺失。 先看 diag-installed.txt 是否有非零退出，再检查 native-events.json 是否被日志过滤排除，最后确认 run() 是否提前填了 AgentContext 导致 post-run 跳过。不要先怀疑 Validator，因为它尚未取得输入。
工具引用校验失败。 对每个 observation 追溯同一步 tool_calls[].tool_call_id。并行调用时不能把所有输出都挂到第一个 ID；上游没有提供关联信息，就省略 source_call_id 并在 notes 解释，不能编造映射。
轨迹通过但训练重复。 查 summarization 前后是否拆段、continuation 是否形成无环链、复制 step 是否标 is_copied_context=True。再检查消费端是否真的执行过滤，而不是只相信生产端标志。
日志中仍有秘密。 暂停上传与训练，撤销相关凭据；从原生事件、ATIF、命令 stdout/stderr、Job 配置和外部日志逐层定位。修复后用已吊销的合成 Canary 值做回归，不要拿真实 Key 测正则。
合法 ATIF 却无法 harbor traces export。 查看 result.json 中的 Agent 名称，并核对 v0.18.0 的 AgentName/Factory 门槛。对 import-path Agent，这是已知版本边界，不应通过伪造内置名称绕过。
21.8 本章小结
- Installed Agent 的关键不是“装进容器”，而是明确安装、用户、Headless、日志同步和异常传播契约。
- ATIF v1.7 要求连续 step、精确工具引用和真实的 LLM 调用计数；Validator 不验证事实真实性或脱敏。
raw_content与linear_history是 Harbor 生产者选项，不是 ATIF 根字段，也不会由BaseInstalledAgent自动实现。- 压缩后应使用 context boundary、continuation 和
is_copied_context表达真实输入历史。 - 配置脱敏不等于轨迹脱敏；原生日志、工具参数、observation 和外部日志都要独立治理。
- Viewer、SFT 与 RL 的消费契约不同；v0.18.0 对自定义 Agent 的 SFT 门槛和零调用过滤必须显式处理。
21.9 练习
- 给契约探针加入两个并行 tool call，证明每个 observation 只引用自己的
tool_call_id；再故意交换 ID，观察 Validator 错误。 - 把夹具改为一次真实的本地假模型调用：写入
llm_call_count=1和确定性的 token 夹具，但不得声称这些 token 来自外部模型。为llm_call_count=0 + metrics添加失败测试。 - 构造
trajectory.json与trajectory.cont-1.json，让第二段含 copied context。验证 continuation 链，并编写过滤 copied 与零调用 step 的 SFT 预处理器。 - 用一组已公开、已吊销的合成 Canary 字符串攻击
scrub()：Header、URL、JSON 嵌套、base64 和多行 stderr。记录漏检并改成字段 allowlist。 - 选择一个真实 Headless Agent CLI，写兼容性清单：固定版本、安装网络、非 root 用户、stdin、退出码、认证来源、原生日志、ATIF 转换、SFT/RL 缺口。先完成 mock 测试，再决定是否运行付费 Trial。