Earendil发布Pi Durable：支持故障恢复的代理框架

2026 年 10 月 1 日，Earendil 工程团队联合 Pi 社区正式发布 Pi 1.0。此次发布不仅标志着经过长期硬化、维护与积极开发的成果落地，更伴随推出了一款名为 Pi Durable 的实验性新产品。Pi Durable 专为长期运行设计，强调耐用性与可塑性，旨在支持代理应用程序在任何环境下稳定运行。

设计理念：从终端到通用框架

Pi 1.0 的核心优势在于其单终端、单人驱动的模式：若进程停止，用户可直接介入并指示继续。这一特性保持不变，但 Earendil 希望将技术普及至更多场景，因此推出了 Pi Durable。它并非 Pi 编码代理的替代品，而是一个构建各类代理应用（包括编码代理）的基础框架。

Pi Durable 遵循极简主义和可塑性原则，支持与 pi-ai 等组件共享代码。通过独立于编码代理进行探索，团队得以在不干扰核心产品的情况下，验证代理应用的设计经验。

“带”的概念定义

在 Pi Durable 语境下，“带”（Belt）指代运行一个或多个与大语言模型并行对话所需的存储及计算环境。它提供模型所需工具及其执行环境。

• 对话：用户与代理之间的互动记录，以成绩单形式呈现。
• 代理：大语言模型及其配置（如思维水平、可用工具）。
• 工具与执行环境：工具通过执行环境（本地笔记本、远程虚拟机或内存沙箱）完成任务。每次对话的工具与环境配置可能不同。
• 任务：从调用模型到执行工具的完整流程。

为便于 AI Agent 理解，Pi Durable 源码（不含测试）约 15,000 行，对应 GPT 约 150,000 tokens，Claude 约 250,000 tokens。实际开发中，Agent 通常无需加载全部代码，例如仅存储后端部分就有 3,000 行，往往可以跳过。

跨平台运行能力

Pi Durable 目前支持任何拥有 JavaScript 运行时的环境。一个“带”可在多种存储后端打开，包括内存、SQLite 和 JSONL。其中，SQLite 和 JSONL 存储不依赖 Node API，通过小型适配器即可在 Bun 或 Cloudflare 持久对象中运行。存储接口简洁，易于扩展。

在 SQLite 模式下，“带”仅在内存中保留工作集（活动转录、实时任务、待提交数据），其余数据留存磁盘。活跃转录受模型上下文窗口限制，旧消息会在溢出前被压缩总结，确保即使包含数千条信息的长对话也能高效驻留内存。

工具可从执行环境获取文件或外部资源，支持本地文件访问。执行环境接口简单，允许将远程环境暴露给工具，实现“带”在一台机器运行而工具在另一台机器执行。每个对话的工作目录独立构建，确保位置隔离。

import { BACKGROUND_CONTEXT } from ;
import { createModels } from ;
import { openaiProvider } from ;
import { createRegistry, Harness } from "@earendil-works/pi-durable";
import { NodeExecutionEnv } from ;
import {
    openNodeSqliteStorage,
} from "@earendil-works/pi-durable/storage/sqlite/node";
import { CodingTools } from ;

const context = BACKGROUND_CONTEXT; // every call takes a context for cancellation
const models = createModels();
models.setProvider(openaiProvider());

const registry = createRegistry();
registry.install(CodingTools); // read, write, edit, bash

const env = ({ cwd }: { cwd?: string }) =>
    new NodeExecutionEnv({ cwd: cwd ?? process.cwd() });
const harness = await Harness.open(
    await openNodeSqliteStorage("./agent.sqlite"),
    { models, registry, env },
    context,
);
// The root conversation: created on first use, and the same one after every
// restart.
const root = await harness.root(context, {
    agent: {
        model: { provider: "openai", modelId: "gpt-6.1-sol" },
        cwd: ,
    },
});

故障恢复机制

Pi Durable 具备极强的容错性，无论笔记本电脑休眠、容器重新部署或内存耗尽，系统均可恢复。每一步运行均作为任务处理，并在前进前存储检查点。若进程终止，新进程打开相同存储，识别未完成任务并从最后检查点继续。

具体行为如下：

