Onyx Craft 定时任务Scheduled Tasks架构解析从 Cron 调度到无头 Agent 执行的完整实现指南【免费下载链接】danswerOpen Source AI Platform - AI Chat with advanced features that works with every LLM项目地址: https://gitcode.com/GitHub_Trending/da/danswer本文以 docs/craft/features/scheduled-tasks/overview.md 为核心结合 Onyx原 danswer仓库中该功能的完整源码实现系统讲解 Craft 定时任务的产品目标、核心设计决策、Celery 调度架构、数据模型、REST API 规范、Cron 调度语义、运行生命周期与测试策略。读完本文你将掌握保存 Prompt 定时执行这一产品形态在 Onyx 中的落地方案并能在本地部署、调用 API、阅读源码和编写测试时做到心中有数。一、产品目标与 V1 范围Scheduled Tasks定时任务是 Onyx Craft 的一个产品面用户保存一段 Prompt 与一个调度计划系统按定时器把这段 Prompt 作为 Craft 任务无头headless执行。每一次触发都会创建一个全新的 Craft 会话BuildSession无头执行完整条 Agent 流程并把执行过程完整记录下来。V1 的验收标准来自 overview.md用户能在/craft/v1/tasks创建任务任务能在浏览器关闭后照常触发执行用户点击任意一次历史运行记录即可打开那次执行产生的完整会话视图。V1 明确不做以下能力实时挂接live-attach、事件触发、共享任务、模板、重试策略、预算上限budget caps、运行对比run diffing、日历视图、连续失败自动禁用、外部任务管理 API。二、核心设计决策理解该功能的关键overview 文档沉淀了该功能最重要的十余条设计决策它们是理解整个实现的钥匙一次运行run就是一个BuildSession。scheduled_task_run.session_id是对build_session.id的外键。点击已完成运行的记录打开的就是现有的会话视图无需新建 transcript UI。每次触发都创建全新会话复用现有的SessionManager.create_session__no_commit路径因此沙箱预置、工作区初始化、技能物化、AGENTS.md 生成、数据包packet日志等流程原样复用。无头执行器复用send_message的持久化半段把_stream_cli_agent_response拆成_yield_acp_events纯 ACP 事件生成器与_persist_acp_eventsBuildStreamingState消费者写入BuildMessage行两部分。SSE 端点把两者与 SSE 格式化器组合执行器则用 drain-to-completion 方式包裹。这样转录transcript与交互式完全一致且没有重复代码。专用scheduled_tasksCelery worker——只跑执行器。长时执行器run_scheduled_task运行在新的celery_worker_scheduled_tasks进程上已在 supervisord、dev runner 与 Helm chart 中注册。因为无头 Agent 触发是长时操作LLM 沙箱内工具调用若与heavy队列pruning、权限同步、CSV 导出共用少量触发就可能饿死 heavy 队列的其他任务。专用 worker 意味着独立的线程池、独立的 HPA/KEDA 扩缩容以及独立的 Prometheus 端口9098。而派发器dispatcher与卡死运行清扫器stuck-run sweeper是纯 DB 协调工作运行在 primary 队列——若把派发逻辑也路由到专用池执行器饱和时会反过来拖住派发。调度存储(cron_expression, editor_mode)二元组。三种编辑器模式保存时统一编译为 cronCron 表达式按 UTC 求值editor_mode仅是 UI 提示。next_run_at在每次触发与每次编辑时重算暂停时置为 NULL。并发策略周期性触发用SKIP_IF_RUNNING上一次运行仍在进行 → 写入skipped行并携带skip_reason同时仍推进next_run_atRun Now 用QUEUE_ONE暂停状态下也能工作且不触碰next_run_at。运行以任务作者身份执行create_session__no_commit(user_idtask.user_id)——技能、Onyx 搜索、OAuth 授权、审批策略全部走与交互式 UI 相同的用户作用域路径。软删除保留历史deletedtrue停止派发运行记录与会话保留用户仍可从任务运行历史中打开过去的运行。BuildSession.origin把定时运行排除出侧边栏新增枚举列INTERACTIVE | SCHEDULED默认INTERACTIVE存量行服务端默认interactive由执行器在创建会话时设置侧边栏查询过滤origin INTERACTIVE。定时触发与 Run Now都走执行器、都带originSCHEDULED一并覆盖。之所以用列而不是对scheduled_task_run做NOT EXISTS反连接是因为派发器先于执行器写 run 行——基于 join 的过滤会短暂泄漏。未来非交互式 origineval 运行、自动化可复用同一接缝。V1 无重试失败就是一行记录用户点击 Run Now 或等待下一次触发。通知复用现有Notification模型新增两种类型SCHEDULED_TASK_FAILED、SCHEDULED_TASK_AWAITING_APPROVAL。V1 不发邮件、不发 Slack。卡死运行清扫器每小时queued 15 分钟与running budget→ 标记为failed (stuck)兜底捕获死亡 worker。三、调度架构Beat 双队列 专用 Workeroverview 用一张 ASCII 架构图说明了数据流结合 tasks.py 与 beat_schedule.py 的源码可以还原出精确的执行时序Beat (30s, per tenant) Celery scheduled_tasks queue Primary queue (served by celery_worker_scheduled_tasks) ────────────────────── ──────────────────────────────── dispatch_due_scheduled_tasks run_scheduled_task(run_id) BEGIN; if run.status ! queued: return SELECT FROM scheduled_task mark running WHERE active AND due session SessionManager FOR UPDATE SKIP LOCKED; .create_session__no_commit( for each row: user_idtask.user_id) ├─ if prior run in flight run.session_id ← session.id │ → insert skipped row for event in _yield_acp_events( ├─ insert queued run session, task.prompt): ├─ next_run_at croniter _persist_acp_events([event]) │ .next(now) if budget exceeded → failed └─ enqueue run_scheduled_task ───► if approval required → (run_id, expires900, awaiting_approval queuescheduled_tasks) mark succeeded / failed COMMIT; emit Notification if failed Stuck-run sweep (hourly, primary queue) ────────────────────── │ cleanup_stuck_scheduled_runs ▼ queued 15m → failed (stuck) BuildMessage rows (existing tables, running budget → failed (timeout) written by shared persist consumer)3.1 派发器dispatcher每租户每 30 秒dispatch_due_scheduled_tasks注册在 beat_schedule.py 中scheduletimedelta(seconds30)、队列为 primary、优先级 MEDIUM、expires60。注释说明 30 秒是规格契约60 秒对分钟级 cron 太粗15 秒会让没有到期任务的租户过度占用FOR UPDATE SKIP LOCKED路径。派发逻辑见 tasks.py调用claim_due_scheduled_tasks用FOR UPDATE SKIP LOCKED原子认领到期任务批量上限DISPATCH_BATCH_SIZE 50。对每个认领的任务按优先级判定任务所有者 Craft 被禁用部署级开关、用户级覆盖或工作区默认→ 插入SKIPPED行skip_reasonowner_craft_disabled调度保持存活重新启用后恢复存在进行中的运行has_in_flight_run_for_task检查 QUEUED/RUNNING→ 插入SKIPPED行skip_reasonprior_in_flight否则插入QUEUEDrun 行并收集 run_id 待入队。无论命中哪条分支都推进next_run_atadvance_next_run_at否则下一次 tick 会重复认领同一行。全部在同一事务内提交后再对每个 run_id 发送SCHEDULED_TASKS_RUN到scheduled_tasks队列expiresQUEUE_RESIDENCY_SECONDS。如果 commit 与入队之间崩溃卡死清扫器会在约 15 分钟后回收这些 QUEUED 行。claim_due_scheduled_tasks的实现细节值得注意db/scheduled_task.py查询条件为statusactive AND deletedfalse AND next_run_at IS NOT NULL AND next_run_at now按next_run_at升序取批with_for_update(skip_lockedTrue)保证并发 tick 不会双触发。3.2 执行器executor专用 workerrun_scheduled_task是薄包装tasks.pyacks_lateFalseworker 崩溃不触发 Celery 重试V1 无重试实际逻辑全部在run_scheduled_task_logicexecutor.py拆出来是为了让 Agent 驱动逻辑无需 Celery worker 即可被外部依赖单元测试直接实例化。执行器状态机源码确认幂等守卫run 状态不是 QUEUED 就直接返回Celery 可能重投递或清扫器已先标记失败。沙箱就绪SessionManager.ensure_sandbox_running——没有沙箱则创建等待并发 provisioner上限PROVISION_WAIT_SECONDS 120s原地唤醒 SLEEPING/TERMINATED/FAILED 的沙箱。等待窗口内仍处于 PROVISIONING 则 SKIP其他失败标记error_classsandbox_wake_failed。过渡到 RUNNING 并提交让 UI 与清扫器可见。以originSCHEDULED创建全新BuildSession会话名Scheduled: {task_name}并把 session_id 写回 run 行。通过共享的yield_sandbox_events生成器驱动 Agent用persist_sandbox_event持久化每个事件强制执行预算默认硬上限SCHEDULED_RUN_HARD_CAP_SECONDS 60 * 6060 分钟见 timeouts.py软预算为其 40%24 分钟。注释特别说明 Celery 线程池会静默忽略soft_time_limit/time_limit所以预算必须在任务体内实现。命中RequestPermissionRequest审批门标记AWAITING_APPROVAL、发通知、不写终态返回。恢复机制归审批项目所有在那之前仅作展示用终态。正常流结束finalize_persist→ 提炼约 120 字符的摘要_clip_summary截断时按单词边界断句并加省略号→ 标记 SUCCEEDED。驱动循环内任何异常标记 FAILED带异常类名 详情并发通知。刻意吞掉异常防止 Celery 自动重试。摘要生成有两条路径优先取BuildStreamingState.message_chunks中最近流式文本若内存中无待刷 chunk如中途已 flush则回退扫描持久化消息倒序查找最后一个agent_message元数据 blob见_summary_from_session_messages。3.3 卡死运行清扫器stuck-run sweepercleanup_stuck_scheduled_runs每小时运行beat_schedule.py判定规则tasks.pyQUEUED 超过QUEUE_RESIDENCY_SECONDS 15 * 6015 分钟→failed (stuck)RUNNING 超过SCHEDULED_RUN_HARD_CAP_SECONDS TURN_RECLAIM_SLACK_SECONDS60 分钟 15 分钟→failed (stuck)。两者与 Celery 的expires构成同一策略的三处执行点见 timeouts.py 的注释队列驻留超限的消息被丢弃且行被同阈值回收正常跑满自身预算的运行一定会先把自己标记 FAILED清扫器只是兜底。3.4 专用 worker 进程apps/scheduled_tasks.py 定义了独立 Celery app配置来自onyx.background.celery.configs.scheduled_tasksautodiscover_tasks只加载onyx.background.celery.tasks.scheduled_tasks模块worker_ready时调用start_metrics_server(scheduled_tasks)即 Prometheus 端口 9098。worker 初始化沿用app_base的启动检查等 Redis、等 DB、等文档索引并设置 PostgreSQL 应用名POSTGRES_CELERY_WORKER_SCHEDULED_TASKS_APP_NAME。四、数据模型overview 给出了核心 ORM 模型的骨架与 db/enums.py 和 db/scheduled_task.py 的源码完全对应class SessionOrigin(str, Enum): # INTERACTIVE / SCHEDULED / SLACK # 加到 BuildSession默认 INTERACTIVE class ScheduledTaskStatus(str, Enum): # ACTIVE / PAUSED class ScheduledTaskRunStatus(str, Enum): # QUEUED / RUNNING / SUCCEEDED / # FAILED / SKIPPED / AWAITING_APPROVAL class ScheduledTaskTriggerSource(str, Enum): # SCHEDULED / MANUAL_RUN_NOW class ScheduledTask(Base): __tablename__ scheduled_task id, user_id (FK user, CASCADE) name (str), prompt (text) cron_expression (str), editor_mode (str) status (enum, default ACTIVE) next_run_at (DateTime tz, nullable) # dispatcher 唯一读取的字段 deleted (bool, default False) created_at, updated_at runs ← back-populated, cascade all,delete-orphan __table_args__ ( Index(ix_scheduled_task_dispatch, status, deleted, next_run_at), Index(ix_scheduled_task_user_created, user_id, desc(created_at)), ) class ScheduledTaskRun(Base): __tablename__ scheduled_task_run id, task_id (FK scheduled_task, CASCADE) session_id (FK build_session, SET NULL) # 执行器创建会话后回填 status (enum, default QUEUED), trigger_source (enum) skip_reason / error_class / error_detail (nullable) started_at (default now), finished_at (nullable) summary (str, ~120 chars of final agent message) __table_args__ ( Index(ix_scheduled_task_run_task_started, task_id, desc(started_at)), Index(ix_scheduled_task_run_status, status), )设计要点源码佐证next_run_at是派发器唯一读取的字段暂停置 NULLupdate_scheduled_task中statusPAUSED分支、编辑时重算、deletedtrue从认领查询中排除claim_due_scheduled_tasks的 WHERE 条件。session_id可空在派发器 INSERT 与执行器创建会话之间的短暂窗口内为 NULL。没有attempts计数V1 无重试。ScheduledTaskRunStatus.is_terminal()只把 SUCCEEDED/FAILED/SKIPPED 视为终态enums.pyAWAITING_APPROVAL不在其中。error_class是封闭枚举task_missing/sandbox_wake_failed/executor_error/timeout/stuck/agent_exception让仪表盘与排障查询可以按已知词表透视意外的运行时异常统一记agent_exception真实异常类名与消息放入error_detail。skip_reason也是封闭枚举prior_in_flight/owner_craft_disabled。派发器的认领查询用selectinload(ScheduledTask.user)预加载所有者以便在事务内做 Craft 启用判定。mark_run_status在写入终态时自动填充finished_atinsert_run对 SKIPPED 行也当场填finished_at避免 UI 特判。数据库操作全部集中在 db/scheduled_task.py按项目 CLAUDE.md 约定所有查询下沉到 db 层所有函数首个参数为db_session包括任务 CRUDcreate_scheduled_task/update_scheduled_task/soft_delete_scheduled_task所有权与 NOT_FOUND 抛出语义一致、派发热路径claim_due_scheduled_tasks/advance_next_run_at、run CRUDinsert_run/mark_run_status/list_runs_for_task、卡死扫描find_stuck_runs以及会话横幅辅助get_scheduled_run_context与审批预授权查询get_live_scheduled_run_grants。五、API 规范所有端点抛出OnyxError返回类型化 FastAPI 响应挂载在/api/build/scheduled-tasks复用现有/build前缀与require_onyx_craft_enabled门控作用域限定在认证用户V1 无管理员视图。路由实现见 api.py权限依赖为require_permission(Permission.BASIC_ACCESS)。方法路径说明GET/scheduled-tasks任务列表id、名称、人类可读调度、状态、next_run_at、最近一次运行摘要POST/scheduled-tasks创建。编辑器输入编译为 cron可选run_immediatelyGET/scheduled-tasks/{id}任务详情 未来 3 次触发时间UI 预览PATCH/scheduled-tasks/{id}部分编辑。调度变化时重算next_run_at暂停→NULL恢复→重算。进行中的运行不受影响DELETE/scheduled-tasks/{id}软删除幂等返回 204POST/scheduled-tasks/{id}/run-now插入manual_run_nowrun 并入队执行器暂停时可用不触碰next_run_atGET/scheduled-tasks/{id}/runs分页运行历史50/页cursor参数最新在前GET/build/sessions/{id}/scheduled-run-context若会话来自定时运行返回可选的任务名 id 定时started_at否则 404。供会话视图横幅使用V1 没有 live-attach 端点。请求体校验细节源码确认ScheduledTaskCreatename1–200 字符、prompt非空、editor_mode为 Literal 枚举、editor_payload是受类型约束的联合体、status默认 ACTIVE、run_immediately默认 false另有pre_approved_app_ids与pre_approved_mcp_server_ids预授权目标配合审批门使用。模型extraforbid未知字段直接 422。一个巧妙的校验管道_dispatch_editor_payload作为model_validator(modebefore)把原始editor_payloaddict 按editor_mode路由到对应的类型化模型IntervalPayload/DailyWeeklyPayload/AdvancedPayload不匹配的形状在 FastAPI 层表现为 PydanticValidationError→ 422。PATCH 用_editor_pair_consistency强制editor_mode与editor_payload必须成对出现。_validate_app_ids拒绝未知外部应用 id_validate_mcp_server_ids拒绝该用户在 Craft 中不可用的 MCP server id已授权的存量 grant 可保留。runs 列表的cursor是 ISO-8601 时间上一页最后一行started_atnext_cursor为空表示已到末页游标分页与ix_scheduled_task_run_task_started (task_id, started_at DESC)索引精确对齐。run_immediately与run-now都遵循先落库再入队原则run 行持久化提交后才_enqueue_executor避免 worker 跑到自己的输入前面。入队时附带tenant_id由TenantAwareTask在任务体内设置租户 contextvar否则会在默认 schema 上执行。六、调度语义Cron 编译与求值调度相关的纯函数集中在 schedule.py它自述为 overview 文档中 cron 语义的单一事实来源DB 存规范 5 字段 cron 字符串 editor_modeUI 提示三种编辑器模式保存时统一编译为同一 cron 形式compute_next_run_at返回 UTC 时间比较也在 UTC 进行。6.1 三种编辑器模式间隔IntervalN 分钟/小时/天*/N * * * */0 */N * * */M H */N * *日节奏时 M:H 是编辑器要求的具体时刻。每日/每周Daily/WeeklyM H * * weekdays如0 9 * * 1,3,5。高级Advanced用户手写用croniter校验。compute_next_run_atcroniter(cron, after).get_next(datetime)存 UTC、UTC 比较。6.2 源码级校验规则Pydantic 模型在 HTTP 边界完成了全部校验因此compile_to_cron是纯变换源码注释明确没有dict[str, Any]查找、没有临时字符串解析_HH_MM_RE ^([01]?\d|2[0-3]):([0-5]\d)$24 小时制HH:MM格式与 UI 的input typetime对齐。IntervalPayloadevery 1unit days时time_of_day必填model_validator 强制。DailyWeeklyPayloadweekdays是 0–6 的整数列表0周日cron 约定空列表 每天等价于*。AdvancedPayload必须恰好 5 个字段且croniter.is_valid通过。compile_to_cron的边界兜底分钟间隔every 59折叠为0 * * * **/60 * * * *非法小时间隔every 23折叠为0 0 * * *——与 UI 依赖的历史行为一致。读侧compute_next_run_at/next_n_fires/human_readable也有_validate_cron防御把非法表达式、非 5 字段、无未来触发如不可能表达式统一转成OnyxError(INVALID_INPUT)。human_readable用cron-descriptor的ExpressionDescriptor渲染人类可读描述失败时回退原始 cron 字符串。6.3 时间语义要点暂停中途触发进行中的运行照常完成next_run_at→ NULL恢复时从恢复时刻起重算不回火。过期任务只触发一次并向前推进标准 cron 行为。没有未来触发的 cron在 POST/PATCH 时直接拒绝。调度变更重算规则update_scheduled_taskcron 变更且任务为或变为ACTIVE → 从 now 重算变为 PAUSED → NULL变为 ACTIVE → 重算。各预授权目标字段遵循普通 patch 语义提供的替换该类目标省略的不动。七、运行生命周期queued ─► running ─► succeeded │ └─► failed (executor crash | budget | ACP error) └─► awaiting_approval ─► (approvals project resumes) ─► running ─► ... dispatcher also writes: skipped (prior run still in flight; next_run_at advances)补充说明源码确认AWAITING_APPROVAL的恢复机制归审批项目所有在该项目落地前仅作展示用终态。多个定时运行与交互式send_message路径可以并发执行在同一沙箱上——没有序列化租约。执行器仍会为每个 BuildSession 获取 prompt slot 锁session_manager.prompt_slot这是与交互式路径对称的防御性做法当前每个定时运行都有全新 BuildSession、锁不会竞争但未来若让定时与交互共享 BuildSession此锁自动提供保护。若锁未获取成功并发回合在飞run 被标记 FAILEDerror_classagent_exception。每次触发都会stamp_turn_deadline软预算 min(软预算, budget)硬上限 budget离开时clear_turn_deadlineslot 未丢失时避免后续回合继承过期的定时截止时间。超时与完成的竞态处理很精细先检查PromptResponse/Error再检查预算——截止时间恰好在 Agent 完成瞬间触发时不能把成功误标为超时预算检查先于持久化防止失控 Agent 无限增长转录。stop_reason cancelled的PromptResponse被视为失败而非成功opencode 中止 Agent如MessageAbortedError。八、UI 设计列表页/craft/v1/tasks名称、调度、状态、last_run相对时间 成功/失败图标、next_run、行操作Run now、Pause/Resume、Edit、Delete。空状态展示产品文档中的三个 starter prompts。编辑器/new、/:id/edit单一表单 三 Tab 分段控件选择调度模式Prompt 输入框复用 Craft 聊天输入产品需求 10未来 3 次运行预览由前端用cron-parser计算。详情页/:id头部名称、调度、状态开关、next_run、操作按钮 分页运行历史表格。行点击 → 打开会话视图succeeded/failedqueued/running/awaiting_approval/skipped不可点击、显示 tooltip。会话视图横幅当 scheduled-run-context 端点返回结果时在转录上方渲染This session was started by scheduled task X at Y. ← Back to task.。聊天输入保持可用——用户可以在定时运行的会话上发送后续消息。通知两个新的铃铛条目——Task X failed、Task X needs approval深链到 run 行。通知实现在执行器的_notify中标题Scheduled task {task_name} failed/needs approvaladditional_data携带task_id/run_id且刻意 best-effort——通知失败绝不掩盖底层 run 状态更新。移动端列表/详情可容忍编辑器明确要求桌面端产品需求 8。九、测试策略详见 tests.md覆盖计划、既有测试套件布局与手工冒烟清单。E2E 冒烟web/tests/e2e/scheduled-tasks.spec.ts——单一 speccreate, run, and verify a run row exists登录标准 worker 用户 → 访问/craft/v1/tasks若重定向到/app则 Craft 功能开关关闭spec 软跳过→ 用唯一名称 promptsay hi 每 5 分钟间隔创建任务 → 点save-and-run-now→ 断言详情页 active 状态芯片 → 最多等 60 秒看到data-run-statussucceeded或failed的 run 行两者皆可证明 dispatcher → executor → 运行历史链路端到端可达。前端选择器已锁定new-task-button、task-name-input、task-prompt-input、interval-every、save-and-run-now、task-status-active、data-run-status。运行方式npx playwright test scheduled-tasks需要完整本地 Onyx 栈web、API、Postgres、Redis以及专用celery_worker_scheduled_tasksworker——缺了它 run 永远到不了终态测试在第 6 步超时。刻意不自动化依赖 code review 手工清单FOR UPDATE SKIP LOCKED派发并发、卡死清扫器、审批路径、侧边栏过滤、HTTP API 的用户所有权边界。手工冒烟清单合并前逐项过一遍每 2 分钟任务 Onyx 搜索 prompt离开 6 分钟回来确认三次运行都有完整会话与合理summary周一/三/五 9 AM验证next_run_at正确、UTC 9 AM 强制 tick 能触发中途暂停进行中运行完成、无新触发恢复 → 下次触发从now()向前计算不回火审批边界run 停在AWAITING_APPROVAL通知铃铛有条目同一沙箱上的交互式 Craft 仍可用中途杀 workerrun 处于 RUNNING 时停掉celery_worker_scheduled_tasks一小时内清扫器把它转为FAILED、error_classstuck。十、部署与运行要求调度功能由 Celery Beat 驱动30 秒派发 每小时清扫需要 Beat 进程在跑必须部署专用 worker 进程celery_worker_scheduled_taskssupervisord / dev runner / Helm chart 均已注册执行器只在其上运行多租户模式下派发与执行都依赖租户 contextvartenant_id随任务入队确保任务落到正确 schemaV1 无重试Celery 消息带expires死消费者不会无限堆积 backlog每个入队点都传 expires符合项目 CLAUDE.md 约定。十一、深入阅读的源码地图功能总设计docs/craft/features/scheduled-tasks/overview.md测试计划docs/craft/features/scheduled-tasks/tests.md数据库操作层backend/onyx/db/scheduled_task.py调度纯函数cron 编译/校验/求值backend/onyx/server/features/build/scheduled_tasks/schedule.pyREST 路由backend/onyx/server/features/build/scheduled_tasks/api.py无头执行器backend/onyx/server/features/build/scheduled_tasks/executor.pyCelery 任务派发/执行/清扫backend/onyx/background/celery/tasks/scheduled_tasks/tasks.pyBeat 调度注册backend/onyx/background/celery/tasks/beat_schedule.py专用 worker appbackend/onyx/background/celery/apps/scheduled_tasks.py枚举定义状态机/错误词表backend/onyx/db/enums.py超时注册表预算与清扫阈值backend/onyx/server/features/build/timeouts.pyE2E 冒烟测试web/tests/e2e/scheduled-tasks.spec.ts【免费下载链接】danswerOpen Source AI Platform - AI Chat with advanced features that works with every LLM项目地址: https://gitcode.com/GitHub_Trending/da/danswer创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Onyx Craft 定时任务（Scheduled Tasks）架构解析：从 Cron 调度到无头 Agent 执行的完整实现指南
Onyx Craft 定时任务Scheduled Tasks架构解析从 Cron 调度到无头 Agent 执行的完整实现指南【免费下载链接】danswerOpen Source AI Platform - AI Chat with advanced features that works with every LLM项目地址: https://gitcode.com/GitHub_Trending/da/danswer本文以 docs/craft/features/scheduled-tasks/overview.md 为核心结合 Onyx原 danswer仓库中该功能的完整源码实现系统讲解 Craft 定时任务的产品目标、核心设计决策、Celery 调度架构、数据模型、REST API 规范、Cron 调度语义、运行生命周期与测试策略。读完本文你将掌握保存 Prompt 定时执行这一产品形态在 Onyx 中的落地方案并能在本地部署、调用 API、阅读源码和编写测试时做到心中有数。一、产品目标与 V1 范围Scheduled Tasks定时任务是 Onyx Craft 的一个产品面用户保存一段 Prompt 与一个调度计划系统按定时器把这段 Prompt 作为 Craft 任务无头headless执行。每一次触发都会创建一个全新的 Craft 会话BuildSession无头执行完整条 Agent 流程并把执行过程完整记录下来。V1 的验收标准来自 overview.md用户能在/craft/v1/tasks创建任务任务能在浏览器关闭后照常触发执行用户点击任意一次历史运行记录即可打开那次执行产生的完整会话视图。V1 明确不做以下能力实时挂接live-attach、事件触发、共享任务、模板、重试策略、预算上限budget caps、运行对比run diffing、日历视图、连续失败自动禁用、外部任务管理 API。二、核心设计决策理解该功能的关键overview 文档沉淀了该功能最重要的十余条设计决策它们是理解整个实现的钥匙一次运行run就是一个BuildSession。scheduled_task_run.session_id是对build_session.id的外键。点击已完成运行的记录打开的就是现有的会话视图无需新建 transcript UI。每次触发都创建全新会话复用现有的SessionManager.create_session__no_commit路径因此沙箱预置、工作区初始化、技能物化、AGENTS.md 生成、数据包packet日志等流程原样复用。无头执行器复用send_message的持久化半段把_stream_cli_agent_response拆成_yield_acp_events纯 ACP 事件生成器与_persist_acp_eventsBuildStreamingState消费者写入BuildMessage行两部分。SSE 端点把两者与 SSE 格式化器组合执行器则用 drain-to-completion 方式包裹。这样转录transcript与交互式完全一致且没有重复代码。专用scheduled_tasksCelery worker——只跑执行器。长时执行器run_scheduled_task运行在新的celery_worker_scheduled_tasks进程上已在 supervisord、dev runner 与 Helm chart 中注册。因为无头 Agent 触发是长时操作LLM 沙箱内工具调用若与heavy队列pruning、权限同步、CSV 导出共用少量触发就可能饿死 heavy 队列的其他任务。专用 worker 意味着独立的线程池、独立的 HPA/KEDA 扩缩容以及独立的 Prometheus 端口9098。而派发器dispatcher与卡死运行清扫器stuck-run sweeper是纯 DB 协调工作运行在 primary 队列——若把派发逻辑也路由到专用池执行器饱和时会反过来拖住派发。调度存储(cron_expression, editor_mode)二元组。三种编辑器模式保存时统一编译为 cronCron 表达式按 UTC 求值editor_mode仅是 UI 提示。next_run_at在每次触发与每次编辑时重算暂停时置为 NULL。并发策略周期性触发用SKIP_IF_RUNNING上一次运行仍在进行 → 写入skipped行并携带skip_reason同时仍推进next_run_atRun Now 用QUEUE_ONE暂停状态下也能工作且不触碰next_run_at。运行以任务作者身份执行create_session__no_commit(user_idtask.user_id)——技能、Onyx 搜索、OAuth 授权、审批策略全部走与交互式 UI 相同的用户作用域路径。软删除保留历史deletedtrue停止派发运行记录与会话保留用户仍可从任务运行历史中打开过去的运行。BuildSession.origin把定时运行排除出侧边栏新增枚举列INTERACTIVE | SCHEDULED默认INTERACTIVE存量行服务端默认interactive由执行器在创建会话时设置侧边栏查询过滤origin INTERACTIVE。定时触发与 Run Now都走执行器、都带originSCHEDULED一并覆盖。之所以用列而不是对scheduled_task_run做NOT EXISTS反连接是因为派发器先于执行器写 run 行——基于 join 的过滤会短暂泄漏。未来非交互式 origineval 运行、自动化可复用同一接缝。V1 无重试失败就是一行记录用户点击 Run Now 或等待下一次触发。通知复用现有Notification模型新增两种类型SCHEDULED_TASK_FAILED、SCHEDULED_TASK_AWAITING_APPROVAL。V1 不发邮件、不发 Slack。卡死运行清扫器每小时queued 15 分钟与running budget→ 标记为failed (stuck)兜底捕获死亡 worker。三、调度架构Beat 双队列 专用 Workeroverview 用一张 ASCII 架构图说明了数据流结合 tasks.py 与 beat_schedule.py 的源码可以还原出精确的执行时序Beat (30s, per tenant) Celery scheduled_tasks queue Primary queue (served by celery_worker_scheduled_tasks) ────────────────────── ──────────────────────────────── dispatch_due_scheduled_tasks run_scheduled_task(run_id) BEGIN; if run.status ! queued: return SELECT FROM scheduled_task mark running WHERE active AND due session SessionManager FOR UPDATE SKIP LOCKED; .create_session__no_commit( for each row: user_idtask.user_id) ├─ if prior run in flight run.session_id ← session.id │ → insert skipped row for event in _yield_acp_events( ├─ insert queued run session, task.prompt): ├─ next_run_at croniter _persist_acp_events([event]) │ .next(now) if budget exceeded → failed └─ enqueue run_scheduled_task ───► if approval required → (run_id, expires900, awaiting_approval queuescheduled_tasks) mark succeeded / failed COMMIT; emit Notification if failed Stuck-run sweep (hourly, primary queue) ────────────────────── │ cleanup_stuck_scheduled_runs ▼ queued 15m → failed (stuck) BuildMessage rows (existing tables, running budget → failed (timeout) written by shared persist consumer)3.1 派发器dispatcher每租户每 30 秒dispatch_due_scheduled_tasks注册在 beat_schedule.py 中scheduletimedelta(seconds30)、队列为 primary、优先级 MEDIUM、expires60。注释说明 30 秒是规格契约60 秒对分钟级 cron 太粗15 秒会让没有到期任务的租户过度占用FOR UPDATE SKIP LOCKED路径。派发逻辑见 tasks.py调用claim_due_scheduled_tasks用FOR UPDATE SKIP LOCKED原子认领到期任务批量上限DISPATCH_BATCH_SIZE 50。对每个认领的任务按优先级判定任务所有者 Craft 被禁用部署级开关、用户级覆盖或工作区默认→ 插入SKIPPED行skip_reasonowner_craft_disabled调度保持存活重新启用后恢复存在进行中的运行has_in_flight_run_for_task检查 QUEUED/RUNNING→ 插入SKIPPED行skip_reasonprior_in_flight否则插入QUEUEDrun 行并收集 run_id 待入队。无论命中哪条分支都推进next_run_atadvance_next_run_at否则下一次 tick 会重复认领同一行。全部在同一事务内提交后再对每个 run_id 发送SCHEDULED_TASKS_RUN到scheduled_tasks队列expiresQUEUE_RESIDENCY_SECONDS。如果 commit 与入队之间崩溃卡死清扫器会在约 15 分钟后回收这些 QUEUED 行。claim_due_scheduled_tasks的实现细节值得注意db/scheduled_task.py查询条件为statusactive AND deletedfalse AND next_run_at IS NOT NULL AND next_run_at now按next_run_at升序取批with_for_update(skip_lockedTrue)保证并发 tick 不会双触发。3.2 执行器executor专用 workerrun_scheduled_task是薄包装tasks.pyacks_lateFalseworker 崩溃不触发 Celery 重试V1 无重试实际逻辑全部在run_scheduled_task_logicexecutor.py拆出来是为了让 Agent 驱动逻辑无需 Celery worker 即可被外部依赖单元测试直接实例化。执行器状态机源码确认幂等守卫run 状态不是 QUEUED 就直接返回Celery 可能重投递或清扫器已先标记失败。沙箱就绪SessionManager.ensure_sandbox_running——没有沙箱则创建等待并发 provisioner上限PROVISION_WAIT_SECONDS 120s原地唤醒 SLEEPING/TERMINATED/FAILED 的沙箱。等待窗口内仍处于 PROVISIONING 则 SKIP其他失败标记error_classsandbox_wake_failed。过渡到 RUNNING 并提交让 UI 与清扫器可见。以originSCHEDULED创建全新BuildSession会话名Scheduled: {task_name}并把 session_id 写回 run 行。通过共享的yield_sandbox_events生成器驱动 Agent用persist_sandbox_event持久化每个事件强制执行预算默认硬上限SCHEDULED_RUN_HARD_CAP_SECONDS 60 * 6060 分钟见 timeouts.py软预算为其 40%24 分钟。注释特别说明 Celery 线程池会静默忽略soft_time_limit/time_limit所以预算必须在任务体内实现。命中RequestPermissionRequest审批门标记AWAITING_APPROVAL、发通知、不写终态返回。恢复机制归审批项目所有在那之前仅作展示用终态。正常流结束finalize_persist→ 提炼约 120 字符的摘要_clip_summary截断时按单词边界断句并加省略号→ 标记 SUCCEEDED。驱动循环内任何异常标记 FAILED带异常类名 详情并发通知。刻意吞掉异常防止 Celery 自动重试。摘要生成有两条路径优先取BuildStreamingState.message_chunks中最近流式文本若内存中无待刷 chunk如中途已 flush则回退扫描持久化消息倒序查找最后一个agent_message元数据 blob见_summary_from_session_messages。3.3 卡死运行清扫器stuck-run sweepercleanup_stuck_scheduled_runs每小时运行beat_schedule.py判定规则tasks.pyQUEUED 超过QUEUE_RESIDENCY_SECONDS 15 * 6015 分钟→failed (stuck)RUNNING 超过SCHEDULED_RUN_HARD_CAP_SECONDS TURN_RECLAIM_SLACK_SECONDS60 分钟 15 分钟→failed (stuck)。两者与 Celery 的expires构成同一策略的三处执行点见 timeouts.py 的注释队列驻留超限的消息被丢弃且行被同阈值回收正常跑满自身预算的运行一定会先把自己标记 FAILED清扫器只是兜底。3.4 专用 worker 进程apps/scheduled_tasks.py 定义了独立 Celery app配置来自onyx.background.celery.configs.scheduled_tasksautodiscover_tasks只加载onyx.background.celery.tasks.scheduled_tasks模块worker_ready时调用start_metrics_server(scheduled_tasks)即 Prometheus 端口 9098。worker 初始化沿用app_base的启动检查等 Redis、等 DB、等文档索引并设置 PostgreSQL 应用名POSTGRES_CELERY_WORKER_SCHEDULED_TASKS_APP_NAME。四、数据模型overview 给出了核心 ORM 模型的骨架与 db/enums.py 和 db/scheduled_task.py 的源码完全对应class SessionOrigin(str, Enum): # INTERACTIVE / SCHEDULED / SLACK # 加到 BuildSession默认 INTERACTIVE class ScheduledTaskStatus(str, Enum): # ACTIVE / PAUSED class ScheduledTaskRunStatus(str, Enum): # QUEUED / RUNNING / SUCCEEDED / # FAILED / SKIPPED / AWAITING_APPROVAL class ScheduledTaskTriggerSource(str, Enum): # SCHEDULED / MANUAL_RUN_NOW class ScheduledTask(Base): __tablename__ scheduled_task id, user_id (FK user, CASCADE) name (str), prompt (text) cron_expression (str), editor_mode (str) status (enum, default ACTIVE) next_run_at (DateTime tz, nullable) # dispatcher 唯一读取的字段 deleted (bool, default False) created_at, updated_at runs ← back-populated, cascade all,delete-orphan __table_args__ ( Index(ix_scheduled_task_dispatch, status, deleted, next_run_at), Index(ix_scheduled_task_user_created, user_id, desc(created_at)), ) class ScheduledTaskRun(Base): __tablename__ scheduled_task_run id, task_id (FK scheduled_task, CASCADE) session_id (FK build_session, SET NULL) # 执行器创建会话后回填 status (enum, default QUEUED), trigger_source (enum) skip_reason / error_class / error_detail (nullable) started_at (default now), finished_at (nullable) summary (str, ~120 chars of final agent message) __table_args__ ( Index(ix_scheduled_task_run_task_started, task_id, desc(started_at)), Index(ix_scheduled_task_run_status, status), )设计要点源码佐证next_run_at是派发器唯一读取的字段暂停置 NULLupdate_scheduled_task中statusPAUSED分支、编辑时重算、deletedtrue从认领查询中排除claim_due_scheduled_tasks的 WHERE 条件。session_id可空在派发器 INSERT 与执行器创建会话之间的短暂窗口内为 NULL。没有attempts计数V1 无重试。ScheduledTaskRunStatus.is_terminal()只把 SUCCEEDED/FAILED/SKIPPED 视为终态enums.pyAWAITING_APPROVAL不在其中。error_class是封闭枚举task_missing/sandbox_wake_failed/executor_error/timeout/stuck/agent_exception让仪表盘与排障查询可以按已知词表透视意外的运行时异常统一记agent_exception真实异常类名与消息放入error_detail。skip_reason也是封闭枚举prior_in_flight/owner_craft_disabled。派发器的认领查询用selectinload(ScheduledTask.user)预加载所有者以便在事务内做 Craft 启用判定。mark_run_status在写入终态时自动填充finished_atinsert_run对 SKIPPED 行也当场填finished_at避免 UI 特判。数据库操作全部集中在 db/scheduled_task.py按项目 CLAUDE.md 约定所有查询下沉到 db 层所有函数首个参数为db_session包括任务 CRUDcreate_scheduled_task/update_scheduled_task/soft_delete_scheduled_task所有权与 NOT_FOUND 抛出语义一致、派发热路径claim_due_scheduled_tasks/advance_next_run_at、run CRUDinsert_run/mark_run_status/list_runs_for_task、卡死扫描find_stuck_runs以及会话横幅辅助get_scheduled_run_context与审批预授权查询get_live_scheduled_run_grants。五、API 规范所有端点抛出OnyxError返回类型化 FastAPI 响应挂载在/api/build/scheduled-tasks复用现有/build前缀与require_onyx_craft_enabled门控作用域限定在认证用户V1 无管理员视图。路由实现见 api.py权限依赖为require_permission(Permission.BASIC_ACCESS)。方法路径说明GET/scheduled-tasks任务列表id、名称、人类可读调度、状态、next_run_at、最近一次运行摘要POST/scheduled-tasks创建。编辑器输入编译为 cron可选run_immediatelyGET/scheduled-tasks/{id}任务详情 未来 3 次触发时间UI 预览PATCH/scheduled-tasks/{id}部分编辑。调度变化时重算next_run_at暂停→NULL恢复→重算。进行中的运行不受影响DELETE/scheduled-tasks/{id}软删除幂等返回 204POST/scheduled-tasks/{id}/run-now插入manual_run_nowrun 并入队执行器暂停时可用不触碰next_run_atGET/scheduled-tasks/{id}/runs分页运行历史50/页cursor参数最新在前GET/build/sessions/{id}/scheduled-run-context若会话来自定时运行返回可选的任务名 id 定时started_at否则 404。供会话视图横幅使用V1 没有 live-attach 端点。请求体校验细节源码确认ScheduledTaskCreatename1–200 字符、prompt非空、editor_mode为 Literal 枚举、editor_payload是受类型约束的联合体、status默认 ACTIVE、run_immediately默认 false另有pre_approved_app_ids与pre_approved_mcp_server_ids预授权目标配合审批门使用。模型extraforbid未知字段直接 422。一个巧妙的校验管道_dispatch_editor_payload作为model_validator(modebefore)把原始editor_payloaddict 按editor_mode路由到对应的类型化模型IntervalPayload/DailyWeeklyPayload/AdvancedPayload不匹配的形状在 FastAPI 层表现为 PydanticValidationError→ 422。PATCH 用_editor_pair_consistency强制editor_mode与editor_payload必须成对出现。_validate_app_ids拒绝未知外部应用 id_validate_mcp_server_ids拒绝该用户在 Craft 中不可用的 MCP server id已授权的存量 grant 可保留。runs 列表的cursor是 ISO-8601 时间上一页最后一行started_atnext_cursor为空表示已到末页游标分页与ix_scheduled_task_run_task_started (task_id, started_at DESC)索引精确对齐。run_immediately与run-now都遵循先落库再入队原则run 行持久化提交后才_enqueue_executor避免 worker 跑到自己的输入前面。入队时附带tenant_id由TenantAwareTask在任务体内设置租户 contextvar否则会在默认 schema 上执行。六、调度语义Cron 编译与求值调度相关的纯函数集中在 schedule.py它自述为 overview 文档中 cron 语义的单一事实来源DB 存规范 5 字段 cron 字符串 editor_modeUI 提示三种编辑器模式保存时统一编译为同一 cron 形式compute_next_run_at返回 UTC 时间比较也在 UTC 进行。6.1 三种编辑器模式间隔IntervalN 分钟/小时/天*/N * * * */0 */N * * */M H */N * *日节奏时 M:H 是编辑器要求的具体时刻。每日/每周Daily/WeeklyM H * * weekdays如0 9 * * 1,3,5。高级Advanced用户手写用croniter校验。compute_next_run_atcroniter(cron, after).get_next(datetime)存 UTC、UTC 比较。6.2 源码级校验规则Pydantic 模型在 HTTP 边界完成了全部校验因此compile_to_cron是纯变换源码注释明确没有dict[str, Any]查找、没有临时字符串解析_HH_MM_RE ^([01]?\d|2[0-3]):([0-5]\d)$24 小时制HH:MM格式与 UI 的input typetime对齐。IntervalPayloadevery 1unit days时time_of_day必填model_validator 强制。DailyWeeklyPayloadweekdays是 0–6 的整数列表0周日cron 约定空列表 每天等价于*。AdvancedPayload必须恰好 5 个字段且croniter.is_valid通过。compile_to_cron的边界兜底分钟间隔every 59折叠为0 * * * **/60 * * * *非法小时间隔every 23折叠为0 0 * * *——与 UI 依赖的历史行为一致。读侧compute_next_run_at/next_n_fires/human_readable也有_validate_cron防御把非法表达式、非 5 字段、无未来触发如不可能表达式统一转成OnyxError(INVALID_INPUT)。human_readable用cron-descriptor的ExpressionDescriptor渲染人类可读描述失败时回退原始 cron 字符串。6.3 时间语义要点暂停中途触发进行中的运行照常完成next_run_at→ NULL恢复时从恢复时刻起重算不回火。过期任务只触发一次并向前推进标准 cron 行为。没有未来触发的 cron在 POST/PATCH 时直接拒绝。调度变更重算规则update_scheduled_taskcron 变更且任务为或变为ACTIVE → 从 now 重算变为 PAUSED → NULL变为 ACTIVE → 重算。各预授权目标字段遵循普通 patch 语义提供的替换该类目标省略的不动。七、运行生命周期queued ─► running ─► succeeded │ └─► failed (executor crash | budget | ACP error) └─► awaiting_approval ─► (approvals project resumes) ─► running ─► ... dispatcher also writes: skipped (prior run still in flight; next_run_at advances)补充说明源码确认AWAITING_APPROVAL的恢复机制归审批项目所有在该项目落地前仅作展示用终态。多个定时运行与交互式send_message路径可以并发执行在同一沙箱上——没有序列化租约。执行器仍会为每个 BuildSession 获取 prompt slot 锁session_manager.prompt_slot这是与交互式路径对称的防御性做法当前每个定时运行都有全新 BuildSession、锁不会竞争但未来若让定时与交互共享 BuildSession此锁自动提供保护。若锁未获取成功并发回合在飞run 被标记 FAILEDerror_classagent_exception。每次触发都会stamp_turn_deadline软预算 min(软预算, budget)硬上限 budget离开时clear_turn_deadlineslot 未丢失时避免后续回合继承过期的定时截止时间。超时与完成的竞态处理很精细先检查PromptResponse/Error再检查预算——截止时间恰好在 Agent 完成瞬间触发时不能把成功误标为超时预算检查先于持久化防止失控 Agent 无限增长转录。stop_reason cancelled的PromptResponse被视为失败而非成功opencode 中止 Agent如MessageAbortedError。八、UI 设计列表页/craft/v1/tasks名称、调度、状态、last_run相对时间 成功/失败图标、next_run、行操作Run now、Pause/Resume、Edit、Delete。空状态展示产品文档中的三个 starter prompts。编辑器/new、/:id/edit单一表单 三 Tab 分段控件选择调度模式Prompt 输入框复用 Craft 聊天输入产品需求 10未来 3 次运行预览由前端用cron-parser计算。详情页/:id头部名称、调度、状态开关、next_run、操作按钮 分页运行历史表格。行点击 → 打开会话视图succeeded/failedqueued/running/awaiting_approval/skipped不可点击、显示 tooltip。会话视图横幅当 scheduled-run-context 端点返回结果时在转录上方渲染This session was started by scheduled task X at Y. ← Back to task.。聊天输入保持可用——用户可以在定时运行的会话上发送后续消息。通知两个新的铃铛条目——Task X failed、Task X needs approval深链到 run 行。通知实现在执行器的_notify中标题Scheduled task {task_name} failed/needs approvaladditional_data携带task_id/run_id且刻意 best-effort——通知失败绝不掩盖底层 run 状态更新。移动端列表/详情可容忍编辑器明确要求桌面端产品需求 8。九、测试策略详见 tests.md覆盖计划、既有测试套件布局与手工冒烟清单。E2E 冒烟web/tests/e2e/scheduled-tasks.spec.ts——单一 speccreate, run, and verify a run row exists登录标准 worker 用户 → 访问/craft/v1/tasks若重定向到/app则 Craft 功能开关关闭spec 软跳过→ 用唯一名称 promptsay hi 每 5 分钟间隔创建任务 → 点save-and-run-now→ 断言详情页 active 状态芯片 → 最多等 60 秒看到data-run-statussucceeded或failed的 run 行两者皆可证明 dispatcher → executor → 运行历史链路端到端可达。前端选择器已锁定new-task-button、task-name-input、task-prompt-input、interval-every、save-and-run-now、task-status-active、data-run-status。运行方式npx playwright test scheduled-tasks需要完整本地 Onyx 栈web、API、Postgres、Redis以及专用celery_worker_scheduled_tasksworker——缺了它 run 永远到不了终态测试在第 6 步超时。刻意不自动化依赖 code review 手工清单FOR UPDATE SKIP LOCKED派发并发、卡死清扫器、审批路径、侧边栏过滤、HTTP API 的用户所有权边界。手工冒烟清单合并前逐项过一遍每 2 分钟任务 Onyx 搜索 prompt离开 6 分钟回来确认三次运行都有完整会话与合理summary周一/三/五 9 AM验证next_run_at正确、UTC 9 AM 强制 tick 能触发中途暂停进行中运行完成、无新触发恢复 → 下次触发从now()向前计算不回火审批边界run 停在AWAITING_APPROVAL通知铃铛有条目同一沙箱上的交互式 Craft 仍可用中途杀 workerrun 处于 RUNNING 时停掉celery_worker_scheduled_tasks一小时内清扫器把它转为FAILED、error_classstuck。十、部署与运行要求调度功能由 Celery Beat 驱动30 秒派发 每小时清扫需要 Beat 进程在跑必须部署专用 worker 进程celery_worker_scheduled_taskssupervisord / dev runner / Helm chart 均已注册执行器只在其上运行多租户模式下派发与执行都依赖租户 contextvartenant_id随任务入队确保任务落到正确 schemaV1 无重试Celery 消息带expires死消费者不会无限堆积 backlog每个入队点都传 expires符合项目 CLAUDE.md 约定。十一、深入阅读的源码地图功能总设计docs/craft/features/scheduled-tasks/overview.md测试计划docs/craft/features/scheduled-tasks/tests.md数据库操作层backend/onyx/db/scheduled_task.py调度纯函数cron 编译/校验/求值backend/onyx/server/features/build/scheduled_tasks/schedule.pyREST 路由backend/onyx/server/features/build/scheduled_tasks/api.py无头执行器backend/onyx/server/features/build/scheduled_tasks/executor.pyCelery 任务派发/执行/清扫backend/onyx/background/celery/tasks/scheduled_tasks/tasks.pyBeat 调度注册backend/onyx/background/celery/tasks/beat_schedule.py专用 worker appbackend/onyx/background/celery/apps/scheduled_tasks.py枚举定义状态机/错误词表backend/onyx/db/enums.py超时注册表预算与清扫阈值backend/onyx/server/features/build/timeouts.pyE2E 冒烟测试web/tests/e2e/scheduled-tasks.spec.ts【免费下载链接】danswerOpen Source AI Platform - AI Chat with advanced features that works with every LLM项目地址: https://gitcode.com/GitHub_Trending/da/danswer创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考