深入 Open Agent SDK（五）：会话持久化与安全防线

本文是「深入 Open Agent SDK (Swift)」系列第五篇。系列目录见这里。

Agent 不只是一次性问答工具。真正有用的 Agent 要做到三件事：记住上下文（上次聊到哪了）、控制权限（哪些操作能做）、审计行为（谁在什么时候干了什么）。Open Agent SDK 用四个子系统来覆盖这些需求——SessionStore、PermissionPolicy、SandboxSettings、HookRegistry。

这篇文章分析这四个子系统的实现细节，看它们各自怎么工作，以及怎么组合起来构建一个安全的 Agent。

一、会话持久化：SessionStore

Agent Loop 每次运行会产生一组 messages 数组。如果不保存，进程退出就没了。SessionStore 负责把这些对话历史持久化到磁盘，下次启动时恢复。

SessionStore 是什么

SessionStore 是一个 actor，所有方法都需要 await 调用。默认把会话存在 ~/.open-agent-sdk/sessions/ 目录下，每个 session 一个子目录，里面放一个 transcript.json。

let sessionStore = SessionStore()  // 默认路径
let sessionStore = SessionStore(sessionsDir: "/custom/path")  // 自定义路径

五个核心操作

SessionStore 提供五个核心方法，覆盖会话的完整生命周期。

save — 保存会话。把 messages 数组和元数据序列化成 JSON 写入磁盘：

try await sessionStore.save(
    sessionId: "my-session",
    messages: messages,
    metadata: PartialSessionMetadata(
        cwd: "/project",
        model: "claude-sonnet-4-6",
        summary: "代码分析会话",
        tag: "analysis",
        firstPrompt: "分析项目结构"
    )
)

存储结构长这样：

~/.open-agent-sdk/sessions/
  my-session/
    transcript.json    // { "metadata": {...}, "messages": [...] }

文件权限是 0600，目录权限是 0700——只有当前用户能读写。每次 save 会保留第一次创建时的 createdAt 时间戳，只更新 updatedAt。

load — 加载会话。从磁盘读取 transcript.json，反序列化为 SessionData：

if let data = try await sessionStore.load(sessionId: "my-session") {
    print("Messages: \(data.metadata.messageCount)")
    print("Model: \(data.metadata.model)")
    // data.messages 是 [[String: Any]] 数组
}

load 支持分页参数 limit 和 offset，不需要加载全部消息时可以只取尾部：

// 只加载最近 50 条消息
let recent = try await sessionStore.load(sessionId: "my-session", limit: 50, offset: nil)

list — 列出所有会话，按 updatedAt 降序排列（最近的在前）：

let sessions = try await sessionStore.list(limit: 10)
for session in sessions {
    print("\(session.id) — \(session.summary ?? "(无标题)") [\(session.messageCount) 条消息]")
}

SessionMetadata 包含 id、cwd、model、createdAt、updatedAt、messageCount，以及可选的 summary、tag、firstPrompt、gitBranch、fileSize。

fork — 分叉会话。从已有会话复制消息到新 session，可以指定截断点：

// 完整复制
let newId = try await sessionStore.fork(sourceSessionId: "my-session")

// 只复制前 10 条消息
let truncatedId = try await sessionStore.fork(
    sourceSessionId: "my-session",
    upToMessageIndex: 10
)

// 指定新 session ID
let customId = try await sessionStore.fork(
    sourceSessionId: "my-session",
    newSessionId: "forked-session"
)

delete — 删除整个会话目录：

let deleted = try await sessionStore.delete(sessionId: "my-session")

此外还有 rename（改标题）和 tag（打标签）两个辅助方法。

会话恢复的三种模式

把 SessionStore 注入 Agent 后，SDK 提供三种恢复策略：

1. 指定 sessionId 恢复

最直接的方式：给定一个 session ID，Agent 启动时从 SessionStore 加载历史消息，追加到 messages 数组前面：