• 中断的样式请求会重新发送；已删除的请求再次发送。
• 部分答案留在成绩单中标记为中止。
• 安全的工具调用会重放；否则通知模型中断情况。
• 子代理虽无内置功能，但几行代码即可实现。安全重运行的子代理工具会找回子代理并等待回复。
• 排队消息保持队列状态，客户端崩溃后重试可获得原始提交结果，避免重复执行。

const job = {
    type: "input",
    content: "Fix the flaky login test",
    requestId: "job-42",
} as const;
await root.submit(job, context);
// The process dies here, in the middle of a tool call.

// A new process opens the same storage.
const harness = await Harness.open(
    await openNodeSqliteStorage("./agent.sqlite"),
    { models, registry, env },
    context,
);
harness.resume(); 
const root = await harness.root(context);
// the same submission, answered
const settled = await (await root.submit(job, context)).wait(context);

并发对话支持

单个“带”支持同时运行多个互不阻塞的对话。对话可从起点开始或在转录任意点分叉，并继承父历史。例如，Slack 频道中代理回答提及，随后开启线程，两者并行运行。

每个对话拥有独立的代理配置：模型、思维级别、扩展选择、活跃工具、额外指令及工作目录。主代理旁的审核员可使用低成本模型、只读工具及独立结账。

const channel = await harness.root(context);
const question = await channel.submit(
    { type: "input", content: "@agent why did the deploy fail?" },
    context,
);
const answered = await question.wait(context);

// Someone replies to the agent's answer in a thread. Every conversation names
// its owner, which decides what an abort reaches (more on that under Tasks).
// The thread has none.
const thread = await channel.fork(
    answered.answer!,
    { ownership: { kind: "ownerless" } },
    context,
);

// Both conversations work at the same time.
const inThread = await thread.submit(
    { type: "input", content: "@agent can we roll it back?" },
    context,
);
const inChannel = await channel.submit(
    { type: "input", content: "@agent who is on call today?" },
    context,
);
await Promise.all([inThread.wait(context), inChannel.wait(context)]);

扩展体系

Pi Durable 通过扩展实现模块化，确保每个插入部件都具备耐用性。扩展是系统提示部分、工具、子和任务的命名捆绑包。应用在注册表中安装扩展，每个对话选择使用的扩展和工具，仅存储名称。

系统提示部分

每次请求前，系统提示从对话扩展部分重建，以便下一个请求捕获更改。Pi Durable 记录转录中的变化及位置，重启或分叉时可见模型视角。对于支持动态修改的系统提示和工具，仅发送变更内容，保持提示缓存有效。

import { defineExtension, section } from "@earendil-works/pi-durable";

const ProjectContext = defineExtension({
    name: "project-context",
    sections: [
        // Read from the conversation's execution environment. The files can be
        // loaded and watched in the background; every request renders the
        // latest state.
        section("agents_md", (input) => agentsMd.latest(input.env)),
        section("skills", (input) => skills.latest(input.env)),
    ],
});

工具定义与安全重放

每个工具调用作为持久任务运行，意图在执行前存储。崩溃后，仅标记为“安全”的工具才能重新启动；否则通知模型中断并保存输出供决策。对话可定制工具集。

import { Type } from "@earendil-works/pi-ai";
import { defineTool } from "@earendil-works/pi-durable";

const searchIssues = defineTool({
    name: "search_issues",
    description: "Search the issue tracker",
    parameters: Type.Object({ query: Type.String() }),
    replay: "safe", // only reads, so a rerun after a crash is fine
    execute: async (args, api) => {
        
        api.output(`searching for ${args.query}\n`);
        return {
            content: [{ type: "text", text: await tracker.search(args.query) }],
        };
    },
});

const deploy = defineTool({
    name: "deploy",
    description: "Deploy a version to production",
    parameters: Type.Object({ version: Type.String() }),
    // No replay: a deploy interrupted by a crash is reported to the model,
    // never repeated.
    execute: async (args) => ({
        content: [{ type: "text", text: await ci.deploy(args.version) }],
    }),
});

registry.install(defineExtension({ name: "ops", tools: [searchIssues, deploy] }));

// The thread may search, but not deploy.
await thread.configure({ tools: { remove: [deploy] } }, context);

工具可访问“带”API，提交条目、启动任务和对话，与其他对话交互。这使得构建子代理仅需几行代码：创建拥有该调用的对话，配置小模型与特定指令，并等待答案。UI 可在通话中显示子代理状态。

