@chao.song/task-agent

用于构建“单 Task、可继续、可持久化”AI Agent 的领域中立 TypeScript 框架。

• 当前实现导览：DESIGN.md
• 安全边界：SECURITY.md
• 架构决策：docs/adr
• 仓库目标态设计：task-agent framework design

安装

npm install @chao.song/task-agent

要求 Node.js 22.19.0 或更高版本。框架使用 Node 内置 node:sqlite。

核心模型

Task

Task 是长期业务任务，拥有独立目录、持久 Session 和领域状态。Task 可以跨进程、跨多次 CLI 调用继续。

created -> active -> waiting -> active -> completed
                    `-------> failed

Run

Run 是 Agent 的一次有界唤醒。create 和每次 continue 都产生一个 Run。同一 Task 同时最多有一个 active Run。

queued -> running -> succeeded | failed | aborted

Run succeeded 不代表 Task completed。等待用户或 Job 时，Run succeeded、Task waiting。

Durable Job

Durable Job 表示无需占住 Agent Loop 的外部异步工作。Provider Adapter 实现：

submit -> query -> collect
              `-> cancel (optional)

Job 请求、Provider task ID、状态、nextPollAt、lease、结果和错误保存在 kernel.sqlite。Provider 不保证提交幂等且网络结果不确定时，Job 进入 submission_uncertain，框架不会自动重复付费提交。

定义 Scenario

import {
  defineScenario,
  type ScenarioTool,
  type ToolResult,
} from "@chao.song/task-agent";

interface InspectInput {
  path: string;
}

const inspectTool: ScenarioTool<InspectInput> = {
  name: "inspect_record",
  description: "Inspect one record without changing it.",
  inputSchema: {
    type: "object",
    properties: { path: { type: "string", minLength: 1 } },
    required: ["path"],
    additionalProperties: false,
  },
  concurrency: "safe",
  async execute(input, context): Promise<ToolResult> {
    context.reportProgress(`Inspecting ${input.path}`);
    return { ok: true, output: { path: input.path, valid: true } };
  },
};

export const scenario = defineScenario({
  name: "inventory",
  version: "1.0.0",
  systemPrompt: "Inspect inventory records and complete only with verified evidence.",
  tools: [inspectTool],
  security: {
    workspaceTools: ["workspace_read", "workspace_grep", "workspace_find", "workspace_list"],
    maxToolResultBytes: 64 * 1024,
    toolTimeoutMs: 30_000,
  },
  limits: { maxTurns: 40, maxWallTimeMinutes: 30, maxToolConcurrency: 2 },
});

Scenario 可配置：

• 完整 System Prompt。
• inline 或 durable Tool。
• 显式 skill 目录。
• workspace 路径策略。
• Tool 结果、超时和并发限制。
• prepareResources 领域注册 Hook。
• validateCompletion 产品完成验证器。
• compaction 需要保留的摘要章节。

产品不应直接依赖 Pi 类型。只有 src/runtime/pi/ 知道 Pi Session 和 Tool API。

使用 Application Service

import { TaskApplicationService } from "@chao.song/task-agent";
import { scenario } from "./scenario.js";

const service = new TaskApplicationService();
const taskDir = "/var/lib/example-agent/tasks/inventory-001";

await service.createTaskAndWait({
  taskDir,
  scenario,
  prompt: "检查所有库存记录",
  idempotencyKey: "request-001",
  model: {
    provider: "example",
    protocol: "openai-completions",
    baseUrl: "https://example.com/v1",
    apiKey: process.env.EXAMPLE_API_KEY!,
    model: "model-name",
  },
});

const snapshot = await service.getTask(taskDir);
console.log(snapshot.task.status, snapshot.waitingFor);

await service.continueTaskAndWait({
  taskDir,
  scenario,
  prompt: "继续并处理上次的问题",
  idempotencyKey: "request-002",
});

主要 API：

| API | 用途 | | --- | --- | | createTaskAndWait | 创建 Task，执行并等待本次 Run 结束 | | continueTaskAndWait | 在同一 Task 和 Session 创建新 Run | | getTask / getRun | 查询持久状态 | | waitRun | 等待指定 Run 终态 | | listJobs / getJob | 查询 Durable Job | | workJobs | 推进当前到期 Job | | serveTaskUntilBlocked | 单 Task 前台调度并自动恢复 Agent | | cancelJob | 请求取消 Durable Job |