let agent = createAgent(options: AgentOptions(
    apiKey: apiKey,
    model: "claude-sonnet-4-6",
    sessionStore: sessionStore,
    sessionId: "my-session"      // 指定恢复哪个 session
))

2. continueRecentSession — 自动接续最近的会话

不知道 session ID 时，让 SDK 自动找最近的一个：

let agent = createAgent(options: AgentOptions(
    apiKey: apiKey,
    model: "claude-sonnet-4-6",
    sessionStore: sessionStore,
    continueRecentSession: true   // 自动加载最近的 session
))

内部实现是调 sessionStore.list() 取第一个（已按 updatedAt 降序排列），把它的 ID 作为恢复目标。

3. forkSession + resumeSessionAt — 分叉并截断

在已有会话的基础上分叉一个新分支，还可以截断到指定消息：

let agent = createAgent(options: AgentOptions(
    apiKey: apiKey,
    model: "claude-sonnet-4-6",
    sessionStore: sessionStore,
    sessionId: "my-session",
    forkSession: true,                     // 复制到新 session
    resumeSessionAt: "msg-uuid-123"        // 截断到这条消息
))

SDK 内部的解析顺序是：先 continueRecentSession 确定 session ID，再 forkSession 创建分叉，再 resumeSessionAt 截断历史。这三个选项可以独立使用也可以组合。

SessionStore 的安全细节

SessionStore 在 session ID 校验上做了路径遍历防护：

private func validateSessionId(_ sessionId: String) throws {
    guard !sessionId.isEmpty else {
        throw SDKError.sessionError(message: "Session ID must not be empty")
    }
    let forbidden = ["/", "\\", ".."]
    for component in forbidden {
        if sessionId.contains(component) {
            throw SDKError.sessionError(message: "Session ID contains invalid character: '\(component)'")
        }
    }
}

session ID 里不能包含 /、\、..——防止攻击者通过构造 ID 来读写预期之外的路径。

二、权限控制：PermissionPolicy

会话持久化解决了"记住"的问题，权限控制解决的是"能做什么"的问题。

六种 PermissionMode

SDK 定义了 6 种权限模式：

模式	行为
`default`	每次工具执行前询问用户
`plan`	只读工具直接执行，写操作需要确认
`auto`	自动执行所有工具，危险操作除外
`acceptEdits`	文件编辑自动执行，其他操作需要确认
`dontAsk`	不询问用户，根据上下文自动判断
`bypassPermissions`	跳过所有权限检查

let agent = createAgent(options: AgentOptions(
    apiKey: apiKey,
    model: "claude-sonnet-4-6",
    permissionMode: .plan  // 只读工具直接跑，写操作要确认
))

canUseTool 回调：比 PermissionMode 更细粒度

permissionMode 是全局开关，粒度比较粗。如果你需要按工具名称或工具属性做精细控制，用 canUseTool 回调：

let agent = createAgent(options: AgentOptions(
    apiKey: apiKey,
    model: "claude-sonnet-4-6",
    permissionMode: .bypassPermissions,
    canUseTool: { tool, input, context in
        if tool.name == "Bash" {
            return CanUseToolResult.deny("Bash is not allowed")
        }
        return nil  // nil 表示"我没意见，交给 permissionMode 决定"
    }
))

canUseTool 返回 CanUseToolResult?。返回 nil 表示该回调没有意见，交给下一个检查环节；返回非 nil 结果时，SDK 用回调的决定，不再看 permissionMode。

CanUseToolResult 有三个工厂方法：

CanUseToolResult.allow()                              // 允许
CanUseToolResult.deny("原因")                          // 拒绝
CanUseToolResult.allowWithInput(modifiedInput)         // 允许但修改输入参数

allowWithInput 比较少见但很实用——你可以在权限检查时修改工具的输入参数。比如把文件写入路径重定向到安全目录。

策略模式：可组合的权限规则

直接写闭包虽然灵活，但不方便复用。SDK 提供了 PermissionPolicy 协议，把权限判断封装成可组合的策略：

