二、API Key 管理

OpenViking 多租户设计方案

Context

OpenViking 已定义了 UserIdentifier(account_id, user_id, agent_id) 三元组（PR #120），但多租户隔离尚未实施。当前状态：

• 认证：单一全局 api_key，HMAC 比较（openviking/server/auth.py）
• 无 RBAC：所有认证用户拥有完全访问权限
• 无存储隔离：VikingFS._uri_to_path 将 viking:// 映射到 /local/，无 account_id 前缀
• VectorDB：单一 context collection，无租户过滤
• 服务层：OpenVikingService 持有单例 _user，不支持请求级用户上下文

目标：实现完整的多租户支持，包括 API Key 管理、RBAC、存储隔离。不考虑向后兼容。

一、整体架构

Request
  │
  ▼
[Auth Middleware] ── 提取 API Key，按前缀分派：root 比对 / acct 查表 / user 解签名 → (account_id, user_id, role)
  │
  ▼
[RBAC Guard] ── 按角色检查操作权限
  │
  ▼
[RequestContext] ── UserIdentifier + Role 注入为 FastAPI 依赖
  │
  ▼
[Router] ── 传递 RequestContext 到 Service
  │
  ▼
[Service Layer] ── 请求级用户上下文（非单例）
  │
  ├─► [VikingFS] ── 单例，接受 RequestContext 参数，_uri_to_path 按 account_id 隔离，逐层权限过滤
  └─► [VectorDB] ── 单 collection，查询注入 account_id + owner_space 过滤

核心原则：

• 身份从 API Key 解析，贯穿全链路
• account 级隔离：AGFS 路径前缀 + VectorDB account_id 过滤
• user/agent 级隔离：目录遍历时逐层过滤，只展示当前用户有权限的目录和文件
• VikingFS 通过 RequestContext 获取租户和用户信息

二、API Key 管理

2.1 三层 Key 结构

2.2 User Key 方案（待评审选型）

User Key 有两种候选方案，需要在设计评审中确定。

共同点：两个方案的 key 格式都是 ovk_user_...，对外 API 完全一致，差异仅在内部生成和验证机制。

方案 A：加密式 token（确定性推导，适用用 user 过多，存储同步压力大）

核心思路：把 account_id:user_id 用 AES-GCM 加密生成 key。服务端解密即可还原身份，key 本身不需要存储。

生成：AES_GCM_Encrypt(private_key, "acme:alice") → ovk_user_Ek3xJ9fQm2... 验证：AES_GCM_Decrypt(private_key, key) → "acme:alice" → 查注册表确认用户存在

特征：同一个 (account_id, user_id) 永远推导出同一个 key（确定性）。

完整场景：

1. 管理员注册 alice，获取 key
   POST /api/v1/admin/accounts/acme/users         {"user_id": "alice"}  → 注册成功
   GET  /api/v1/admin/accounts/acme/users/alice/key                     → ovk_user_Ek3x...

2. 再次获取 alice 的 key（幂等，返回同一个值）
   GET  /api/v1/admin/accounts/acme/users/alice/key                     → ovk_user_Ek3x...（和上次完全一样）

3. alice 用 key 访问 API
   GET /api/v1/fs/ls?uri=viking://  -H "X-API-Key: ovk_user_Ek3x..."   → 200 OK

4. alice 丢了 key，管理员重新获取（无感恢复，key 不变）
   GET  /api/v1/admin/accounts/acme/users/alice/key                     → ovk_user_Ek3x...（还是同一个）

5. alice 的 key 泄露，想换一个 → 无法单独换，必须轮转 private_key（全部用户的 key 都会变）

6. 管理员移除 alice
   DELETE /api/v1/admin/accounts/acme/users/alice                       → 注册表删除 alice
   alice 再用 key 访问 → 解密成功但注册表找不到 → 401

存储：keys.json 只存注册列表，不存 key。ov.conf 需要 private_key 字段。

{ "accounts": { "acme": { "users": ["alice", "bob"] } } }

方案 B：随机 key + 查表（每个 user key 都存储）

核心思路：注册用户时生成随机 key，存入 keys.json。验证时查表匹配。

生成：secrets.token_hex(32) → ovk_user_7f3a...（存入 keys.json） 验证：在 keys.json 中查找 key → 得到 account_id 和 user_id

特征：每次获取 key 都生成新的随机值，旧 key 立即失效。

完整场景：

1. 管理员注册 alice，获取 key
   POST /api/v1/admin/accounts/acme/users         {"user_id": "alice"}  → 注册成功
   GET  /api/v1/admin/accounts/acme/users/alice/key                     → ovk_user_7f3a...

2. 再次获取 alice 的 key（生成新 key，旧 key 立即失效）
   GET  /api/v1/admin/accounts/acme/users/alice/key                     → ovk_user_b82d...（新 key）
   alice 用旧的 ovk_user_7f3a... 访问 → 401（已失效）

3. alice 用新 key 访问 API
   GET /api/v1/fs/ls?uri=viking://  -H "X-API-Key: ovk_user_b82d..."   → 200 OK

4. alice 丢了 key，管理员重新获取（生成新 key，alice 需要更新）
   GET  /api/v1/admin/accounts/acme/users/alice/key                     → ovk_user_c93f...（又一个新 key）

5. alice 的 key 泄露，想换一个 → 重新获取即可，只影响 alice，不影响 bob

6. 管理员移除 alice
   DELETE /api/v1/admin/accounts/acme/users/alice                       → 注册表和 key 一起删除
   alice 再用 key 访问 → 查表找不到 → 401

存储：keys.json 存注册列表 + 所有 user key。ov.conf 不需要 private_key。