import type { AssistantMessage } from "@earendil-works/pi-ai";
import { AssistantEntry, configure } from "@earendil-works/pi-durable";

const triage = defineTool({
    name: "triage",
    description: "Label an incoming issue as bug, feature, or question",
    parameters: Type.Object({ issue: Type.String() }),
    // a rerun after a crash finds the same subagent and the same submission
    replay: "safe",
    execute: async (args, api, context) => {
        const child = await api.commit(async (tx) => {
            const existing = (
                await tx.scanConversations({ ownerTaskId: api.taskId }, 1)
            ).items[0];
            if (existing !== undefined) return existing.id;
            // Owned by this call, so aborting the call aborts the subagent.
            const created = await tx.createConversation({
                ownership: { kind: "task", taskId: api.taskId },
            });
            // It starts as a copy of this conversation's agent. Make it a small
            // model without tools.
            await configure(tx, created.id, {
                model: { provider: "openai", modelId: "gpt-6-luna" },
                tools: [],
                instructions: "Answer with one word: bug, feature, or question.",
            });
            return created.id;
        }, context);
        // lets a UI show the subagent under the call
        await api.details({ conversationId: child }, context);
        const subagent = await api.conversation(child, context);
        const request = {
            type: "input",
            content: args.issue,
            requestId: `triage:${api.taskId}`,
        } as const;
        const settled = await (
            await subagent!.submit(request, context)
        ).wait(context);
        // The answer is an entry in the subagent's transcript. Read it and take
        // its text.
        const entry = await api.commit(
            (tx) => tx.entry(AssistantEntry, settled.answer!),
            context,
        );
        const message = entry?.model?.[0] as AssistantMessage;
        const text = message.content
            .flatMap((content) => (content.type === "text" ? [content.text] : []))
            .join("");
        return { content: [{ type: "text", text }] };
    },
});

扩展还可修改其他扩展的工具。同名工具在后装扩展中覆盖先前版本（如在 Python virtualenv 中运行 bash）。包装器可装饰任何获胜工具，不受扩展选择顺序影响。

import { wrapTool } from "@earendil-works/pi-durable";
import { createBashTool } from ;

// Times every bash call, whichever bash the conversation ends up with.
const Timing = defineExtension({
    name: "timing",
    wraps: [
        wrapTool(createBashTool(), (bash) => ({
            ...bash,
            execute: async (args, api, context) => {
                const start = Date.now();
                try {
                    return await bash.execute(args, api, context);
                } finally {
                    metrics.record("bash", Date.now() - start);
                }
            },
        })),
    ],
});

Hooks 钩子机制

扩展可通过 Hooks 进入任务（生成响应、调用工具、压缩等）。Hooks 可在请求进入模型前重写、阻止或替换结果，或自行写摘要。崩溃后可重运行，决策结果存储在备忘录中与任务绑定。

import { hook, ToolTask } from "@earendil-works/pi-durable";

const Approval = defineExtension({
    name: "approval",
    hooks: [
        hook(ToolTask, {
            beforeTool: async (call, api, context) => {
                if (call.name !== "deploy") return undefined;
                // After a restart, the hook finds the stored answer instead of
                // asking again.
                let approved = await api.memo<boolean>(
                    "approval:deploy",
                    context,
                );
                approved ??= await api.memo(
                    "approval:deploy",
                    await askInSlack(call),
                    context,
                );
                return approved
                    ? undefined
                    : { block: "Nobody approved the deploy." };
            },
        }),
    ],
});

多个扩展可连接同一目标，形成链式结构。按对话选择扩展的顺序执行，每个链接定义其行为。工具传递参数至下一链，首个块可停止链；后续工具传递结果。观察者类型（如 AfterResponse）始终运行。抛出异常报告后链继续，但在 beforeTool 中抛出会阻止调用。

Tasks 任务管理

“带”运行内置任务：每个模型请求、工具调用及压缩各为一个任务。扩展带来自定义任务，享有相同机制：步骤后检查点、定时器重启生存、等待其他任务。

多卡支付场景中，若一张被拒，其他付款取消并退款：

import { defineTask, type TaskId } from "@earendil-works/pi-durable";

const Payment = defineTask<{ card: string }, { phase: "charge" }, string>({
    name: "shop.payment",
    version: 1,
    initial: () => ({ phase: "charge" }),
    phases: {
        charge: