[RFC] Accessor 数据同步标准化与 Email 数据源试点

[zihengli-bytedance]

目标:为 OV Accessor 层定义一套统一的数据同步标准——鉴权声明、cursor 增量、镜像清理、可靠性四个维度——与商业化 Viking Connectors 语义对齐;以新增 EmailAccessor(IMAP)作为试点从零验证标准接口,存量 Accessor 零改动。

1. 总体定位

OV 的数据接入已经通过 Accessor 模式实现了主动拉取(WebFeedAccessor 抓整站、FeishuAccessor 拉文档、GitAccessor clone),但各 Accessor 在鉴权、增量、镜像、可靠性上各自为战。需要一套统一的数据同步定义,把这些跨源共性能力前置,在不同场景各自扩展:

• 开源场景 — Accessor 内置模块,本地运行,灵活实验版,快速功能探索。
• 商业化场景 — Viking Connectors,云端企业级,大规模数据同步。

类比 OV 内置的 Bot 模块:开源 Bot 定义了 Agent 使用 OV 的模式,Demo MVP 价值;商业化用生产级 ArkClaw 基于标准 MCP 协议对接。数据接入同理:两种实现范式,服务不同场景,汇入同一管线(Parser → AGFS → L0/L1/L2)。

本 RFC 做两件事:

1. 定义 Accessor 层的标准接口(第 4、5 节),语义与商业化 Connector 对齐;
2. 以新增 EmailAccessor(IMAP) 为试点(第 6 节),验证标准的完备性后再推广。

2. 为什么需要标准化

2.1 数据范式全景 — 不只是文档

当前 Accessor 覆盖的主要是"文档批量导入",但 Agent 需要的 context 更多:

范式	典型数据源	接入特征	现状
文档	飞书 Wiki、GitHub README、网站 sitemap	半静态,有版本,批量拉取	✅ Accessor 已覆盖
对话/交互	Slack 频道、飞书群聊、Email 线程、GitHub Discussions	时序追加,thread/reply 结构,只增不改	❌ 需要 append-only cursor 语义
结构化数据	数据库表、飞书多维表格、Google Sheets	有 schema,cell 级变更	❌ 需要保留结构而非转 markdown
实时事件	GitHub webhook、飞书事件订阅、监控告警	事件驱动、推送式	❌ 需要 push 模型而非 pull
状态快照	项目看板、OKR 进度、日历日程	关心"此刻是什么"	❌ 需要快照 + diff 语义
多媒体	Figma 设计稿、会议录音	需 VLM/ASR 转译	⚠️ 有 VLM,无标准接入

每种范式对标准的要求不同,标准设计必须留出扩展空间。本次试点选择的 email 属于"对话/交互"范式,恰好检验 cursor 抽象能否容纳文档水位之外的语义(见 6.2)。

2.2 四个维度的共性问题

鉴权:每个数据源鉴权体系完全不同(GitHub PAT/App/OAuth 三种、飞书三层 token、IMAP 用户名密码/OAuth2/App Password 协议级分裂)。现状是每个 Accessor 自己处理(WebFeed 不管、Feishu 自己搞 OAuth)。标准应该让 Accessor 声明需要什么认证(类型 + 字段),框架统一管理凭证存储、刷新、过期。

增量:全量重建无法规模化(飞书团队知识库 ~5000 篇全量 1 小时+,增量 <2 分钟;Confluence 企业空间 ~50000 篇全量不现实)。每个源的增量语义不同(GitHub since、Notion last_edited_time、IMAP SEARCH SINCE),但 cursor 的持久化和回灌是通用的——这部分标准化,Accessor 只管拉数据,框架管存取。

镜像:源端删除了内容,OV 里残留脏数据——Agent 用了过时知识,比没有知识更危险。模式统一为:全量同步时 Accessor 返回完整 doc_id 集合,框架 diff 出孤儿,自动清理;增量模式不删除(看不到全貌)。

可靠性:中途崩溃要断点续传、单文档失败要隔离、限流要退避、超时要熔断、长任务要进度可见。Accessor 上报错误分类(transient/permanent)和进度,框架负责重试、隔离、持久化断点。

3. Accessor vs Connector:两种范式,一套语义

	OV Accessor(开源)	商业化 Connector
