Connectors & Worker 深度解析
这一页只看 OpenBot 的一个局部模块，用真实源码行号解释它承担的契约、失败路径和与主架构的关系。
导入链与契约
▸小 runner，强契约
worker connector runner 很小，但契约明确。ConnectorPersistence 类型从 worker/src/connector-runner.ts:6 定义，runConnector 入口是 worker/src/connector-runner.ts:17，discover 调用在 worker/src/connector-runner.ts:23；batch 路径由 worker/src/connector-runner.ts:27 选择，能力检查是 worker/src/connector-runner.ts:30，逐条 apply 位于 worker/src/connector-runner.ts:35，cursor 提交和成功记录分别是 worker/src/connector-runner.ts:36、worker/src/connector-runner.ts:40 显示 adapter.discover、persistence.cursor、batch/individual writes 和 recordRun 是它的全部接口。它不是业务 extractor，而是所有 knowledge source 同步的事务外壳。
1import type { 2 ConnectorAdapter, 3 ConnectorChange, 4} from "../../server/src/connectors/contract"; 5 6export type ConnectorPersistence = { 7 cursor: () => Promise<string | null>; 8 apply?: (change: ConnectorChange) => Promise<void>; 9 commitCursor?: (cursor: string) => Promise<void>; 10 persistBatch?: ( 11 changes: ConnectorChange[], 12 cursor: string | null, 13 ) => Promise<void>; 14 recordRun?: (status: "succeeded" | "failed") => Promise<void>; 15}; 16 17export async function runConnector( 18 adapter: ConnectorAdapter, 19 persistence: ConnectorPersistence, 20 mode: "sync" | "reconcile" = "sync", 21) { 22 try { 23 const discovered = await adapter.discover({ 24 cursor: await persistence.cursor(), 25 mode, 26 }); 27 if (persistence.persistBatch) { 28 await persistence.persistBatch(discovered.changes, discovered.nextCursor); 29 } else { 30 if (!persistence.apply || !persistence.commitCursor) { 31 throw new Error( 32 "Connector persistence must support batch or individual writes.", 33 ); 34 } 35 for (const change of discovered.changes) await persistence.apply(change); 36 if (discovered.nextCursor !== null) { 37 await persistence.commitCursor(discovered.nextCursor); 38 } 39 } 40 await persistence.recordRun?.("succeeded"); 41 } catch (error) { 42 await persistence.recordRun?.("failed"); 43 throw error; 44 } 45}
核心实现，逐段拆
▸batch 优先，逐条 fallback
worker/src/connector-runner.ts:27 先用 persistBatch，worker/src/connector-runner.ts:30 检查 individual write 能力，worker/src/connector-runner.ts:35 逐条 apply，worker/src/connector-runner.ts:36 到 worker/src/connector-runner.ts:38 最后提交 cursor。
17export async function runConnector( 18 adapter: ConnectorAdapter, 19 persistence: ConnectorPersistence, 20 mode: "sync" | "reconcile" = "sync", 21) { 22 try { 23 const discovered = await adapter.discover({ 24 cursor: await persistence.cursor(), 25 mode, 26 }); 27 if (persistence.persistBatch) { 28 await persistence.persistBatch(discovered.changes, discovered.nextCursor); 29 } else { 30 if (!persistence.apply || !persistence.commitCursor) { 31 throw new Error( 32 "Connector persistence must support batch or individual writes.", 33 ); 34 } 35 for (const change of discovered.changes) await persistence.apply(change); 36 if (discovered.nextCursor !== null) { 37 await persistence.commitCursor(discovered.nextCursor); 38 } 39 } 40 await persistence.recordRun?.("succeeded"); 41 } catch (error) { 42 await persistence.recordRun?.("failed"); 43 throw error; 44 } 45}
▸channel start 的 UI 顺序也有事务感
useStartChannel 从 app/src/lib/channels/start.ts:13 导出，mutation 创建在 app/src/lib/channels/start.ts:16，start 函数是 app/src/lib/channels/start.ts:20；createChannel 调用在 app/src/lib/channels/start.ts:21，cache seed 是 app/src/lib/channels/start.ts:22，首消息暂存是 app/src/lib/channels/start.ts:23，navigate 从 app/src/lib/channels/start.ts:24 到 app/src/lib/channels/start.ts:27 完成 展示 UI 开新 channel 的顺序：create mutation、seed query cache、stash first message、navigate。app/src/lib/channels/start.ts:10 到 app/src/lib/channels/start.ts:11 的注释说明这是为了让第一条消息在 thread join 前可见。
13export function useStartChannel() { 14 const queryClient = useQueryClient(); 15 const navigate = useNavigate(); 16 const createChannel = useMutation(createChannelMutationOptions(queryClient)); 17 18 return { 19 pending: createChannel.isPending, 20 start: async (agentId: string, text: string) => { 21 const channel = await createChannel.mutateAsync([agentId]); 22 queryClient.setQueryData(channelKeys.detail(channel.id), channel); 23 stashFirstMessage(channel.id, text); 24 await navigate({ 25 params: { channelId: channel.id }, 26 replace: true, 27 to: "/channel/$channelId", 28 }); 29 }, 30 }; 31}
关键流程
▸同步游标只代表已持久化位置
这个流程的要点是 cursor 代表“已经安全持久化到哪里”，所以 worker/tests/connector-runner.test.ts:27 的失败测试很重要：如果 apply 抛错，worker/tests/connector-runner.test.ts:48 断言 cursors 仍为空。
边界 / 失败模式
▸失败不推进游标
worker/src/connector-runner.ts:31在缺少 batch 和 individual write 能力时直接抛错，避免静默丢数据。worker/src/connector-runner.ts:41到worker/src/connector-runner.ts:43失败会记录 run failed 并继续抛出，调度层能重试。worker/tests/connector-runner.test.ts:51验证 reconcile mode 被传给 adapter，说明同步模式不是 runner 私自决定。