public protocol PermissionPolicy: Sendable {
    func evaluate(
        tool: ToolProtocol,
        input: Any,
        context: ToolContext
    ) async -> CanUseToolResult?
}

SDK 内置了四个策略：

ToolNameAllowlistPolicy — 白名单，只允许指定的工具：

let policy = ToolNameAllowlistPolicy(allowedToolNames: ["Read", "Glob", "Grep"])
// Write、Edit、Bash 等工具全部被拒绝

ToolNameDenylistPolicy — 黑名单，拒绝指定的工具：

let policy = ToolNameDenylistPolicy(deniedToolNames: ["Bash", "Write"])
// 其他工具正常执行

ReadOnlyPolicy — 只允许只读工具（isReadOnly == true）：

let policy = ReadOnlyPolicy()
// Read、Glob、Grep、WebSearch 等只读工具允许
// Write、Edit、Bash 等变更工具被拒绝

CompositePolicy — 组合多个策略，按顺序评估：

let policy = CompositePolicy(policies: [
    ToolNameDenylistPolicy(deniedToolNames: ["Bash"]),
    ReadOnlyPolicy()
])
// 先检查黑名单（Bash 被拒绝），再检查只读策略

CompositePolicy 的评估规则：

任何子策略返回 deny，整体 deny（短路）
子策略返回 nil（没意见），跳过
所有子策略都 allow 或没意见，整体 allow

用 canUseTool(policy:) 桥接函数把策略转成回调：

let policy = CompositePolicy(policies: [
    ToolNameDenylistPolicy(deniedToolNames: ["Bash"]),
    ReadOnlyPolicy()
])

let agent = createAgent(options: AgentOptions(
    apiKey: apiKey,
    model: "claude-sonnet-4-6",
    permissionMode: .bypassPermissions,
    canUseTool: canUseTool(policy: policy)
))

三、沙盒机制：SandboxSettings + SandboxChecker

权限控制管的是"这个工具能不能执行"，沙盒管的是"这个操作在不在允许范围内"。比如 Bash 工具通过了权限检查，但你还得确保它不会 rm -rf /。

SandboxSettings 的配置项

let sandbox = SandboxSettings(
    // 路径控制
    allowedReadPaths: ["/project/"],
    allowedWritePaths: ["/project/build/"],
    deniedPaths: ["/etc/", "/var/"],

    // 命令控制
    deniedCommands: ["rm", "sudo"],           // 黑名单
    // allowedCommands: ["git", "swift"],     // 白名单（和黑名单二选一）

    // 行为控制
    allowNestedSandbox: false,
    autoAllowBashIfSandboxed: false,          // 沙箱激活时自动批准 Bash
    allowUnsandboxedCommands: false,
    enableWeakerNestedSandbox: false,

    // 网络控制
    network: SandboxNetworkConfig(
        allowedDomains: ["api.example.com"],
        allowLocalBinding: false
    )
)

路径和命令各有两种模式：

路径：allowedReadPaths / allowedWritePaths 是白名单（空数组=全部允许），deniedPaths 是黑名单（优先级更高）
命令：allowedCommands 是白名单（设为非 nil 就只允许列出的命令），deniedCommands 是黑名单。allowedCommands 优先级高于 deniedCommands

SandboxChecker 的执行逻辑

SandboxChecker 是一个无状态的枚举类，提供 isPathAllowed、checkPath、isCommandAllowed、checkCommand 四个静态方法。isXxx 返回 Bool，checkXxx 不通过时抛出 SDKError.permissionDenied。

路径检查用前缀匹配加段边界保证：

// /project/ 匹配 /project/src/file.swift
// /project/ 不匹配 /project-backup/file.swift
SandboxChecker.isPathAllowed("/project/src/main.swift", for: .read, settings: sandbox)
// -> true

SandboxChecker.isPathAllowed("/project-backup/old.swift", for: .read, settings: sandbox)
// -> false（段边界不匹配）

实现关键在于 SandboxPathNormalizer——先把路径规范化（解析 ..、.、symlink），再做前缀比较时保证尾部有 / 来强制段边界。