{
    "accounts": { "acme": { "users": ["alice", "bob"] } },
    "user_keys": {
        "ovk_user_c93f...": { "account_id": "acme", "user_id": "alice" },
        "ovk_user_d91f...": { "account_id": "acme", "user_id": "bob" }
    }
}

方案对比

2.3 Key 存储

• Root Key：ov.conf 的 server 段
• Private Key（仅方案 A）：ov.conf 的 server 段
• Account Keys + User Keys/注册列表：AGFS /_system/keys.json，启动时加载到内存，写操作同步持久化
• _system 路径与租户 URI 自然隔离（合法 scope 为 resources/user/agent/session/queue/transactions）

为什么 Account/User Key 存 AGFS 而不是本地文件：

root_api_key 是静态配置（部署时手动写入 ov.conf），但 account key 和 user key 是运行时通过 Admin API 动态增删的，不能放 ov.conf。动态存储有两个选择：

选择 AGFS 的核心理由是多节点一致性：生产环境可能多个 OpenViking server 共享同一个 AGFS 后端，存 AGFS 则所有节点读同一份数据，一个节点创建的 account 其他节点立即可见。启动依赖不是问题——APIKeyManager 在 service.initialize() 之后才初始化（见 T5）。

2.4 新模块 openviking/server/api_keys.py

class APIKeyManager:
    """API Key 生命周期管理与解析"""

    def __init__(self, root_key: Optional[str], private_key: Optional[bytes], agfs_url: str)
    async def load()                                     # 从 AGFS 加载 keys.json 到内存
    async def save()                                     # 持久化到 AGFS
    def resolve(api_key: str) -> ResolvedIdentity        # Key → 身份 + 角色（含注册检查）
    def create_account(account_id: str) -> str           # 创建账户，返回 account key
    def delete_account(account_id: str)                  # 删除账户（清理 key 和用户列表）
    def derive_user_key(account_id, user_id) -> str      # 推导用户 key
    def register_user(account_id, user_id)               # 注册用户到账户
    def remove_user(account_id, user_id)                 # 移除用户

三、认证流程

3.1 核心类型

新建 openviking/server/identity.py：

class Role(str, Enum):
    ROOT = "root"
    ACCOUNT_ADMIN = "account_admin"
    USER = "user"

@dataclass
class ResolvedIdentity:
    role: Role
    account_id: Optional[str] = None
    user_id: Optional[str] = None
    agent_id: Optional[str] = None  # 来自 X-OpenViking-Agent header

@dataclass
class RequestContext:
    user: UserIdentifier       # account_id + user_id + agent_id
    role: Role

3.2 认证流程

1. 从 X-API-Key 或 Authorization: Bearer 提取 Key
2. 若未配置 root_api_key，进入本地开发模式：返回 (role=ROOT, account_id="default", user_id="default")
3. 按前缀分派：
   • ovk_root_ → HMAC 比对 root key
   • ovk_acct_ → 查内存 account_keys dict
   • ovk_user_ → 解析身份（方式取决于 2.2 选型）+ 注册检查
4. 从 X-OpenViking-Agent header 读取 agent_id（默认 "default"）
5. 构造 RequestContext(UserIdentifier(account_id, user_id, agent_id), role)

3.3 FastAPI 依赖注入

改动 openviking/server/auth.py：

async def resolve_identity(request, x_api_key, authorization, x_openviking_agent) -> ResolvedIdentity
def require_role(*roles) -> Depends  # 角色守卫工厂
def get_request_context(identity) -> RequestContext  # 构造 RequestContext

所有 Router 从 Depends(verify_api_key) 迁移到 Depends(get_request_context)。

四、RBAC 模型

4.1 为什么是三层角色

采用 ROOT / ACCOUNT_ADMIN / USER 三层角色，而非 ROOT + USER 两层，理由如下：

1. 与三层 Key 结构自然对应：root key → ROOT，account key → ACCOUNT_ADMIN，user key → USER。角色和 key 一一映射，认证解析逻辑简单直接。
2. 委托式管理链路：ROOT 创建 account 并下发 account key → 租户管理员自行注册用户并下发 user key。如果只有两层，所有用户管理都必须经过 ROOT，ROOT 成为运营瓶颈。
3. 权限最小化：user key 泄露只影响单个用户数据；account key 泄露影响该租户但不波及其他租户；root key 影响全局。三层划分使得每层 key 的爆炸半径最小。
4. 数据访问边界：ACCOUNT_ADMIN 可访问本 account 下所有用户数据（管理审计需要），USER 只能访问自己的隔离空间。这个区分在两层模型中无法表达。
5. 增量成本低：实现上只是 Role 枚举多一个值 + Admin API 多几个权限检查，不会增加架构复杂度。

4.2 角色与权限

权限矩阵：

4.3 Agent 归属（待评审）

Agent 通过 User 的 API Key 认证（继承用户身份）。但 Agent 的数据目录（记忆/技能/指令）有两种归属方式：

方案 A：Agent 目录按 agent_id 共享（跨用户共享）

目录仅由 agent_id 决定。多个用户使用同一 agent_id 时，访问同一份 agent 数据。

/{account_id}/agent/{md5(agent_id)[:12]}/memories/cases/
/{account_id}/agent/{md5(agent_id)[:12]}/skills/
/{account_id}/agent/{md5(agent_id)[:12]}/instructions/

场景：团队共用编程助手 agent，alice 教它的技能 bob 也能用。

方案 B：Agent 目录按 user_id + agent_id 隔离（每用户独立）

目录由 user_id + agent_id 共同决定。每个用户与 agent 的组合有独立数据空间。

/{account_id}/agent/{md5(user_id + agent_id)[:12]}/memories/cases/
/{account_id}/agent/{md5(user_id + agent_id)[:12]}/skills/
/{account_id}/agent/{md5(user_id + agent_id)[:12]}/instructions/