Durable Tool 示例

import type { DurableTool } from "@chao.song/task-agent";

const renderTool: DurableTool = {
  name: "render",
  description: "Render one artifact asynchronously.",
  inputSchema: { type: "object" },
  execution: "durable",
  provider: "render-service",
  submitIdempotent: true,
  async submit(input, context) {
    const id = await provider.submit(input, context.idempotencyKey, context.signal);
    return { providerTaskId: id, nextPollAt: new Date(Date.now() + 5000).toISOString() };
  },
  async query(providerTaskId, context) {
    return provider.query(providerTaskId, context.signal);
  },
  async collect(providerTaskId, context) {
    return { ok: true, output: await provider.collect(providerTaskId, context.workDir) };
  },
};

Agent 调用 durable Tool 只会登记 Job。随后应调用 Kernel 的 wait_for_jobs 结束当前 Run。Worker 完成 Job 后，调用方可以显式 continue，或使用 serveTaskUntilBlocked() 自动恢复 Agent。

Run 结束协议

每个新控制面 Run 必须通过一个 Kernel Tool 给出明确 disposition：

| Tool | Run | Task | | --- | --- | --- | | request_user_input | succeeded | waiting | | wait_for_jobs | succeeded | waiting | | complete_task | succeeded | completed | | fail_task(recoverable: true) | failed | waiting | | fail_task(recoverable: false) | failed | failed |

模型自然停止但没有调用结束 Tool 时，框架记录可恢复的 missing_run_disposition，不会把任务误判为完成。

持久化与幂等

每个 Task 的 state/kernel.sqlite 是控制面唯一事实源，保存：

• Task、Run 与 Session 文件位置。
• invocation 幂等键和请求 hash。
• prepareResources reservation。
• waiting、failure 和 completion disposition。
• Resource、Run-Resource 关联和 OutputReference。
• Durable Job、lease、Provider ID、结果和错误。

相同幂等键和相同请求返回原 Run；相同幂等键绑定不同请求会被拒绝。控制 Tool 使用 (runId, toolCallId) 保证重放安全。

events.jsonl 和 Pi Session JSONL 是辅助记录，不参与 Task 状态恢复。产品投影文件也不能替代 SQLite 事实。

Workspace 与资源

<taskDir>/
|- input/                 导入的不可变输入
|- work/                  Tool 中间文件
|- output/                最终交付
|- state/
|  |- kernel.sqlite
|  |- run.lock
|  |- abort.requested
|  `- pi-agent/
`- logs/events.jsonl

资源导入要求绝对路径和 allowed roots，拒绝逃逸符号链接，并在复制前后校验文件身份、大小和 SHA-256。Runtime Prompt 只获得 Task 内相对路径。

完成输出必须位于 output/。产品 validateCompletion 先验证领域证据，Kernel 再校验 canonical path、普通文件、bytes 和 SHA-256。

公共产品 CLI

产品调用 runAgentCli() 可获得统一命令：

runAgentCli({
  command: "inventory-agent",
  displayName: "Inventory Agent",
  scenario,
  envPrefix: "INVENTORY_AGENT",
  serve: true,
});

serve: true 会额外暴露 serve 和 create --serve。ops-agent 未启用，media-agent 已启用。

测试

测试 Scenario 时可使用确定性的 Fake Runtime：

import { FakeRuntime } from "@chao.song/task-agent/testing";
import { TaskAgentKernel, TaskApplicationService } from "@chao.song/task-agent";

const service = new TaskApplicationService(
  new TaskAgentKernel(new FakeRuntime([{ type: "complete_task", paths: [] }])),
);

当前边界

• 当前是单机、同一 PID namespace 模型，不支持多主机共享 Task 根目录。
• serveTaskUntilBlocked 是单 Task 前台循环，不负责全局任务发现。
• 尚无后台 daemon、tasks catalog、active Run 崩溃接管和事务 Inbox/outbox。
• A2A/HTTP Adapter 尚未实现。
• 路径检查、Tool schema、审批和超时是应用层控制，不是 OS 沙箱。