// 路径遍历攻击会被 normalize 掉
let normalized = SandboxPathNormalizer.normalize("/project/src/../../etc/passwd")
// -> "/etc/passwd"，然后被 deniedPaths 拦截

命令检查分三个阶段：

Shell 元字符检测——识别 bash -c "cmd"、$(cmd)、`cmd` 等绕过模式
Basename 提取——从 /usr/bin/rm -rf /tmp 提取出 rm
白名单/黑名单匹配

// 黑名单里有 "rm"
SandboxChecker.isCommandAllowed("rm -rf /tmp", settings: blocklist)
// -> false

// 路径形式的命令也能识别
SandboxChecker.isCommandAllowed("/usr/bin/rm -rf /tmp", settings: blocklist)
// -> false（提取 basename 得到 "rm"）

// 反斜杠绕过
SandboxChecker.isCommandAllowed("\\rm -rf /tmp", settings: blocklist)
// -> false（去掉前导 \ 后得到 "rm"）

// 引号绕过
SandboxChecker.isCommandAllowed("\"rm\" -rf /tmp", settings: blocklist)
// -> false（去掉引号后得到 "rm"）

// 子 shell 绕过
SandboxChecker.isCommandAllowed("bash -c \"rm -rf /tmp\"", settings: blocklist)
// -> false（递归检查内部命令）

对于无法可靠解析的命令（比如多层嵌套的 bash -c "bash -c 'rm ...'"），默认拒绝。

命令参数中的文件路径也会被提取并检查——如果命令里出现了 deniedPaths 中的路径，命令也会被拒绝。

autoAllowBashIfSandboxed

这个选项是沙盒和权限系统的桥梁。当 autoAllowBashIfSandboxed = true 时，Bash 工具会跳过 canUseTool 权限回调检查，但仍然经过 SandboxChecker.checkCommand() 的命令过滤。

设计思路是：如果你已经配了完善的沙盒规则，Bash 命令能做什么已经被限制住了，不需要再弹一次权限确认。

四、Hook 系统：20+ 生命周期事件

前三个系统解决的是"能不能做"的问题，Hook 系统解决的是"做了之后要知道"和"做之前要干预"的问题。

20+ 个 HookEvent

SDK 定义了 24 个生命周期事件：

事件	触发时机
`preToolUse`	工具执行前
`postToolUse`	工具执行成功后
`postToolUseFailure`	工具执行失败后
`sessionStart`	Agent 会话开始
`sessionEnd`	Agent 会话结束
`stop`	Agent Loop 停止
`subagentStart`	子 Agent 启动
`subagentStop`	子 Agent 完成
`userPromptSubmit`	用户提交 prompt
`permissionRequest`	权限检查发生
`permissionDenied`	权限被拒绝
`taskCreated`	任务创建
`taskCompleted`	任务完成
`configChange`	配置变更
`cwdChanged`	工作目录变更
`fileChanged`	文件变更
`notification`	通知事件
`preCompact`	对话压缩前
`postCompact`	对话压缩后
`teammateIdle`	团队成员空闲
`setup`	Agent 初始化
`worktreeCreate`	工作树创建
`worktreeRemove`	工作树移除

函数 Hook vs Shell Hook

Hook 有两种实现方式：函数回调和 Shell 命令。

函数 Hook — Swift 闭包，适合进程内逻辑：

await registry.register(.preToolUse, definition: HookDefinition(
    handler: { input in
        // input 是 HookInput，包含 event、toolName、toolInput、sessionId 等
        return HookOutput(message: "拦截成功", block: true)
    }
))

Shell Hook — 外部命令，适合集成非 Swift 脚本：

await registry.register(.preToolUse, definition: HookDefinition(
    command: "python3 /path/to/check.py"  // HookInput 通过 stdin JSON 传入
))

Shell Hook 通过 ShellHookExecutor 执行：用 /bin/bash -c 启动进程，把 HookInput 序列化为 JSON 写入 stdin，从 stdout 读取 HookOutput JSON。Shell 命令的标准输出如果不是合法 JSON，会被包装成 HookOutput(message: stdout)。