场景：alice 和 bob 各自使用同一 agent，但各自有独立的记忆和技能空间。

方案对比

讨论要点：产品形态偏向团队协作共享 agent 还是个人私有 agent？是否需要混合模式？

4.4 Admin API

新增 Router: openviking/server/routers/admin.py

POST   /api/v1/admin/accounts                              创建账户 (ROOT)
GET    /api/v1/admin/accounts                              列出账户 (ROOT)
DELETE /api/v1/admin/accounts/{account_id}                 删除账户 (ROOT)，级联清理数据
POST   /api/v1/admin/accounts/{account_id}/users           注册用户 (ROOT, ACCOUNT_ADMIN)
DELETE /api/v1/admin/accounts/{account_id}/users/{uid}     移除用户 (ROOT, ACCOUNT_ADMIN)
POST   /api/v1/admin/accounts/{account_id}/key             生成 Account Key (ROOT)
GET    /api/v1/admin/accounts/{account_id}/users/{uid}/key 下发 User Key (ROOT, ACCOUNT_ADMIN)

五、存储隔离

5.1 三维隔离模型

存储隔离有三个独立维度：account、user、agent。

• account：顶层隔离，不同租户之间完全不可见
• user：同一 account 内，不同用户的私有数据互不可见。用户记忆、资源、session 属于用户本人
• agent：同一 account 内，不同 agent 的数据互不可见。agent 的学习记忆、技能、指令属于 agent 本身。agent 目录归属方式（跨用户共享 vs 每用户独立）见 4.3 节待评审

Space 标识符：UserIdentifier 新增两个方法，拆分现有的 unique_space_name()：

def user_space_name(self) -> str:
    """用户级 space，不含 agent_id"""
    return f"{self._account_id}_{hashlib.md5(self._user_id.encode()).hexdigest()[:8]}"

def agent_space_name(self) -> str:
    """Agent 级 space，实现取决于 4.3 节选型"""
    # 方案 A: md5(agent_id)  方案 B: md5(user_id + agent_id)
    return hashlib.md5(self._agent_id.encode()).hexdigest()[:12]  # 方案 A 示例

5.2 各 Scope 的隔离方式

5.3 AGFS 文件系统隔离

改动文件: openviking/storage/viking_fs.py

VikingFS 保持单例，不持有任何租户状态。多租户通过参数传递实现：

调用链路：

1. 公开方法（ls、read、write 等）接收 ctx: RequestContext 参数
2. 公开方法从 ctx.account_id 提取 account_id，传给内部方法
3. 内部方法（_uri_to_path、_path_to_uri、_collect_uris 等）接收 account_id: str 参数，不依赖 ctx

URI → AGFS 路径转换（加 account_id 前缀）：

viking://user/{user_space}/memories/x + account_id="acme"
→ /local/acme/user/{user_space}/memories/x

AGFS 路径 → URI 转换（去 account_id 前缀）：

/local/acme/user/{user_space}/memories/x + account_id="acme"
→ viking://user/{user_space}/memories/x

返回给调用方的 URI 不含 account_id，对用户透明。account_id 只存在于 AGFS 物理路径层。

# 公开方法：接收 ctx，提取 account_id，结果按权限过滤
async def ls(self, uri: str, ctx: RequestContext) -> List[str]:
    path = self._uri_to_path(uri, account_id=ctx.account_id)
    entries = await self._agfs.ls(path)
    uris = [self._path_to_uri(e, account_id=ctx.account_id) for e in entries]
    return [u for u in uris if self._is_accessible(u, ctx)]  # 权限过滤，见 5.4

# 内部方法：只接收 account_id，不依赖 ctx
def _uri_to_path(self, uri: str, account_id: str = "") -> str:
    remainder = uri[len("viking://"):].strip("/")
    if account_id:
        return f"/local/{account_id}/{remainder}" if remainder else f"/local/{account_id}"
    return f"/local/{remainder}" if remainder else "/local"

def _path_to_uri(self, path: str, account_id: str = "") -> str:
    inner = path[len("/local/"):]                    # "acme/user/{space}/memories/x"
    if account_id and inner.startswith(account_id + "/"):
        inner = inner[len(account_id) + 1:]          # "user/{space}/memories/x"
    return f"viking://{inner}"

5.4 逐层权限过滤（Phase2）

user/agent 级隔离通过逐层遍历时过滤实现。用户可以从公共根目录（如 viking://resources）开始遍历，但每一层只能看到自己有权限的条目。

示例：

# alice（USER 角色）
ls viking://resources           → 只看到 {alice_user_space}/
ls viking://agent/memories      → 只看到 alice 当前 agent 的 {agent_space}/
ls viking://user/memories       → 只看到 {alice_user_space}/

# admin（ACCOUNT_ADMIN 角色）
ls viking://resources           → 看到所有用户的 space 目录

实现：VikingFS 新增 _is_accessible() 方法：

def _is_accessible(self, uri: str, ctx: RequestContext) -> bool:
    """判断当前用户是否能访问该 URI"""
    if ctx.role in (Role.ROOT, Role.ACCOUNT_ADMIN):
        return True

    # 结构性目录（不含 space，如 viking://user/memories）→ 允许遍历
    space_in_uri = self._extract_space_from_uri(uri)
    if space_in_uri is None:
        return True

    # 含 space 的 URI → 检查 space 是否属于当前用户或其 agent
    return space_in_uri in (
        ctx.user.user_space_name(),
        ctx.user.agent_space_name(),
    )

• 列举操作（ls、tree、glob）：AGFS 返回全量结果后，用 _is_accessible 过滤
• 读写操作（read、write、mkdir 等）：执行前调 _is_accessible 校验，无权限则拒绝
• 将来加 ACL：_is_accessible 内部扩展为查 ACL 表，接口不变（见 5.7）

5.5 VectorDB 租户隔离

改动文件: openviking/storage/collection_schemas.py

单 context collection，schema 新增两个字段：

• account_id（string）：account 级过滤
• owner_space（string）：user/agent 级过滤，值为记录所有者的 user_space_name() 或 agent_space_name()

查询过滤策略（由 retriever 根据 ctx 构造）：

写入时，Context 对象携带 account_id 和 owner_space，通过 EmbeddingMsgConverter 透传到 VectorDB。owner_space 始终只存原始所有者，不因共享而修改。

5.6 目录初始化

改动文件: openviking/core/directories.py

• 创建新账户时，初始化 account 级预设目录结构（公共根：viking://user、viking://agent、viking://resources 等）
• 用户首次访问时，懒初始化 user space 子目录（viking://user/{user_space}/memories/preferences 等）
• agent 首次使用时，懒初始化 agent space 子目录（viking://agent/{agent_space}/memories/cases 等）

5.7 未来 ACL 扩展方向（本版不实现）

当需要支持用户间资源共享（如 alice 共享某个 resources 目录给 bob）时，有两种扩展路径：

方案 a：独立 ACL 表

共享关系存储在独立的 ACL 表中（AGFS 或 VectorDB），不修改数据记录本身：

# ACL 记录
{ "grantee_space": "bob_user_space", "granted_uri_prefix": "viking://resources/{alice_space}/project-x" }

# bob 查询时
1. 解析可访问 space 列表：own spaces + 查 ACL 表得到被授权的 spaces
2. VectorDB filter: owner_space IN [bob_user_space, bob_agent_space, alice_user_space]
3. VikingFS _is_accessible: 检查 own space OR ACL 授权

优势：数据记录不变，授权/撤销即时生效，不需要批量更新记录。

方案 b：VectorDB 新增 shared_spaces 字段

在被共享的目录记录（非叶子节点）上新增 shared_spaces 列表字段，标记哪些 space 有访问权限：

# 目录记录
{ "uri": "viking://resources/{alice_space}/project-x", "owner_space": "alice_space", "shared_spaces": ["bob_space"] }

# bob 遍历时
_is_accessible 检查: owner_space 匹配 OR space in shared_spaces

优势：权限信息自包含在目录节点上，遍历时不需要额外查 ACL 表。需要配合遍历时的权限继承（子节点继承父目录的 shared_spaces）。

两种方案可结合使用。具体选型在 ACL 设计时确定。

六、配置变更

ov.conf server 段

{
  "server": {
    "host": "0.0.0.0",
    "port": 1933,
    "root_api_key": "ovk_root_...",
    "private_key": "hex-encoded-32-byte-secret",
    "cors_origins": ["*"]
  }
}

改动文件: openviking/server/config.py

@dataclass
class ServerConfig:
    host: str = "0.0.0.0"
    port: int = 1933
    root_api_key: Optional[str] = None   # 替代原 api_key
    private_key: Optional[str] = None    # 仅 User Key 方案 A 需要，见 2.2 节
    cors_origins: List[str] = field(default_factory=lambda: ["*"])

• root_api_key：替代原有的 api_key，用于 ROOT 身份认证。为 None 时进入本地开发模式（跳过认证），等同当前行为。
• private_key：hex 编码的 32 字节密钥，用于 User Key 方案 A（加密式 token）中 AES 加密/解密 user key payload。如果最终选型为方案 B（随机 key + 查表），此字段不需要，可移除。见 2.2 节 User Key 方案选型。

七、客户端变更

核心变化：多租户前客户端需要自行传递 account_id 和 user_id，多租户后这两个字段由服务端从 API Key 解析，客户端只需提供 api_key 和可选的 agent_id。

7.1 Python SDK

改动文件: openviking_cli/client/http.py, openviking_cli/client/sync_http.py

# 多租户后：身份由服务端从 api_key 解析
client = ov.SyncHTTPClient(
    url="http://localhost:1933",
    api_key="ovk_user_7f3a9c1e...",   # 服务端解析出 account_id + user_id
    agent_id="coding-agent",           # 可选，默认 "default"
)

7.2 CLI

改动文件: openviking_cli/session/user_id.py

ovcli.conf 新增 agent_id 字段：

{
  "url": "http://localhost:1933",
  "api_key": "ovk_user_7f3a9c1e...",
  "agent_id": "coding-agent",
  "output": "table"
}

CLI 发起请求时通过 X-OpenViking-Agent header 携带 agent_id。不再需要配置 account_id 和 user_id。

7.3 嵌入模式（不做多租户）

嵌入模式直接调用 Service 层，内部构造固定默认 RequestContext，不涉及 API Key 和 RBAC：

client = ov.Client(path="/data/openviking")
# 内部等价于 RequestContext(user=default, role=ROOT)

八、部署模式与单租户兼容（待评审）

8.1 背景

多租户改变了存储路径结构（新增 {account_id} 前缀和 {user_space} 子目录层）和 VectorDB schema（新增 account_id/owner_space 字段）。现有单租户部署只使用 root_api_key，不涉及 account/user/agent 概念，存储路径是扁平的。需要决定是否兼容这类部署。

8.2 方案 A：双模式共存（配置开关）

通过 ov.conf 的 multi_tenant 字段区分两种互斥的部署模式，启动时确定，运行期不可切换。

单租户模式（multi_tenant 不设或 false，默认）：

与当前行为完全一致，升级无影响：

• 不配置 root_api_key → dev 模式，跳过认证
• 配置 root_api_key → 简单 HMAC 比对，通过即可访问全部数据
• 存储路径不变：viking://resources/doc.md → /local/resources/doc.md
• VectorDB 记录不含租户字段，查询不注入租户过滤
• Admin API 端点不注册，访问返回 404

