oh-my-opencode 后台代理全局并发上限设计：maxBackgroundAgents 配置从 Zod Schema 到 ConcurrencyManager 落地的完整方案

oh-my-opencode 后台代理全局并发上限设计maxBackgroundAgents 配置从 Zod Schema 到 ConcurrencyManager 落地的完整方案【免费下载链接】oh-my-openagentOmO: Drop your tokens. Ultrawork. Done.项目地址: https://gitcode.com/gh_mirrors/oh/oh-my-openagent本文基于仓库中的变更规格文档 code-changes.md完整讲解如何为 oh-my-opencode 的background_task配置新增全局并发上限maxBackgroundAgents从 Zod Schema 字段定义、校验测试、ConcurrencyManager全局计数器到BackgroundManager在任务启动、跟踪、完成、取消、报错全链路上的槽位获取与释放。读完本文你能掌握该项目的后台代理并发控制机制按模型/Provider 分层的 lane 限制 队列交接模式并理解一个全局资源上限类配置项从声明到运行时强制的完整设计路径。背景现有的按模型并发限制缺少全局兜底要理解这次变更的动机先看当前仓库中已经存在的并发控制实现。oh-my-opencode 的后台代理系统packages/omo-opencode包下的 background-agent 特性通过ConcurrencyManager限制每个模型/Provider同时运行的任务数。当前实现见 concurrency.tsgetConcurrencyLimit(model: string): number { const modelLimit this.config?.modelConcurrency?.[model] if (modelLimit ! undefined) { return modelLimit 0 ? Infinity : modelLimit } const provider model.split(/)[0] const providerLimit this.config?.providerConcurrency?.[provider] if (providerLimit ! undefined) { return providerLimit 0 ? Infinity : providerLimit } const defaultLimit this.config?.defaultConcurrency if (defaultLimit ! undefined) { return defaultLimit 0 ? Infinity : defaultLimit } return 5 }从源码结构看这是一套三级回退的 lane 机制modelConcurrency精确到provider/model全名→providerConcurrency按provider/model前缀切出的 provider 名→defaultConcurrency全局默认→ 硬编码兜底值 5。配置中把某一级设为0表示不限返回Infinityconcurrency.ts。acquire()的语义是限流即排队当某个 key 的计数达到上限时请求不会失败而是被压入Mapkey, QueueEntry[]等待队列concurrency.ts。release()释放槽位时优先把空位交接给队列中的等待者计数不变没有等待者才真正递减计数concurrency.ts。队列项使用settled标志防止已被 release 解决的条目再被cancelWaiters()重复拒绝这是典型的 double-resolution 防护模式concurrency.ts。这套机制的盲区正是本次变更要解决的问题所有限制都是按 lane的。一个用户在anthropic/claude-opus-4-6、openai/gpt-5、google/gemini三个 lane 各开 5 个任务时每个 lane 都没超限但整机上已经跑了 15 个后台代理——系统资源可能已经耗尽。官方文档 omo-json.md 中关于 senpi 侧global_concurrency的描述也印证了这一点OpenCodebackground_taskis unaffected (parity is a follow-up)——即 OpenCode 侧的全局上限尚属待补齐项。maxBackgroundAgents就是补齐这块的单一旋钮无论任务跑在哪个模型上同时运行的后台代理总数不得超过该值。变更一Zod Schema 新增 maxBackgroundAgents 字段规格文档针对的源文件路径写作src/config/schema/background-task.ts对应当前仓库中的 background-task.ts。规格中给出的完整 schema 快照如下包含新增的maxBackgroundAgents字段及其 JSDoc 注释import { z } from zod export const BackgroundTaskConfigSchema z.object({ defaultConcurrency: z.number().min(1).optional(), providerConcurrency: z.record(z.string(), z.number().min(0)).optional(), modelConcurrency: z.record(z.string(), z.number().min(0)).optional(), maxDepth: z.number().int().min(1).optional(), maxDescendants: z.number().int().min(1).optional(), /** Maximum number of background agents that can run simultaneously across all models/providers (default: 5, minimum: 1) */ maxBackgroundAgents: z.number().int().int().min(1).optional(), /** Stale timeout in milliseconds - interrupt tasks with no activity for this duration (default: 180000 3 minutes, minimum: 60000 1 minute) */ staleTimeoutMs: z.number().min(60000).optional(), /** Timeout for tasks that never received any progress update, falling back to startedAt (default: 1800000 30 minutes, minimum: 60000 1 minute) */ messageStalenessTimeoutMs: z.number().min(60000).optional(), syncPollTimeoutMs: z.number().min(60000).optional(), }) export type BackgroundTaskConfig z.infertypeof BackgroundTaskConfigSchema说明上述是规格文档中的 schema 快照用于展示新增字段的写法当前仓库 background-task.ts 实际还包含taskTtlMs、sessionGoneTimeoutMs、maxToolCalls、circuitBreaker等后续演进出来的字段且并发子字段名如maxDescendants对应当前的maxLiveDescendantsPerRoot已调整。本文以规格文档为准讲解maxBackgroundAgents的引入方式。三个设计要点z.number().int().min(1).optional()—— 完全复用maxDepth、maxDescendants的既有模式整数、下限 1、可选。可选意味着配置未提供时不报错运行时默认值 5 由ConcurrencyManager兜底见下文getMaxBackgroundAgents()。JSDoc 注释即文档—— 注释写明 default: 5, minimum: 1配置语义不依赖外部文档即可自解释。无需改动 barrel 导出—— 规格文档指出src/config/schema.ts已经export * from ./schema/background-task类型由z.infer自动推导加字段零侵入。变更二Schema 校验测试规格要求在 background-task.test.ts 中已有的syncPollTimeoutMsdescribe 块之后追加maxBackgroundAgents测试块覆盖合法值、下边界、低于下限、未提供、非整数五类场景describe(maxBackgroundAgents, () { describe(#given valid maxBackgroundAgents (10), () { test(#when parsed #then returns correct value, () { const result BackgroundTaskConfigSchema.parse({ maxBackgroundAgents: 10 }) expect(result.maxBackgroundAgents).toBe(10) }) }) describe(#given maxBackgroundAgents of 1 (minimum), () { test(#when parsed #then returns correct value, () { const result BackgroundTaskConfigSchema.parse({ maxBackgroundAgents: 1 }) expect(result.maxBackgroundAgents).toBe(1) }) }) describe(#given maxBackgroundAgents below minimum (0), () { test(#when parsed #then throws ZodError, () { let thrownError: unknown try { BackgroundTaskConfigSchema.parse({ maxBackgroundAgents: 0 }) } catch (error) { thrownError error } expect(thrownError).toBeInstanceOf(ZodError) }) }) describe(#given maxBackgroundAgents not provided, () { test(#when parsed #then field is undefined, () { const result BackgroundTaskConfigSchema.parse({}) expect(result.maxBackgroundAgents).toBeUndefined() }) }) describe(#given maxBackgroundAgents is non-integer (2.5), () { test(#when parsed #then throws ZodError, () { let thrownError: unknown try { BackgroundTaskConfigSchema.parse({ maxBackgroundAgents: 2.5 }) } catch (error) { thrownError error } expect(thrownError).toBeInstanceOf(ZodError) }) }) })测试命名沿用该文件既有的#given/#when/#then嵌套 describe 风格与maxDepth、maxDescendants、syncPollTimeoutMs的测试完全一致。其中两个细节值得注意0被判为非法与并发子字段不同0表示不限全局上限0没有合理语义——永远不许运行后台代理的场景应由用户干脆不启动代理来表达因此min(1)直接拒绝2.5触发 ZodError.int()约束保证上限是整数避免1.5 个代理这类歧义状态。变更三ConcurrencyManager 增加全局计数器核心实现改动集中在 concurrency.ts。规格文档给出的目标形态如下...处为未改动的既有方法当前仓库中已存在见 concurrency.tsimport type { BackgroundTaskConfig } from ../../config/schema const DEFAULT_MAX_BACKGROUND_AGENTS 5 /** * Queue entry with settled-flag pattern to prevent double-resolution. * * The settled flag ensures that cancelWaiters() doesnt reject * an entry that was already resolved by release(). */ interface QueueEntry { resolve: () void rawReject: (error: Error) void settled: boolean } export class ConcurrencyManager { private config?: BackgroundTaskConfig private counts: Mapstring, number new Map() private queues: Mapstring, QueueEntry[] new Map() private globalRunningCount 0 constructor(config?: BackgroundTaskConfig) { this.config config } getMaxBackgroundAgents(): number { return this.config?.maxBackgroundAgents ?? DEFAULT_MAX_BACKGROUND_AGENTS } getGlobalRunningCount(): number { return this.globalRunningCount } canSpawnGlobally(): boolean { return this.globalRunningCount this.getMaxBackgroundAgents() } acquireGlobal(): void { this.globalRunningCount } releaseGlobal(): void { if (this.globalRunningCount 0) { this.globalRunningCount-- } } getConcurrencyLimit(model: string): number { // ... existing implementation unchanged ... } async acquire(model: string): Promisevoid { // ... existing implementation unchanged ... } release(model: string): void { // ... existing implementation unchanged ... } cancelWaiters(model: string): void { // ... existing implementation unchanged ... } clear(): void { for (const [model] of this.queues) { this.cancelWaiters(model) } this.counts.clear() this.queues.clear() this.globalRunningCount 0 } getCount(model: string): number { return this.counts.get(model) ?? 0 } getQueueLength(model: string): number { return this.queues.get(model)?.length ?? 0 } }关键变化逐条拆解DEFAULT_MAX_BACKGROUND_AGENTS 5常量未配置时的全局上限与getConcurrencyLimit()的硬编码兜底 5 保持同一量级行为可预期globalRunningCount私有字段一个跨所有 lane 的单一计数器与 per-model 的counts: Map正交——两者可以同时生效实际并发受全局上限和所在 lane 上限中更紧的一方约束canSpawnGlobally()/acquireGlobal()/releaseGlobal()全局槽位的查/取/还三件套。与 per-model 的acquire()限流即排队不同全局槽位是立即失败语义——launch()里发现canSpawnGlobally()为 false 时直接抛错见下文不做排队等待。这个设计差异是合理的per-model 排队是为了让同 lane 任务公平轮转而全局资源耗尽时新任务再等下去只会堆积明确报错让用户等已有任务完成或调大配置才是更快的反馈releaseGlobal()的 0保护多调一次release不会让计数变成负数容忍竞态下的重复释放clear()重置全局计数manager 清理/关闭时全局计数与 per-model 状态一起归零防止幽灵槽位永久占用全局上限。变更四全局上限单元测试规格要求在 concurrency.test.ts 中追加独立的describe块用 given/when/then 注释风格覆盖默认值、配置值、上下限行为与重置describe(ConcurrencyManager global background agent limit, () { test(should default max background agents to 5 when no config, () { // given const manager new ConcurrencyManager() // when const max manager.getMaxBackgroundAgents() // then expect(max).toBe(5) }) test(should use configured maxBackgroundAgents, () { // given const config: BackgroundTaskConfig { maxBackgroundAgents: 10 } const manager new ConcurrencyManager(config) // when const max manager.getMaxBackgroundAgents() // then expect(max).toBe(10) }) test(should allow spawning when under global limit, () { // given const config: BackgroundTaskConfig { maxBackgroundAgents: 2 } const manager new ConcurrencyManager(config) // when manager.acquireGlobal() // then expect(manager.canSpawnGlobally()).toBe(true) expect(manager.getGlobalRunningCount()).toBe(1) }) test(should block spawning when at global limit, () { // given const config: BackgroundTaskConfig { maxBackgroundAgents: 2 } const manager new ConcurrencyManager(config) // when manager.acquireGlobal() manager.acquireGlobal() // then expect(manager.canSpawnGlobally()).toBe(false) expect(manager.getGlobalRunningCount()).toBe(2) }) test(should allow spawning again after release, () { // given const config: BackgroundTaskConfig { maxBackgroundAgents: 1 } const manager new ConcurrencyManager(config) manager.acquireGlobal() // when manager.releaseGlobal() // then expect(manager.canSpawnGlobally()).toBe(true) expect(manager.getGlobalRunningCount()).toBe(0) }) test(should not go below zero on extra release, () { // given const manager new ConcurrencyManager() // when manager.releaseGlobal() // then expect(manager.getGlobalRunningCount()).toBe(0) }) test(should reset global count on clear, () { // given const config: BackgroundTaskConfig { maxBackgroundAgents: 5 } const manager new ConcurrencyManager(config) manager.acquireGlobal() manager.acquireGlobal() manager.acquireGlobal() // when manager.clear() // then expect(manager.getGlobalRunningCount()).toBe(0) }) })七个用例合起来恰好锁死了全局计数的全部状态转换默认 5 → 配置 10 → 未达上限可 spawn → 达到上限被阻塞 → 释放后可再 spawn → 多余 release 不越界 →clear()归零。变更五BackgroundManager 全链路上的强制与释放计数器本身不会自动生效真正的难点在 manager.ts 中把所有任务生命周期事件与全局槽位的获取/释放正确配对。规格文档覆盖了两处入口检查和四处释放点。launch()启动前做全局检查创建后取槽async launch(input: LaunchInput): PromiseBackgroundTask { // ... existing logging ... if (!input.agent || input.agent.trim() ) { throw new Error(Agent parameter is required) } // Check global background agent limit before spawn guard if (!this.concurrencyManager.canSpawnGlobally()) { const max this.concurrencyManager.getMaxBackgroundAgents() const current this.concurrencyManager.getGlobalRunningCount() throw new Error( Background agent spawn blocked: ${current} agents running, max is ${max}. Wait for existing tasks to complete or increase background_task.maxBackgroundAgents. ) } const spawnReservation await this.reserveSubagentSpawn(input.parentSessionID) try { // ... existing code ... // After task creation, before queueing: this.concurrencyManager.acquireGlobal() // ... rest of existing code ... } catch (error) { spawnReservation.rollback() throw error } }三个设计细节报错信息自带运维指引错误消息包含当前运行数、上限值并直接提示等待现有任务完成或增大background_task.maxBackgroundAgents用户无需查文档即可自救检查放在reserveSubagentSpawn之前全局上限被拒时不产生任何 spawn 预留避免无谓占用父会话的子代理名额仓库中该预留机制见 subagent-spawn-limits.tsacquireGlobal()在任务创建成功后、入队前调用且catch分支回滚 spawn 预留——确保创建失败的任务不占全局槽位。trackTask()外部任务注册同样受全局上限约束async trackTask(input: { ... }): PromiseBackgroundTask { const existingTask this.tasks.get(input.taskId) if (existingTask) { // ... existing re-registration logic unchanged ... return existingTask } // Check global limit for new external tasks if (!this.concurrencyManager.canSpawnGlobally()) { const max this.concurrencyManager.getMaxBackgroundAgents() const current this.concurrencyManager.getGlobalRunningCount() throw new Error( Background agent spawn blocked: ${current} agents running, max is ${max}. Wait for existing tasks to complete or increase background_task.maxBackgroundAgents. ) } // ... existing task creation ... this.concurrencyManager.acquireGlobal() // ... rest unchanged ... }trackTask()是外部来源任务非launch()直接创建的登记入口。已存在的 taskId 走幂等的重注册逻辑直接返回不重复取槽新任务则与launch()完全一致地受全局上限约束。两个入口共用同一检查逻辑保证了不管任务从哪条路径进入管理器全局计数口径一致。tryCompleteTask()完成路径释放槽位private async tryCompleteTask(task: BackgroundTask, source: string): Promiseboolean { if (task.status ! running) { // ... existing guard ... return false } task.status completed task.completedAt new Date() // ... existing history record ... removeTaskToastTracking(task.id) // Release per-model concurrency if (task.concurrencyKey) { this.concurrencyManager.release(task.concurrencyKey) task.concurrencyKey undefined } // Release global slot this.concurrencyManager.releaseGlobal() // ... rest unchanged ... }注意释放的顺序与配对先释放 per-model 的concurrencyKey槽位释放即可能把 lane 空位交接给等待者再释放全局槽位task.concurrencyKey置undefined防止后续路径重复释放同一 lane 槽位。cancelTask()取消路径释放槽位pending 任务除外async cancelTask(taskId: string, options?: { ... }): Promiseboolean { // ... existing code up to concurrency release ... if (task.concurrencyKey) { this.concurrencyManager.release(task.concurrencyKey) task.concurrencyKey undefined } // Release global slot (only for running tasks, pending never acquired) if (task.status ! pending) { this.concurrencyManager.releaseGlobal() } // ... rest unchanged ... }这里有一个关键的不对称处理pending状态的任务不释放全局槽位因为尚未进入 per-model 队列、真正开始运行的任务从未执行过acquireGlobal()——对 pending 任务调用释放会造成计数漂移。这个只有 running 任务才持有全局槽位的约定是整条释放链正确性的基石。session.error 与 prompt 错误路径异常也要还槽任务因会话级错误终止时if (event.type session.error) { // ... existing error handling ... task.status error // ... if (task.concurrencyKey) { this.concurrencyManager.release(task.concurrencyKey) task.concurrencyKey undefined } // Release global slot this.concurrencyManager.releaseGlobal() // ... rest unchanged ... }startTask()内部 prompt 请求失败含模型建议重试的 catch 分支同样处理promptWithModelSuggestionRetry(this.client, { ... }).catch((error) { // ... existing error handling ... if (existingTask) { existingTask.status interrupt // ... if (existingTask.concurrencyKey) { this.concurrencyManager.release(existingTask.concurrencyKey) existingTask.concurrencyKey undefined } // Release global slot this.concurrencyManager.releaseGlobal() // ... rest unchanged ... } })如果漏掉这两个异常路径一次 provider 报错就会让全局槽位永久泄漏跑满上限后所有新任务都会被Background agent spawn blocked拒绝且永不恢复——这正是规格文档在配套 PR 描述中强调的Release global slots on task completion, cancellation, error, and interrupt to prevent slot leaks见 pr-description.md。变更汇总与验证方式规格文档给出的最终统计文件新增行数修改行数src/config/schema/background-task.ts20src/config/schema/background-task.test.ts~500src/features/background-agent/concurrency.ts~251clear()src/features/background-agent/concurrency.test.ts~700src/features/background-agent/manager.ts~200总计约 167 行新增、1 行修改横跨 5 个文件路径对应仓库中packages/omo-opencode/前缀下的同名文件。验证命令按配套 PR 描述执行bun test src/config/schema/background-task.test.ts # schema 校验 bun test src/features/background-agent/concurrency.test.ts # 全局上限 bun run typecheck bun run build用户侧的配置用法对应.opencode/oh-my-opencode.jsonc{ background_task: { maxBackgroundAgents: 10 // default: 5, min: 1 } }小结一个全局上限配置项的正确姿势这套变更示范了在 oh-my-opencode 这类插件架构里新增资源类配置的完整方法论Schema 层z.number().int().min(1).optional()可选字段 明确下限默认值不在 schema 硬编码留给运行时运行时层在ConcurrencyManager中引入与 per-model 计数正交的globalRunningCount提供查/取/还三接口且release带越界保护、clear()随 manager 生命周期归零强制层入口launch()/trackTask()统一检查并立即失败报错信息含自救指引出口完成、取消、会话错误、prompt 失败四条释放路径与acquireGlobal()严格配对且明确pending 任务不持有全局槽位的不变量测试层schema 五类边界用例 计数器七个状态转换用例两层测试独立可回归。需要说明的是规格文档位于.agents/skills/work-with-pr-workspace/iteration-1/eval-1/with_skill/outputs/目录属于 PR 工作流工作区中产出的变更设计当前仓库主干代码background-task.ts、concurrency.ts尚未包含maxBackgroundAgents字段本文将其作为该功能的设计规格来解读实际合入状态请以仓库当前代码为准。【免费下载链接】oh-my-openagentOmO: Drop your tokens. Ultrawork. Done.项目地址: https://gitcode.com/gh_mirrors/oh/oh-my-openagent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考