Shell Hook 的环境变量里会注入 HOOK_EVENT、HOOK_TOOL_NAME、HOOK_SESSION_ID、HOOK_CWD，方便脚本直接用环境变量判断上下文。

HookRegistry 的注册与执行

HookRegistry 是一个 actor，内部维护 [HookEvent: [HookDefinition]] 映射：

let registry = HookRegistry()

// 注册函数 Hook
await registry.register(.preToolUse, definition: HookDefinition(
    handler: { input in
        return HookOutput(message: "Bash blocked", block: true)
    },
    matcher: "Bash"  // 只匹配 Bash 工具
))

// 注册 Shell Hook
await registry.register(.postToolUse, definition: HookDefinition(
    command: "/usr/bin/logger 'Tool executed'",
    timeout: 5000  // 5 秒超时
))

// 执行所有注册在某事件上的 Hook
let results = await registry.execute(.preToolUse, input: hookInput)
// results: [HookOutput]，包含所有匹配的 Hook 的返回值

matcher 过滤：每个 HookDefinition 可以设一个 matcher（正则表达式）。执行时先检查 input.toolName 是否匹配 matcher，不匹配就跳过这个 Hook。matcher 为 nil 时匹配所有工具。

超时处理：函数 Hook 用 withThrowingTaskGroup 实现超时——把实际执行和 Task.sleep 放在同一个 TaskGroup 里，谁先完成用谁。超时的 Hook 不影响其他 Hook 执行。Shell Hook 通过 DispatchQueue.asyncAfter 设置超时，到时间就 terminate 进程。

执行顺序：同一事件上的 Hook 按注册顺序串行执行。

HookOutput 的能力

HookOutput 可以做这些事：

HookOutput(
    message: "日志消息",                          // 附加信息
    block: true,                                  // 拦截操作
    notification: HookNotification(               // 发送通知
        title: "警告",
        body: "检测到危险操作",
        level: .warning
    ),
    permissionUpdate: PermissionUpdate(           // 动态修改权限
        tool: "Bash",
        behavior: .deny
    ),
    systemMessage: "请在沙箱内操作",               // 注入系统消息
    reason: "安全策略",                            // 拦截原因
    updatedInput: ["command": "echo safe"],       // 修改工具输入
    decision: .block                              // 显式 approve/block
)

其中 block: true 会阻止工具执行，返回一个错误结果给 LLM。permissionUpdate 可以在 Hook 运行时动态修改工具权限。updatedInput 可以替换工具的输入参数。

五、实战组合：构建一个安全的 Agent

四个子系统各有分工：

SessionStore — 记住对话历史
PermissionPolicy — 控制工具能不能执行
SandboxSettings — 限制操作范围
HookRegistry — 审计和拦截

下面用一个完整的例子展示怎么把它们组合起来：

import Foundation
import OpenAgentSDK

// 1. 创建 SessionStore
let sessionStore = SessionStore()

// 2. 创建 HookRegistry，注册审计和安全拦截
let hookRegistry = HookRegistry()

// 记录所有工具执行
await hookRegistry.register(.postToolUse, definition: HookDefinition(
    handler: { input in
        if let toolName = input.toolName {
            print("[审计] 工具 \(toolName) 执行完成")
        }
        return nil
    }
))

// 拦截 Bash 中的危险命令
await hookRegistry.register(.preToolUse, definition: HookDefinition(
    handler: { input in
        return HookOutput(
            message: "Bash 被安全策略拦截",
            block: true,
            decision: .block
        )
    },
    matcher: "Bash"
))

// 记录权限拒绝事件
await hookRegistry.register(.permissionDenied, definition: HookDefinition(
    handler: { input in
        print("[安全告警] 权限被拒绝: \(input.error ?? "unknown")")
        return nil
    }
))