多租户模式（multi_tenant: true）：

{
  "server": {
    "multi_tenant": true,
    "root_api_key": "ovk_root_...",
    "cors_origins": ["*"]
  }
}

• 必须配置 root_api_key
• 启动时初始化 APIKeyManager，加载 keys.json，注册 Admin API 端点
• 存储路径带 account_id 前缀和 space 目录层
• VectorDB 写入携带 account_id + owner_space，查询按角色注入过滤
• 完整 RBAC（ROOT / ACCOUNT_ADMIN / USER）

两种模式对比：

启动时分派：

if config.multi_tenant:
    api_key_manager = APIKeyManager(root_key=config.root_api_key, ...)
    await api_key_manager.load()
    app.state.api_key_manager = api_key_manager
    app.include_router(admin_router)        # 注册 Admin API
    viking_fs.set_multi_tenant(True)        # VikingFS 使用隔离路径
else:
    app.state.api_key_manager = None        # 单租户，无 key 管理
    viking_fs.set_multi_tenant(False)       # VikingFS 使用扁平路径

约束：

• 同一份数据不能跨模式使用（路径结构和 VectorDB schema 不同）
• 嵌入模式始终为单租户，不受 multi_tenant 配置影响

8.3 方案 B：仅多租户

不保留单租户模式，所有部署统一走多租户路径结构。现有单租户部署升级后需要重新导入数据。

• 不配置 root_api_key → dev 模式，跳过认证，使用默认 account/user
• 配置 root_api_key → ROOT 角色，account_id="default"，user_id="default"
• 存储路径统一：viking://resources/doc.md → /local/default/resources/{default_user_space}/doc.md
• VectorDB 记录统一携带 account_id/owner_space 字段

九、实施分期与任务拆解

Phase 1：API 层多租户能力定义（已完成）

实施顺序：T1 → T3 → T2 → T4 → T5 → T10/T11 并行 → T12

T1: 身份与角色类型定义

新建 openviking/server/identity.py，依赖：无

定义三个类型，供后续所有任务引用：

from enum import Enum
from dataclasses import dataclass
from typing import Optional
from openviking.session.user_id import UserIdentifier

class Role(str, Enum):
    ROOT = "root"
    ACCOUNT_ADMIN = "account_admin"
    USER = "user"

@dataclass
class ResolvedIdentity:
    """认证中间件的输出：从 API Key 解析出的原始身份信息"""
    role: Role
    account_id: Optional[str] = None   # ROOT 可能无 account_id
    user_id: Optional[str] = None      # ROOT/ACCOUNT_ADMIN 可能无 user_id
    agent_id: Optional[str] = None     # 来自 X-OpenViking-Agent header

@dataclass
class RequestContext:
    """请求级上下文，贯穿 Router → Service → VikingFS 全链路"""
    user: UserIdentifier    # 完整三元组（account_id, user_id, agent_id）
    role: Role

    @property
    def account_id(self) -> str:
        return self.user.account_id

注意：RequestContext 而非 ResolvedIdentity 是下游使用的类型。ResolvedIdentity 只在 auth 层内部使用，转换为 RequestContext 后传递。原因：ResolvedIdentity 的字段都是 Optional（ROOT 没有 account_id），而 RequestContext.user 是确定的 UserIdentifier——对于 ROOT，填入 account_id="default"。

T3: ServerConfig 更新

修改 openviking/server/config.py，依赖：无

改动点：

# 改前
@dataclass
class ServerConfig:
    host: str = "0.0.0.0"
    port: int = 1933
    api_key: Optional[str] = None                          # ← 删除
    cors_origins: List[str] = field(default_factory=lambda: ["*"])

# 改后
@dataclass
class ServerConfig:
    host: str = "0.0.0.0"
    port: int = 1933
    root_api_key: Optional[str] = None                     # ← 替代 api_key
    multi_tenant: bool = False                             # ← 新增，见第 8 节
    private_key: Optional[str] = None                      # ← 新增，仅 User Key 方案 A
    cors_origins: List[str] = field(default_factory=lambda: ["*"])

load_server_config() 中对应修改读取字段：

config = ServerConfig(
    host=server_data.get("host", "0.0.0.0"),
    port=server_data.get("port", 1933),
    root_api_key=server_data.get("root_api_key"),          # ← 改
    multi_tenant=server_data.get("multi_tenant", False),   # ← 新增
    private_key=server_data.get("private_key"),             # ← 新增
    cors_origins=server_data.get("cors_origins", ["*"]),
)

T2: API Key Manager

新建 openviking/server/api_keys.py，依赖：T1

内存数据结构

# keys.json 在 AGFS 中的结构
{
    "account_keys": {
        "ovk_acct_abc123...": {
            "account_id": "acme_corp",
            "created_at": "2026-02-12T10:00:00Z"
        }
    },
    "accounts": {
        "acme_corp": {
            "created_at": "2026-02-12T10:00:00Z",
            "users": ["alice", "bob"]
        }
    }
}

对应内存结构：

self._account_keys: Dict[str, str] = {}       # {key_str -> account_id}
self._accounts: Dict[str, AccountInfo] = {}    # {account_id -> AccountInfo}
# AccountInfo = dataclass(created_at, users: Set[str])

方法逻辑

__init__(root_key, private_key, agfs_url)：

• 存储 root_key, private_key
• 创建 pyagfs.AGFSClient(agfs_url) 用于读写 /_system/keys.json

async load()：

• 从 AGFS 读取 /_system/keys.json
• 若文件不存在，初始化空结构并写入
• 解析 JSON 填充 _account_keys 和 _accounts

async save()：

• 将内存数据序列化为 JSON，写回 AGFS /_system/keys.json