数据流	access() → 本地目录 → Parser → AGFS	云端 runtime 采集 → KB / OV
上手成本	一个文件,can_handle() + access()	插件化开发,声明式接入
鉴权	各自处理 → 本 RFC 标准化	✅ 凭证统一托管
增量	❌ 全量重建 → 本 RFC 标准化	✅ cursor 真增量
孤儿清理	❌ → 本 RFC 标准化	✅ 全量模式 diff 清理
可靠性	❌ → 本 RFC 标准化	✅ 生产级重试/熔断/心跳
定位	灵活实验,本地优先	稳定扩展,云端企业级

明确一个命名决定:不把 Accessor 改名为 Connector。两个名字的区分本身承载"两种范式"的信息——Accessor 是开源、本地、实验优先的接入模式,Connector 是云端企业级的同步产品;同名会抹掉这层区分,"这是哪种 connector"每次都要额外说明。标准化的是语义,不是名字。

4. 标准接口设计

接口语义与商业化 Connector 的生产实现对齐,鉴权、cursor 增量、孤儿清理等机制已在商业化侧经过实际验证。Accessor 标准复用同一套语义,保持接口更轻,新增能力全部可选。

4.1 现有接口不变

class DataAccessor(ABC):
    def can_handle(self, source, **kwargs) -> bool: ...
    async def access(self, source, **kwargs) -> LocalResource: ...
    def priority(self) -> int: ...

写一个最简 Accessor 仍然只需要这三个方法。

4.2 新增标准类型

@dataclass
class ConnectionStatus:
    success: bool
    message: str = ""

@dataclass
class AccessError:
    doc_id: str | None          # 关联的文档,None 表示源级错误
    kind: Literal["transient", "permanent"]
    message: str

@dataclass
class AccessResult:
    resource: LocalResource            # 本地目录(同原返回值)
    cursor: dict | None = None         # 增量 cursor(可选,对框架不透明)
    doc_ids: set[str] | None = None    # 本次看到的完整 doc_id 集合(可选)
    errors: list[AccessError] = field(default_factory=list)

• doc_ids 非空时,框架 diff doc_ids vs OV 中已有文档,自动清理孤儿;
• 增量模式下 doc_ids 为 None,不做删除——与商业化 Connector 行为一致(全量模式 diff 删除,增量不删);
• errors 中 transient 由框架重试,permanent 隔离并记录。

4.3 鉴权声明(可选能力)

Accessor 只声明需要什么认证(JSON Schema),框架兑现凭证的存储、加密和刷新:

class EmailAccessor(DataAccessor):
    def auth_spec(self) -> dict:
        """声明需要的认证字段 — JSON Schema"""
        return {
            "type": "object",
            "properties": {
                "host":     {"type": "string", "description": "IMAP server hostname"},
                "port":     {"type": "integer", "default": 993},
                "use_ssl":  {"type": "boolean", "default": True},
                "username": {"type": "string"},
                "password": {"type": "string", "description": "password or app password"},
            },
            "required": ["host", "username", "password"],
        }

    def check(self, auth: dict) -> ConnectionStatus:
        """校验凭证有效性;不抛异常,包装为 ConnectionStatus 返回"""
        try:
            with open_imap(**auth):
                return ConnectionStatus(success=True)
        except IMAPAuthError as e:
            return ConnectionStatus(success=False, message=str(e))

框架的职责:检测到 Accessor 实现了 auth_spec(),就接管凭证存储、加密、WatchTask 刷新。

4.4 增量同步(可选能力)

access() 接受 cursor / progress 可选参数,返回 AccessResult:

class EmailAccessor(DataAccessor):
    async def access(self, source, cursor=None, progress=None, **kwargs) -> AccessResult:
        """
        cursor: 上次运行的 cursor(框架回灌),首次为 None
        """
        since = cursor.get("sync_checkpoint") if cursor else None
        tmp_dir = self._create_temp_dir("email_")
        seen_ids: set[str] = set()

        for i, mail in enumerate(imap_fetch_since(source, since)):
            write_markdown(tmp_dir, mail)          # 转 markdown,复用现有 parser
            seen_ids.add(mail.doc_id)              # Message-ID,缺失时 <mailbox8>_<uid> 兜底
            if progress:
                progress(done=i + 1, total=None)   # IMAP 总数未知场景 total 可为 None

        return AccessResult(
            resource=LocalResource(path=tmp_dir, source_type="email", ...),
            cursor={"sync_checkpoint": now_utc_iso()},   # 对框架不透明,下次原样回灌
            doc_ids=seen_ids if since is None else None,  # 仅全量模式返回,用于孤儿检测
        )