// 会话生命周期追踪
await hookRegistry.register(.sessionStart, definition: HookDefinition(
    handler: { _ in print("[会话] 开始"); return nil }
))
await hookRegistry.register(.sessionEnd, definition: HookDefinition(
    handler: { _ in print("[会话] 结束"); return nil }
))

// 3. 配置沙盒：限制路径和命令
let sandbox = SandboxSettings(
    allowedReadPaths: ["/project/"],
    allowedWritePaths: ["/project/src/", "/project/tests/"],
    deniedPaths: ["/etc/", "/var/", "/tmp/"],
    deniedCommands: ["rm", "sudo", "chmod", "chown"],
    autoAllowBashIfSandboxed: false,
    allowNestedSandbox: false
)

// 4. 配置权限策略：只读 + 排除 Bash
let policy = CompositePolicy(policies: [
    ToolNameDenylistPolicy(deniedToolNames: ["Bash"]),
    ReadOnlyPolicy()
])

// 5. 创建 Agent，注入所有组件
let agent = createAgent(options: AgentOptions(
    apiKey: "sk-...",
    model: "claude-sonnet-4-6",
    systemPrompt: "你是一个代码分析助手。只能读取文件，不能修改。",
    maxTurns: 10,
    permissionMode: .bypassPermissions,
    canUseTool: canUseTool(policy: policy),
    sessionStore: sessionStore,
    sessionId: "analysis-session",
    hookRegistry: hookRegistry,
    sandbox: sandbox
))

// 6. 执行查询
let result = await agent.prompt("分析项目中的 Swift 源文件结构")
print(result.text)

// 7. 后续恢复会话
let resumedAgent = createAgent(options: AgentOptions(
    apiKey: "sk-...",
    model: "claude-sonnet-4-6",
    permissionMode: .bypassPermissions,
    canUseTool: canUseTool(policy: policy),
    sessionStore: sessionStore,
    sessionId: "analysis-session",  // 同一个 session ID，自动恢复历史
    hookRegistry: hookRegistry,
    sandbox: sandbox
))

let continued = await resumedAgent.prompt("继续分析测试文件")
print(continued.text)

这个 Agent 的安全特性：

权限层：CompositePolicy 确保只有只读工具能执行，Bash 被黑名单排除
沙盒层：即使工具通过了权限检查，也受路径限制——只能读 /project/ 下的文件，不能碰 /etc/、/var/
Hook 层：所有工具执行被记录（审计），Bash 调用被 preToolUse Hook 二次拦截
会话层：对话自动保存和恢复，重启后能继续之前的工作

多层防御的好处是：即使某一层的配置有疏漏，其他层还能兜底。比如你误把 Bash 加进了白名单，Hook 的 matcher 还能拦截；即使 Hook 没拦住，沙盒的命令过滤还能挡。

小结

SessionStore、PermissionPolicy、SandboxSettings、HookRegistry 四个系统各管一件事，但组合起来就是一套完整的安全框架：

SessionStore 的 actor 隔离和 session ID 校验保证了存储安全
PermissionPolicy 的策略组合提供了灵活的权限管理
SandboxChecker 的路径规范化和段边界匹配防止目录穿越
HookRegistry 的 matcher 过滤和超时机制确保了 Hook 系统的可靠性

下一篇看 SDK 的 多 LLM 提供商：怎么同时支持 Anthropic、OpenAI 和其他 LLM，Provider 协议的设计，以及运行时切换模型的机制。

系列文章：

第 0 篇：Open Agent SDK (Swift)：用原生 Swift 并发构建 AI Agent 应用
第 1 篇：Agent Loop 内核：从 prompt 到多轮对话的完整运转机制
第 2 篇：34 个工具的背后：工具协议、三层架构与自定义扩展
第 3 篇：MCP 集成实战：让 Agent 连接万物
第 4 篇：多 Agent 协作：子代理、团队与任务编排
第 5 篇：会话持久化与安全防线（本文）
第 6 篇：多 LLM 提供商与运行时控制

GitHub：terryso/open-agent-sdk-swift