resolve(api_key) -> ResolvedIdentity：

if key.startswith("ovk_root_"):
    hmac.compare_digest(key, self._root_key) → ResolvedIdentity(role=ROOT)
elif key.startswith("ovk_acct_"):
    account_id = self._account_keys.get(key) → ResolvedIdentity(role=ACCOUNT_ADMIN, account_id)
elif key.startswith("ovk_user_"):
    account_id, user_id = 解析身份（方式取决于 2.2 选型）
    检查 account_id 在 _accounts 中存在
    检查 user_id 在 _accounts[account_id].users 中存在
    → ResolvedIdentity(role=USER, account_id, user_id)
else:
    raise UnauthenticatedError

create_account(account_id) -> str：

• 验证 account_id 格式（复用 UserIdentifier 的验证正则）
• 检查 account_id 不重复
• 生成 ovk_acct_{secrets.token_hex(32)}
• 写入 _account_keys 和 _accounts
• 调用 save() 持久化
• 返回 account key

delete_account(account_id)：

• 从 _accounts 删除
• 从 _account_keys 中删除对应的 key
• 调用 save() 持久化
• 注意：AGFS 数据和 VectorDB 数据的级联清理不在 APIKeyManager 内完成，由 Admin Router 的调用方负责

derive_user_key(account_id, user_id) -> str：

• 检查 account_id 和 user_id 已注册
• 调用 derive_user_key(self._private_key, account_id, user_id) 返回

register_user(account_id, user_id)：

• 检查 account_id 存在
• 将 user_id 加入 _accounts[account_id].users
• 调用 save()

remove_user(account_id, user_id)：

• 从 _accounts[account_id].users 中移除
• 调用 save()

T4: 认证中间件重写

重写 openviking/server/auth.py，依赖：T1, T2, T3

删除现有的 verify_api_key()、get_user_header()、get_agent_header()，替换为：

resolve_identity(request, x_api_key, authorization, x_openviking_agent) -> ResolvedIdentity：

1. api_key_manager = request.app.state.api_key_manager
2. 若 api_key_manager 为 None（本地开发模式）：
   返回 ResolvedIdentity(role=ROOT, account_id="default", user_id="default", agent_id="default")
3. 提取 key（同现有逻辑：X-API-Key 或 Bearer）
4. identity = api_key_manager.resolve(key)
5. identity.agent_id = x_openviking_agent or "default"
6. 返回 identity

get_request_context(identity: ResolvedIdentity = Depends(resolve_identity)) -> RequestContext：

account_id = identity.account_id or "default"
user_id = identity.user_id or "default"
agent_id = identity.agent_id or "default"
return RequestContext(
    user=UserIdentifier(account_id, user_id, agent_id),
    role=identity.role,
)

require_role(*allowed_roles) -> dependency：

def require_role(*allowed_roles: Role):
    async def _check(ctx: RequestContext = Depends(get_request_context)):
        if ctx.role not in allowed_roles:
            raise PermissionDeniedError(f"Requires role: {allowed_roles}")
        return ctx
    return _check

T5: App 初始化集成

修改 openviking/server/app.py，依赖：T2, T4

改动点在 create_app() 和 lifespan()：

# 改前
app.state.api_key = config.api_key

# 改后（按 multi_tenant 配置分派，见第 8 节）
if config.multi_tenant:
    api_key_manager = APIKeyManager(
        root_key=config.root_api_key,
        agfs_url=service._agfs_url,
    )
    await api_key_manager.load()
    app.state.api_key_manager = api_key_manager
    app.include_router(admin_router)        # 注册 Admin API
    viking_fs.set_multi_tenant(True)        # VikingFS 使用隔离路径
else:
    app.state.api_key_manager = None        # 单租户模式

删除 app.state.api_key。

注意：APIKeyManager 初始化必须在 service.initialize() 之后，因为需要 AGFS URL。时序是：

1. service = OpenVikingService() → 启动 AGFS
2. await service.initialize() → 初始化 VikingFS/VectorDB
3. api_key_manager = APIKeyManager(agfs_url=service._agfs_url) → 用 AGFS 读 keys.json
4. await api_key_manager.load()

T10: Router 依赖注入迁移

修改文件：server/routers/ 下所有 router，依赖：T4

Phase 1 改动（已完成）

所有 router 的依赖从 verify_api_key 迁移到 get_request_context，但 service 调用不变（ctx 仅接收，不向下传递）：

# 改前
@router.get("/ls")
async def ls(uri: str, _: bool = Depends(verify_api_key)):
    service = get_service()
    result = await service.fs.ls(uri)
    ...

# Phase 1 改后（ctx 接收但不传递）
@router.get("/ls")
async def ls(uri: str, _ctx: RequestContext = Depends(get_request_context)):
    service = get_service()
    result = await service.fs.ls(uri)  # service 调用不变
    ...

Phase 2 改动（待实施，依赖 T9）

Service 层适配完成后，将 ctx 传给 service 方法：

# Phase 2 改后
async def ls(uri: str, ctx: RequestContext = Depends(get_request_context)):
    service = get_service()
    result = await service.fs.ls(uri, ctx=ctx)  # 传递 ctx
    ...

需要改的 router 列表

T11: Admin Router

新建 openviking/server/routers/admin.py，依赖：T2, T4

端点逻辑

POST /api/v1/admin/accounts — 创建账户

权限：require_role(ROOT)
入参：{"account_id": "acme_corp"}
逻辑：
  1. api_key_manager.create_account(account_id) → account_key
  2. 为新账户初始化 AGFS 目录结构（调用 DirectoryInitializer）
返回：{"account_id": "acme_corp", "account_key": "ovk_acct_..."}

GET /api/v1/admin/accounts — 列出账户