4.5 标准能力总览

能力	Accessor 侧	框架职责
鉴权声明	auth_spec() -> dict(JSON Schema)	凭证存储、加密、刷新
凭证校验	check(auth) -> ConnectionStatus	连接前校验入口
增量同步	access(source, cursor, ...) 返回 AccessResult.cursor	cursor 持久化、下次回灌
数据输出	写文件到 LocalResource 目录	目录 → Parser → AGFS
镜像清理	全量模式返回 doc_ids	diff 出孤儿并清理
错误分类	AccessResult.errors(transient/permanent)	transient 重试,permanent 隔离
进度上报	progress(done, total) 回调	日志/状态/回调展示
同步模式	cursor 是否为 None + doc_ids 是否返回	决定是否做孤儿清理

5. 框架侧职责

能力	落点(拟)	说明
凭证托管	新增 credential store	按 accessor 的 auth_spec() 存取,静态加密,预留 OAuth2 refresh 钩子
cursor 持久化	WatchTask 状态	挂在 watch 任务上,任务完成后原子写入,下次调度回灌
孤儿清理	resource 层 diff	doc_ids vs 已导入文档集合,diff 后走既有删除路径
重试/隔离	watch 执行器	transient 指数退避重试,permanent 记录并跳过
进度	日志 + WatchTask 状态	商业化侧对应云端任务心跳/进度上报

6. 试点:EmailAccessor(IMAP)

6.1 为什么新增数据源,而不是改造存量

1. 对标准的检验更真实。试点要回答"这套标准接口够不够用",最好的检验是只靠标准接口从零写出一个新源。改造存量会把标准验证和存量行为兼容搅在一起,出了问题分不清是标准缺陷还是迁移回归。
2. 存量是生产路径,风险不对称。directory parser 和 code parser 直接依赖 GitAccessor 的元数据格式;试点阶段标准接口还会反复调整,不应让生产 accessor 跟着草案震荡。
3. git clone 不是 cursor 增量的好样本。git 的增量天然由 git 协议解决,cursor/孤儿语义在它身上很勉强。

6.2 为什么选 email

1. 可与商业化实现做 1:1 语义对照。商业化侧已有生产级的 Email Connector(IMAP + 密码鉴权、真增量、邮件转 markdown)。试点要验证的正是"同一套语义、两种实现",email 是能做最严格对照的源。
2. 鉴权 schema 表达力检验更充分。多字段 auth(host/port/use_ssl/username/password)比单 token 更能检验 auth_spec() 的表达力;且 OAuth2(Gmail/M365 已关闭 IMAP 基础认证)只在 schema 层预留,scope 决策商业化已验证过,试点照抄。
3. 补上"对话/交互"范式空白。email 是 append-only 时序数据,检验 cursor 不透明设计能否容纳文档水位之外的语义。
4. 可测试性最好。本地起 dovecot/greenmail 即可全封闭测试,不依赖外网 API、无 token 和限流对 CI 的干扰。

6.3 范围界定

项	试点范围内	明确不做
协议	IMAP(SSL 993 默认)	POP3、JMAP、Gmail API、Graph API
鉴权	用户名 + 密码/App Password;auth_spec() 声明	OAuth2 实现(仅 schema 预留 type 扩展位)
增量	sync_checkpoint 时间水位,IMAP SEARCH SINCE;since_days 防历史巨量回拉	UID/UIDVALIDITY 水位(时间水位实现简单且与商业化一致,UID 方案留作后续)
镜像	全量模式返回完整 doc_id 集合,框架 diff 清理	增量模式删除
输出	邮件转 markdown 落地,复用现有 markdown parser	新增 eml parser
可靠性	单 mailbox 失败隔离并汇总;任一 mailbox 失败则不推进 cursor(下轮重拉,宁可重复不可丢失)	限流退避(IMAP 无标准限流响应,留给第二波)

