@ amaster.ai/ pi- task- 排程器

已排定的任务域类型,执行调度器,以及 pi 代理扩展.

软件包拥有调度解析,任务生命周期状态,流程本地计时器,跑车回调器,调度钩,以及用于自主任务管理的LLM可调用工具.

默认的 Pi 扩展跑取器会作为后续发送预定的提示:一个闲置的代理立即启动,而繁忙的代理则会在同一会话中排出后处理的提示.

存储

基于内置文件存储的包船( 默认, 外部依赖性为零 ) :

• JsonScheduledTaskStore 持续执行 JSON 中带有原子写法的文件的任务
• FileSchedulerLock ——基于PID的文件锁定,用于单所有者执行

用于数据库支持的存储(多租户,Prisma),使用 @amaster.ai/pi-storage/scheduler 用于执行与 DbScheduledTaskStore 财务报告和审定财务报表 RedisSchedulerLock。 。 。 .

默认( 文件存储)

import {
  PersistentTaskScheduler,
  JsonScheduledTaskStore,
  FileSchedulerLock,
} from "@amaster.ai/pi-task-scheduler";

const scheduler = new PersistentTaskScheduler({
  store: new JsonScheduledTaskStore("/var/lib/pi/tasks.json"),
  lock: new FileSchedulerLock("/var/lib/pi/tasks.lock"),
  runner: async (task, run) => {
    await runPrompt(task.prompt, { sessionId: run.sessionId });
  },
});

await scheduler.start();

自定义存储

执行 ScheduledTaskStore 财务报告和审定财务报表 SchedulerLock 用于任何后端(数据库、 Redis等)的接口 :

import { PersistentTaskScheduler } from "@amaster.ai/pi-task-scheduler";

const scheduler = new PersistentTaskScheduler({
  store: myCustomStore,   // implements ScheduledTaskStore
  lock: myCustomLock,     // implements SchedulerLock
  runner: async (task, run) => {
    await runPrompt(task.prompt, { sessionId: run.sessionId });
  },
});

await scheduler.start();

整合模式

软件包提供两条集成路径,取决于谁控制调度器生命周期.

模式1:扩展自动发现(独立/CLI)

当安装为 npm 依赖 pi.extensions 输入 package.json,运行时自动发现并加载扩展. 扩展创建自己的 PersistentTaskScheduler 以文件为基础的存储( e)JsonScheduledTaskStore + 键 FileSchedulerLock——完全自成一体,不需要外部基础设施.

pi.extensions → session_start → creates file-based scheduler → registers tools + commands

• 存储: 每届会议 tasks-<session-hash>.json 财务报告和审定财务报表 scheduler-<session-hash>.lock 文件在 <agentDir>/data (可通过以下方式配置基准目录: dataDir(中文(简体) )
• 升级: 匹配遗留任务 tasks.json 在第一次使用时复制 保留遗留文件
• 生命周期: 调度器开始于 session_start,停下来 session_shutdown
• 图标数字 : /pi-scheduler-status, (中文(简体) ), /pi-scheduler-list, (中文(简体) ), /pi-scheduler-run-now
• LLM 工具数字 : scheduler_create, (中文(简体) ), scheduler_list, (中文(简体) ), scheduler_get, (中文(简体) ), scheduler_update, (中文(简体) ), scheduler_delete, (中文(简体) ), scheduler_run_now

通过设置密钥配置 pi-scheduler数字 :

项目 .pi/settings.json 数值只有在项目信任被接受之后才被加载,而不是环境插入。 用户和代理设置保留了环境插值.

{
  "pi-scheduler": {
    "dataDir": "/custom/path"
  }
}

模式2:依赖性导入(主机控制)

当主机进程拥有调度器实例(例如由数据库支持)时,它绕过扩展并直接使用注册工具 createSchedulerTools数字 :

import { createSchedulerTools } from "@amaster.ai/pi-task-scheduler/tools";
import { PersistentTaskScheduler } from "@amaster.ai/pi-task-scheduler";

const scheduler = new PersistentTaskScheduler({
  store: myStore,
  lock: myLock,
  runner,
});
await scheduler.start();

// Returns ToolDefinition[] — register via your tool injection mechanism
const tools = createSchedulerTools(scheduler);

在这种模式下,自动发现的扩展应被过滤出,以防止双重工具注册. 主机独立管理调度器启动/停止.

host creates scheduler → createSchedulerTools(scheduler) → tools injected into runtime
                       → extension filtered from auto-discovery

选择模式

时间安排

任务支持三种调度类型:

• interval: 长度字符串, 如 30s, (中文(简体) ), 10m, (中文(简体) ), 1h,或 1d
• once: ISO 时间戳或相对值,例如 +5m
• cron: 5/6-字段克龙表达式和一个RRULA小子集

排程器不会在进程中断时间后追赶丢失的运行 。 当它开始时,它会为启用的任务登记未来的计时器.

钩子

虎克是最佳的可观察回调. 如果一钩投出,调度器会吞入错误并保持任务状态的权威.

结构化结果(Pi 0.99+)

调度工具声明 outputSchema列表返回 { tasks, total, truncated }; 创建/获取/更新/运行 现在返回 { task, truncated }; 删除返回 { taskId, deleted, truncated }当任务记录没有标记时,结构化的任务提示已完成 truncated,即使可读摘要仅显示前100个字符。 超大任务记录省略了快取并保存了任务ID以供更新/删除. 获取包括已存储的任务元数据和运行历史标识符/状态/时间戳,不包括原始执行错误消息. 缺少任务、 错误的时间表和后端失败返回 { error, truncated } 与 isError: true。现有的会话范围同样适用于脚本.

const result = await tools.scheduler_list({});
if (result.error || result.truncated) throw new Error("Inspect the complete task list first");
for (const task of result.tasks.filter(t => t.enabled && t.name === "Daily check")) {
  text(await tools.scheduler_update({ taskId: task.id, enabled: false }));
}

结构化输出限于12个KiB,而顶级列表则限于100个完整的记录. truncated: true 表示省略的记录; total 保留已有的原始数。 超大小记录,带有 id 保留完整的ID和小分级元数据,省略大字段,并携带 truncated: true 在记录和信封上;呼叫者必须在重新使用内容进行更新之前检查该旗帜。 没有记录ID的超大小结果返回打出输出限制错误,而不是剪接路径或内容. 当发生这种情况时, 写入可能已经完成: 在重试前检查保存状态 。 可读文本和渲染器细节是独立限定的.