权限：require_role(ROOT)
逻辑：遍历 api_key_manager._accounts
返回：[{"account_id": "acme_corp", "created_at": "...", "user_count": 2}, ...]

DELETE /api/v1/admin/accounts/{account_id} — 删除账户

权限：require_role(ROOT)
逻辑：
  1. api_key_manager.delete_account(account_id)
  2. 级联清理 AGFS：rm -r /{account_id}/ （通过 VikingFS）
  3. 级联清理 VectorDB：删除 account_id=X 的所有记录
返回：{"deleted": true}

POST /api/v1/admin/accounts/{account_id}/users — 注册用户

权限：require_role(ROOT, ACCOUNT_ADMIN)
额外检查：ACCOUNT_ADMIN 只能操作自己的 account
入参：{"user_id": "alice"}
逻辑：api_key_manager.register_user(account_id, user_id)
返回：{"account_id": "acme_corp", "user_id": "alice"}

DELETE /api/v1/admin/accounts/{account_id}/users/{uid} — 移除用户

权限：require_role(ROOT, ACCOUNT_ADMIN)
逻辑：api_key_manager.remove_user(account_id, uid)
返回：{"deleted": true}

POST /api/v1/admin/accounts/{account_id}/key — 生成新 Account Key

权限：require_role(ROOT)
逻辑：生成新 key，替换旧 key（一个 account 只有一个 key）
返回：{"account_key": "ovk_acct_..."}

GET /api/v1/admin/accounts/{account_id}/users/{uid}/key — 下发 User Key

权限：require_role(ROOT, ACCOUNT_ADMIN)
逻辑：api_key_manager.derive_user_key(account_id, uid)
返回：{"user_key": "ovk_user_..."}

注册到 server/routers/__init__.py 和 server/app.py。

T12: 客户端 SDK 更新

Phase 1 改动（已完成）：HTTP 客户端

修改文件：openviking_cli/client/http.py, openviking_cli/client/sync_http.py，依赖：T4

HTTP 模式新增 agent_id 参数，通过 X-OpenViking-Agent header 发送：

def __init__(self, url=None, api_key=None, agent_id=None):
    self._agent_id = agent_id

# headers 构建
headers = {}
if self._api_key:
    headers["X-API-Key"] = self._api_key
if self._agent_id:
    headers["X-OpenViking-Agent"] = self._agent_id

身份由服务端从 API Key 解析，客户端不构造 UserIdentifier。

Phase 2 改动（待实施，依赖 T9）：嵌入模式

修改文件：openviking/client/local.py，依赖：T9

嵌入模式不做多租户。Service 层 ctx 适配完成后，LocalClient 构造固定默认 ctx 传给 service 方法：

def __init__(self, path=None):
    self._service = OpenVikingService(path=path)
    self._ctx = RequestContext(
        user=UserIdentifier.the_default_user(),
        role=Role.ROOT,
    )

async def ls(self, uri, ...):
    return await self._service.fs.ls(uri, ctx=self._ctx)

嵌入模式不涉及 API Key、RBAC、多租户隔离。

T14-P1: 认证与管理测试（已完成）

T14a: APIKeyManager 单元测试

• root key 验证（正确/错误）
• account key 生成和解析
• user key 推导和验证（正确签名/错误签名/未注册用户）
• 用户注册/移除后 key 有效性变化
• keys.json 持久化和加载

T14b: 认证中间件测试

• 三种 key 类型的 resolve_identity 流程
• 本地开发模式（无 root_api_key）
• require_role 守卫
• 无效 key / 缺失 key 的错误码

T14e: 回归

• 现有测试改为使用 dev mode（不配置 root_api_key）

Phase 2：存储层隔离实现（后续）

实施顺序：T6/T7 并行 → T8 → T9 → T13 → T14-P2

T6: VikingFS 多租户改造

修改 openviking/storage/viking_fs.py，依赖：T1

需要加 ctx 参数的方法（全部公开方法）

VikingFS 有以下公开方法需要加 ctx: RequestContext 参数：

核心改动

VikingFS 新增 _multi_tenant: bool 标志，启动时由 app 设置（见 T5、第 8 节）。

_uri_to_path 和 _path_to_uri 行为取决于模式：

• 单租户（_multi_tenant=False）：忽略 account_id，保持旧扁平路径
• 多租户（_multi_tenant=True）：加 account_id 前缀

def _uri_to_path(self, uri: str, account_id: str = "") -> str:
    remainder = uri[len("viking://"):].strip("/")
    if self._multi_tenant and account_id:
        return f"/local/{account_id}/{remainder}" if remainder else f"/local/{account_id}"
    return f"/local/{remainder}" if remainder else "/local"

def _path_to_uri(self, path: str, account_id: str = "") -> str:
    if path.startswith("viking://"):
        return path
    elif path.startswith("/local/"):
        inner = path[7:]  # 去掉 /local/
        if self._multi_tenant and account_id and inner.startswith(account_id + "/"):
            inner = inner[len(account_id) + 1:]  # 去掉 account_id 前缀
        return f"viking://{inner}"
    ...

私有方法的处理

内部方法 _collect_uris, _delete_from_vector_store, _update_vector_store_uris, _ensure_parent_dirs, _read_relation_table, _write_relation_table 不直接接受 ctx，而是由公开方法调用时已经完成了 _uri_to_path 转换，传入的是 AGFS path。

但 _collect_uris 内部调用 _path_to_uri 时需要 account_id 来正确还原 URI → 需要传 account_id 或 ctx 给这些内部方法。

策略：内部方法统一加 account_id: str = "" 参数（不用整个 ctx），公开方法从 ctx.account_id 提取后传入。

T7: VectorDB schema 扩展

修改 openviking/storage/collection_schemas.py，依赖：无