6.4 试点语义设计要点

语义点	设计
auth 字段	host / port / use_ssl / username / password,required: host + username + password
doc_id	Message-ID;缺失时 <mailbox_hash[:8]>_<uid> 兜底,跨 mailbox 不串号
cursor	{"sync_checkpoint": <UTC ISO,Z 后缀>} 时间水位
增量过滤	严格 < 水位比较;== 水位 的边界邮件下轮重拉,由下游去重兜底
失败语义	单 mailbox 失败隔离并继续其余 mailbox,收尾统一报错且整轮不推进 cursor
输出形态	邮件转 markdown 文件 → 现有 markdown parser

以上语义与商业化 Email Connector 保持一致,一致性通过内部对照验证。

6.5 验收标准

1. EmailAccessor 仅依赖第 4 节标准接口实现,不引入任何私有框架钩子;
2. 增量:第二次同步只拉新邮件,cursor 断点崩溃恢复不丢邮件(允许边界重拉);
3. 镜像:源端删信后全量同步能清理对应文档;
4. 鉴权:凭证由框架存取,accessor 代码不落盘任何凭证;
5. 语义对照:6.4 表中设计要点全部落实,与商业化 Email Connector 行为一致(内部对照验证);
6. 存量 accessor 与既有测试零改动、零回归。

7. 代码结构(拟)

openviking/parse/accessors/
  base.py               # +AccessResult / AccessError / ConnectionStatus
                        # +DataAccessor.auth_spec() / check() 可选方法
  email_accessor.py     # 新增:EmailAccessor(IMAP)
  registry.py           # 不变

openviking/resource/    # 框架侧(与 watch 体系集成)
  credential_store.py   # 新增:凭证存储/加密
  watch_*.py            # cursor 持久化与回灌、孤儿 diff、重试/隔离(挂在既有 watch 调度上)

8. 推广节奏

试点(新增 EmailAccessor)
  -> 标准微调(以试点暴露的问题修订第 4 节接口)
  -> 第二波:回灌一个存量 accessor(web_feed 优先,sitemap 孤儿清理收益最直接)
  -> 补验限流退避维度(GitHub API 类源或 web_feed)
  -> 标准定稿

不在只有 greenfield 样本时宣布标准定稿——试点验证"新源好不好写",第二波验证"存量迁移路径可行",两者都过才算标准成立。

9. 非目标

• 不把 Accessor 改名为 Connector(两个名字承载两种范式的区分,见第 3 节);
• 不动存量 accessor(local/http/git/feishu/web_feed/web_importer 零改动);
• 不新增 eml parser(accessor 侧仅做邮件 → markdown 的最小转换,解析仍归现有 parser);
• 不实现 OAuth2(仅 auth_spec() schema 预留);
• 不覆盖 push 型数据源(webhook receiver 是另一个 RFC 的事)。

10. 设计原则

• 最简 Accessor 永远是三个方法;新增能力全部可选,按需渐进。
• cursor 对框架不透明:框架只管持久化和回灌,语义(时间水位/UID/checkpoint)完全由 Accessor 定义。
• 孤儿清理只在全量模式发生;增量模式看不到全貌,永不删除。
• 失败时宁可重复,不可丢失:cursor 只在整轮成功后推进,边界 doc 重拉由下游去重兜底。
• Accessor 声明能力,框架兑现能力:凭证、cursor、diff、重试、进度都是框架职责。
• 开源与商业化对齐的是语义,不是代码:同一套约定,两种实现,汇入同一管线。

Open Questions

1. cursor 存储的具体落点:复用 WatchTask 持久化,还是独立的 sync state store?(倾向前者,但 WatchTask 目前按调度周期建模,一次性导入的 cursor 归属需要定义)
2. 凭证加密方案:本地 keyring / 文件加密 / 明文 + 权限位,开源场景的默认选择? (倾向本地明文，和现有实现保持一致，不和其他需求耦合)
3. 试点是否顺带引入 SourceType.EMAIL 之外的 mailbox 层级映射(一个 mailbox 一个目录 vs 平铺)? (倾向一个 mailbox 一个目录,与商业化一致)