在 context_collection() 的 Fields 列表中新增：

{"FieldName": "account_id", "FieldType": "string"},

位置放在 id 之后、uri 之前。

同时修改 TextEmbeddingHandler.on_dequeue()：inserted_data 中应已包含 account_id（由 T8 中 EmbeddingMsg 携带）。此处不需要额外改动，只需确保 schema 定义了该字段。

T8: 检索层与数据写入的租户过滤

修改文件：retrieve/hierarchical_retriever.py, core/context.py，依赖：T1, T7

8a. Context 对象增加 account_id 和 owner_space

openviking/core/context.py 中 Context 类需增加两个字段：

account_id: str = ""      # 所属 account
owner_space: str = ""     # 所有者的 user_space_name() 或 agent_space_name()

to_dict() 输出包含这两个字段，EmbeddingMsgConverter.from_context() 无需改动即可透传到 VectorDB。

上游构造 Context 时需从 RequestContext 填入这两个字段：

• ResourceService / SkillProcessor → account_id=ctx.account_id, owner_space=ctx.user.user_space_name() 或 agent_space_name()（取决于 scope）
• MemoryExtractor.create_memory() → 同上
• DirectoryInitializer._ensure_directory() → 同上

8b. HierarchicalRetriever 注入多级过滤

retrieve/hierarchical_retriever.py 的 retrieve() 方法需接受 ctx: RequestContext 参数，根据角色构造不同粒度的过滤条件（见第五节 5.5）：

async def retrieve(self, query: TypedQuery, ctx: RequestContext, ...) -> QueryResult:
    filters = []
    if ctx.role == Role.ACCOUNT_ADMIN:
        filters.append({"op": "must", "field": "account_id", "conds": [ctx.account_id]})
    elif ctx.role == Role.USER:
        filters.append({"op": "must", "field": "account_id", "conds": [ctx.account_id]})
        filters.append({"op": "must", "field": "owner_space",
                        "conds": [ctx.user.user_space_name(), ctx.user.agent_space_name()]})
    # ROOT 无过滤

调用方（VikingFS.find(), VikingFS.search()）从 ctx 传入。

T9: Service 层适配

修改文件：service/core.py 及 service/fs_service.py, service/search_service.py, service/session_service.py, service/resource_service.py, service/relation_service.py, service/pack_service.py, service/debug_service.py，依赖：T1, T6

核心变更：去除 _user 单例

OpenVikingService.__init__() 中删除 self._user。 set_dependencies() 调用中删除 user=self.user 参数。

各 sub-service 改动模式

所有 sub-service 当前的模式是：

class XXXService:
    def set_dependencies(self, viking_fs, ..., user=None):
        self._viking_fs = viking_fs
        self._user = user  # ← 删除

    async def some_method(self, ...):
        # 使用 self._viking_fs 和 self._user

改为：

class XXXService:
    def set_dependencies(self, viking_fs, ...):  # 去掉 user
        self._viking_fs = viking_fs

    async def some_method(self, ..., ctx: RequestContext):  # 加 ctx
        # 使用 self._viking_fs 和 ctx

逐 service 改动清单

FSService（service/fs_service.py）：

• 当前：ls(uri), tree(uri), stat(uri), mkdir(uri), rm(uri), mv(old, new), read(uri), abstract(uri), overview(uri), grep(uri, pattern), glob(pattern, uri)
• 改为：所有方法加 ctx 参数，传递给 VikingFS 调用

SearchService（service/search_service.py）：

• 当前：find(query, ...), search(query, ...)
• 改为：加 ctx，传给 VikingFS.find/search

SessionService（service/session_service.py）：

• 当前：session(session_id), sessions(), delete(session_id), extract(session_id) 使用 self._user
• 改为：加 ctx，构造 Session 时从 ctx 获取 user，extract 时传 ctx.user 给 compressor
• session 路径变为 viking://session/{ctx.user.user_space_name()}/{session_id}

ResourceService（service/resource_service.py）：

• 当前：add_resource(...), add_skill(...) 使用 self._user
• 改为：加 ctx，构造 Context 时填入 account_id=ctx.account_id, owner_space=ctx.user.user_space_name()（resources scope）或 ctx.user.agent_space_name()（agent scope）
• 资源路径使用 viking://resources/{ctx.user.user_space_name()}/...，技能路径使用 viking://agent/skills/{ctx.user.agent_space_name()}/...

RelationService（service/relation_service.py）：

• 当前：relations(uri), link(from, to), unlink(from, to)
• 改为：加 ctx，传给 VikingFS

PackService（service/pack_service.py）：

• 当前：export_ovpack(uri), import_ovpack(data)
• 改为：加 ctx，传给 VikingFS

DebugService（service/debug_service.py）：

• 当前：get_status(), observer 等系统级方法
• 改为：部分方法可能不需要 ctx（如 health check），但 observer 需要

T13: 目录初始化适配

修改文件：core/directories.py，依赖：T6, T8

核心改动

DirectoryInitializer 当前在 service.initialize() 中调用，初始化全局预设目录。多租户后改为三种初始化时机：

1. 创建新 account 时（Admin API T11）→ 初始化该 account 的公共根目录（viking://user、viking://agent、viking://resources 等）
2. 用户首次访问时 → 懒初始化 user space 子目录（viking://user/{user_space}/memories/preferences 等）
3. agent 首次使用时 → 懒初始化 agent space 子目录（viking://agent/{agent_space}/memories/cases 等）

方法签名改为接受 ctx: RequestContext：

async def initialize_account_directories(self, ctx: RequestContext) -> int:
    """初始化 account 级公共根目录"""
    ...

async def initialize_user_directories(self, ctx: RequestContext) -> int:
    """初始化 user space 子目录"""
    ...

as

…(truncated)