🔶 kassette：基于对象存储的持久化智能体工作流

使用可嵌入原语让你的智能体工作流持久化，这些原语与你的现有基础设施组合。无需服务器，无需专用运行时，无需新基础设施。

智能体工作流失败的代价令人沮丧：它们运行几分钟到几小时，进行昂贵的 LLM 调用，执行会产生真实世界影响的工具，并经常暂停以等待人类输入或外部系统。然后进程崩溃、超时、被重新部署或直接消失。由于状态在内存中，朴素的重试意味着你要为模型已经给出的答案重新付费，再次触发本应只执行一次的外部影响，并将恢复复杂性推回应用代码。

kassette 通过一个仅追加日志来解决这个问题。它记录已完成的工作，这样在下一次调用时，重放会快速推进日志内容，并从第一个未完成的步骤继续。你现有的队列或作业运行器可以正常重试。kassette 通过重放已完成步骤而不是重新运行它们，确保这些重新调用是安全的。

当你面对的问题不是“如何再次运行这个”，而是“如何避免重复做相同的工作”时，请使用 kassette。也就是说，你的现有技术栈让重试很容易，但不安全。

更详细的理由，请参阅 为什么选择 kassette。

可嵌入，非运行时。 kassette 是一个在你的进程中运行的库，就像 SQLite。它不是独立的服务器、边车或运行/调度你代码的工作者。

无新基础设施。 与你现有的基础设施组合。无需托管数据库，无需部署服务器，无需配置队列。重新调用通过你已有的队列、Webhook 或作业运行器来连接。使用文件系统或对象存储。

纯文本状态。 每次运行都是一个 JSONL 日志。你可以转储它、通过管道传递给 jq、保存为测试固件，或使用 Athena 查询 S3 背书的日志。参见 运维。

等待期间无计算。 正在等待人类审批、CI 流水线或 Webhook？进程直接退出。当信号到达时，重放到你暂停的点并继续。

原生 Serverless。 日志就是状态；进程是一次性的。任何后续可以访问日志的进程都可以继续该运行。复制、同步、跨区域发送。

kassette 提供什么，你需要带来什么

持久执行分解为两个部分：记录已完成工作的日志，以及在崩溃或超时后再次调用相同 runId 的调度器。Temporal 和 Inngest 同时提供了这两者。kassette 仅提供日志。

你连接上你的技术栈中已有的调度器：一个队列、作业运行器，或任何在失败后可以重新运行相同工作的东西。如果一次运行未成功，调度器再次调用它，重放会跳过已经完成的步骤。你不需要心跳、清扫器或运行状态表。

因为调度键和日志键是相同的 runId，重试是安全的。kassette 重放已完成的步骤而不是重新运行它们。如果一个旧的工作进程在重试接管后继续运行，它的下一次日志写入会被拒绝。

具体的队列/作业运行器连接模式，请参见 连接调度器。

• 至多一次日志记录。 一个步骤的结果最多被记录一次。后续调用重放记录的值。条目是仅追加、有序且不可变的。（如果步骤函数的工作完成但其结果未被记录——例如崩溃、防护或并发挂起——该函数可能会运行多次，因此外部影响的幂等性仍由调用者负责。）
• 日志权威。 已记录的工作保持已确定状态。重试的 resume 保留第一个记录的事件，即使后续的重试传递了不同的载荷。kassette 永远不会使日志失效、过期或进行垃圾回收。
• 原子追加。 每个条目完全提交或不提交。像 resume、fork 和自动取消这样的多条目操作是可崩溃恢复的。后端细节在本地存储和远程存储中介绍。
• 单写入者防护。 一次只有一个会话可以向一个运行追加。如果一个旧进程在替换进程启动后继续运行，它的下一次追加会因 FencedError 而失败。防护是按运行进行的。
• 终端状态保护。 一旦 complete、error 或 cancel 落地，在该运行上后续的 start/resume/fork 会抛出 TerminalRunError。调度器可以捕获这个错误并确认陈旧的、无需重新运行工作的重试。
• 确定性重放。 使用相同的日志和代码，重放能重建相同的状态并在第一个未完成的步骤处继续。重放状态仅来自日志。
• 可选版本检查。 传入一个可选的 version 选项，以通过 VersionMismatchError 拒绝不兼容的工作流代码。

kassette 在重放时也会对事件名称冲突、更改的 start 元数据和不匹配的并行分支名称抛出错误。

只需编写一个正常的异步函数，用 ctx.step() 包裹任何你想要记录日志的内容。在重放时，kassette 会返回记录的结果而不是再次运行该步骤。使用 ctx.suspend() 暂停工作流并将控制权返回给调用者。当你恢复时，kassette 会重放到那个点，然后继续处理事件载荷。

没有 DSL、装饰器系统或特殊的控制流。if、while、try/catch 和提早 return 正常工作。将工作流放在框架回调、中间件或你的代码通常运行的任何地方。

import { kassette, LocalStorage } from '@usekassette/kassette';

const storage = new LocalStorage('.kassette');

type Events = {
  'human-approval': { approved: boolean; notes?: string };
};

const agent = kassette<Ticket, Resolution, Events>(
  async (ctx, ticket) => {
    const analysis = await ctx.step('analyze', () => llm.chat('Diagnose this issue and recommend a fix', { ticket }));

    if (analysis.destructive) {
      const approval = await ctx.suspend('human-approval');
      if (!approval.approved) return { outcome: 'skipped', reason: approval.notes };
    }

    const result = await ctx.step('apply-fix', () => executeTool(analysis.suggestedAction));

    return { outcome: 'resolved', result };
  },
  { storage },
);

let result = await agent.start(ticket);
// → { status: 'suspended', event: 'human-approval', runId }

// ─── 你的进程退出 ───────────────────----------------------------

// 通过外部信号如 webhook、UI 审批或回调来恢复:
result = await agent.resume(result.runId, {
  eventName: 'human-approval',
  value: { approved: true },
});
// llm 调用被重放。相同的结果，没有新开销。执行从审批点继续
// → { status: 'success', result: { outcome: 'resolved', ... }, runId }


npm install @usekassette/kassette

@usekassette/kassette — 工作流 API：kassette()、Context、step、suspend、parallel、fork。 @usekassette/core — 低级持久化原语：start、record、waitForEvent、Storage。如果你想将持久性嵌入到现有的智能体循环中而不使用工作流包装器，可直接使用它。 @usekassette/s3 — S3 存储后端。 @usekassette/cli — 用于检查和派生日志的 CLI。参见 CLI。

零运行时依赖。

传递给工作流函数的 Context：

方法	功能
ctx.step(name, fn)	运行并记录一个工作单元。在重放时，返回记录的结果而不是再次运行 fn。如果 fn 抛出异常，根据 { retry: { maxAttempts, delay, backoffRate, maxDelay } } 在进程内重试，然后传播错误。
ctx.suspend(eventName, { timeout?, reason? })	暂停工作流，向调用者返回 { status: 'suspended' }。resume 传递事件载荷。事件载荷通过 TEvents 泛型类型化。如果 timeout 在恢复之前到期，该运行会被最终取消。
ctx.parallel(branches)	并发运行命名分支。如果任何分支挂起，整个并行块都会挂起。
ctx.sleep(ms)	在重放时，仅等待剩余的时间。

kassette() 返回的 Kassette 对象：

方法	功能
workflow.start(input)	启动一个工作流运行并返回 RunResult<TOutput>。
workflow.resume(runId, { eventName, value })	向一个挂起的运行传递事件。重放到挂起点然后继续。
workflow.fork(source)	从现有日志的一部分创建一个新运行。新运行重放到分叉点，然后分道扬镳。

fork() 将现有日志的一部分复制到新的运行中。新运行在内存中重放那个前缀，然后从分叉点继续实时执行。

用它从一个已记录的前缀分支产生多个推测性延续，或者在原始运行出错后回到一个更早的步骤。你保留了分叉前昂贵的工作而无需再次付费。参见 coding-agent。

重放假设工作流代码仍然具有相同的形状。如果你增加、移除或重新排序步骤，旧的日志条目可能会映射到错误的代码。这对于跨越部署挂起的智能体运行来说很重要。

对于步骤的不兼容更改，重命名该步骤。新名称会错过旧的日志条目并实时运行。

// 之前
const analysis = await ctx.step('analyze', () => llm.chat(prompt));

// 之后：不兼容的提示更改
const analysis = await ctx.step('analyze-v2', () => llm.chat(newPrompt));

重命名不足以应对所有更改。重新排序或移除步骤会改变该点之后的日志映射。对于这些更改，配置工作流定义上的 version 作为安全检查。不匹配会抛出 VersionMismatchError，这样你可以丢弃日志并重新开始。

const agent = kassette<Input, Output, Events>(fn, {
  storage,
  version: 'workflow-v3',
});

const result = await agent.start(input);

尽可能将版本与工作流形状关联，而不是与部署 SHA 关联。阅读更多关于 版本控制 的内容。

LocalStorage 将每次运行存储为磁盘上的一个 JSONL 文件。

• 原子追加： 每个条目通过 write() 写入并通过 fsync() 刷新，没有损坏的写入。
• 单写入者防护： 每个 runId 有一个通过原子 link(2) 创建的锁文件。基于 PID 的陈旧检测回收已死进程的锁。

@usekassette/s3 将每次运行存储为 S3 兼容存储中的一个对象。

• 原子追加： 每次 PutObject 原子地写入整个日志
• 单写入者防护： 每次追加使用带有 If-Match 条件的 PutObject，该条件基于上一个 etag。两个会话可能短暂共存，但较旧的会话会在其下一次追加时被防护。

所需后端语义： 读后写一致性与条件写入。S3、R2 和 GCS 均支持这一点。

根据运行可以从何处恢复来选择存储，而不是基于原始速度。对于 kassette 的目标工作负载——10-100 个步骤、小日志条目，以及由 LLM、工具、人类、Webhook 或超时主导的等待——存储延迟不应成为瓶颈。

LocalStorage 追加是 O(1)：一次 fsync，每条约 100µs。当重试进程可以读取相同的持久文件系统时使用它。

RemoteStorage / @usekassette/s3 将每次运行存储为一个对象。每次追加读取当前对象并上传带有一行的整个日志。这使得 start、resume 和 fork 很廉价，因为 readAll 就是一次对象读取，但写入成本随日志增长。一次运行期间总上传字节数为 O(N²)。

这种权衡对于智能体运行通常是可接受的。对于大条目、数百个步骤或多 GB 的日志，尽可能首选 LocalStorage。

kassette 不清理、过期或快照日志，因此存储会随着运行而增长。如果你需要保留限制，请使用生命周期策略或清理作业。

参见 存储后端 和 对象存储设计。

kassette 日志被设计为易于智能体进行检查。智能运行通常小到可以直接检查：数十个昂贵的步骤、KB-MB 级别的日志，以及每次运行只有一个仅追加的 JSONL 日志。调试智能体可以读取整个运行，回答“已经发生了什么？”、“在哪里挂起或失败了？”以及“从哪里可以安全地分叉？”，而无需服务控制台、数据库命令行、快照协议或针对特定应用的检查器。

此仓库在 skills/kassette/SKILL.md 捆绑了一个智能体技能。当智能体需要检查、调试、恢复或分叉一个 kassette 运行时使用它。它涵盖存储发现、日志结构、jq 模式、状态解释、挂起/恢复调试以及安全分叉。

@usekassette/cli 提供 kassette 命令行，用于检查日志和分叉针对任一后端。list、status <runId>、dump <runId> [--offset N]、fork <srcRunId> --from-offset N。传递 --storage file:<path> 或 --storage s3://<bucket>[/<prefix>]；@usekassette/s3 是一个软依赖，需要同时安装以使用 s3://。

开始

概念

生产环境就绪

参考

• agent-loop 一个最小化的持久思考-行动-观察循环。LLM 调用和工具执行被封装在 step() 中。崩溃后，重放跳过已完成的工作并在第一个未完成的步骤继续。
• loan-underwriting 并行数据收集与人工审批网关。parallel() 运行信用检查和房产评估。每个分支可以 suspend() 等待审核人签署。fork() 可以 re-

Make your agentic workflows durable with embeddable primitives that compose with your existing infra. No server, no dedicated runtime, no new infra.

Agentic workflows fail in a frustratingly expensive way: they run for minutes or hours, make costly LLM calls, execute tools with real-world side effects, and often pause for human input or external systems. Then the process crashes, times out, gets redeployed, or simply disappears. Since the state was in memory, naive retry means re-paying for answers the model already gave you, re-firing side effects that were supposed to happen once, and pushing recovery complexity back into app code.

kassette solves this with an append-only journal. It records completed work so that a replay fast-forwards through the journal on the next invocation and resumes live from the first unfinished step. Your existing queue or job runner can retry normally. kassette ensures those re-invocations are safe by replaying completed steps instead of running them again.

Use kassette when the problem is not "how do I run this again?" but "how do I avoid doing the same work twice?" Ie, your existing stack makes retries easy but not safe.

For the longer rationale, see Why kassette.

Embeddable, not a runtime. kassette is a library that runs inside your process, like SQLite. It's not a separate server or sidecar or worker that runs or schedules your code.

No new infrastructure. Composes with your existing infra. No database to host, no server to deploy, no queue to provision. Re-invocation is wired up using the queue, webhook, or job runner you already have. Uses the filesystem or object storage.

Plain-text state. Each run is a JSONL journal. Dump it, pipe it to jq, save it as a test fixture, or query S3-backed journals with Athena. See operations.

No compute while waiting. Waiting on human approval, a CI pipeline, or a webhook? The process simply exits. When the signal arrives, replay to the point you suspended and continue.

Serverless-native. The journal is the state; the process is a disposable. Any later process with access to the journal can continue the run. Copy it, sync it, ship it across regions.

What kassette ships, what you bring

Durable execution decomposes into two halves: a journal that records completed work, and a dispatcher that invokes the same runId again after a crash or timeout. Temporal and Inngest ship both. kassette ships only the journal.

You wire up the dispatcher that's already in your stack: a queue, job runner, or anything else that can run the same work again after failure. If a run doesn't succeed, the dispatcher invokes it again, and replay skips the steps that already finished. You don't need heartbeats, sweepers, or a run-status table.

Because the dispatch key and journal key are the same runId, retries are safe. kassette replays completed steps instead of running them again. If an old worker keeps running after a retry has taken over, its next journal write is rejected.

For concrete queue/job running wiring patterns, see Wiring the dispatcher.

• At-most-once journaling. A step result is recorded at most once. Later invocations replay the recorded value. Entries are append-only, ordered, and immutable. (A step fn may run more than once if its work finishes but its result is not journaled — eg crashes, fencing, or concurrent suspension — so side-effect idempotency remains the caller’s responsibility.)
• Journal authority. Recorded work stays settled. A retried resume keeps the first recorded event, even if a later retry passes a different payload. kassette never invalidates, expires, or garbage-collects a journal.
• Atomic appends. Each entry commits fully or not at all. Multi-entry operations like resume, fork, and auto-cancel are crash-recoverable. Backend details are covered in Local storage and Remote storage.
• Single-writer fencing. Only one session can append to a run. If an older process keeps running after a replacement starts, its next append fails with FencedError. Fencing is per run.
• Terminal-state protection. Once complete, error, or cancel lands, a later start/resume/fork on that run will throw TerminalRunError. Dispatchers can catch this and ack stale retries without re-running work.
• Deterministic replay. With the same journal and code, replay rebuilds the same state and resumes at the first unfinished step. Replay state comes only from the journal.
• Opt-in version check. Pass an optional version option to reject incompatible workflow code with VersionMismatchError.

kassette also throws on event-name collisions, changed start metadata, and mismatched parallel branch names on replay.

Just write a normal async function, wrapping anything you want to journal in ctx.step(). On replay, kassette will return the recorded result instead of running the step again. Use ctx.suspend() to pause the workflow and return control to the caller. When you resume, kassette will replay to that point and continue with the event payload.

There is no DSL, decorator system, or special control flow. if, while, try/catch, and early return work normally. Put the workflow inside a framework callback, middleware, or wherever your code already runs.

import { kassette, LocalStorage } from '@usekassette/kassette';

const storage = new LocalStorage('.kassette');

type Events = {
  'human-approval': { approved: boolean; notes?: string };
};

const agent = kassette<Ticket, Resolution, Events>(
  async (ctx, ticket) => {
    const analysis = await ctx.step('analyze', () => llm.chat('Diagnose this issue and recommend a fix', { ticket }));

    if (analysis.destructive) {
      const approval = await ctx.suspend('human-approval');
      if (!approval.approved) return { outcome: 'skipped', reason: approval.notes };
    }

    const result = await ctx.step('apply-fix', () => executeTool(analysis.suggestedAction));

    return { outcome: 'resolved', result };
  },
  { storage },
);

let result = await agent.start(ticket);
// → { status: 'suspended', event: 'human-approval', runId }

// ─── your process exits ───────────────────----------------------------

// resume from an external signal like a webhook, UI approval, or callback:
result = await agent.resume(result.runId, {
  eventName: 'human-approval',
  value: { approved: true },
});
// the llm call is replayed. same result, no new spend. execution continues
// from the approval point
// → { status: 'success', result: { outcome: 'resolved', ... }, runId }


npm install @usekassette/kassette

@usekassette/kassette — Workflow API: kassette(), Context, step, suspend, parallel, fork. @usekassette/core — Low-level durability primitives: start, record, waitForEvent, Storage. Use this directly if you want to embed durability into an existing agent loop without the workflow wrapper. @usekassette/s3 — S3 storage backend. @usekassette/cli — CLI for inspecting and forking journals. See CLI.

Zero runtime dependencies.

The Context passed to your workflow function:

Method	What it does
ctx.step(name, fn)	Runs and records a unit of work. On replay, returns the recorded result instead of running fn again. If fn throws, retries in-process according to { retry: { maxAttempts, delay, backoffRate, maxDelay } }, then propagates the error.
ctx.suspend(eventName, { timeout?, reason? })	Suspends the workflow and returns { status: 'suspended' } to the caller. resume delivers the event payload. Event payloads are typed with the TEvents generic. If timeout expires before resume, the run is canceled terminally.
ctx.parallel(branches)	Runs named branches concurrently. If any branch suspends, the whole parallel block suspends.
ctx.sleep(ms)	On replay, waits only for the remaining time.

The Kassette object returned by kassette():

Method	What it does
workflow.start(input)	Starts a workflow run and returns RunResult<TOutput>.
workflow.resume(runId, { eventName, value })	Delivers an event to a suspended run. Replays to the suspend point and then continues.
workflow.fork(source)	Creates a new run from part of an existing journal. The new run replays to the fork point, then diverges.

fork() copies part of an existing journal into a new run. The new run replays that prefix in memory, then continues live from the fork point.

Use this to branch one recorded prefix into several speculative continuations, or to go back to an earlier step after the original run went wrong. You keep the expensive work before the fork without paying for it again. See coding-agent.

Replay assumes the workflow code still has the same shape. If you add, remove, or reorder steps, old journal entries may map to the wrong code. This will matter for agent runs that suspend across deployments.

For incompatible changes to a step, rename the step. The new name will miss the old journal entry and run live.

// Before
const analysis = await ctx.step('analyze', () => llm.chat(prompt));

// After: incompatible prompt change
const analysis = await ctx.step('analyze-v2', () => llm.chat(newPrompt));

Renaming is not enough for every change. Reordering or removing steps can shift the journal mapping after that point. For those changes, configure version on the workflow definition as a safety check. A mismatch throws VersionMismatchError, so you can discard the journal and restart.

const agent = kassette<Input, Output, Events>(fn, {
  storage,
  version: 'workflow-v3',
});

const result = await agent.start(input);

When possible, tie version to the workflow shape, not the deploy SHA. Read more about Versioning.

LocalStorage stores each run as a JSONL file on disk.

• Atomic appends: each entry is written with write() and flushed with fsync(), no torn writes.
• Single-writer fencing: each runId has a lockfile created with atomic link(2). PID-based stale detection reclaims locks from dead processes.

@usekassette/s3 stores each run as one object in S3-compatible storage.

• Atomic appends: each PutObject writes the full journal atomically
• Single-writer fencing: each append uses conditional PutObject with If-Match on the previous etag. Two sessions may briefly coexist, but the older one is fenced on its next append.

Required backend semantics: read-after-write consistency with conditional writes. S3, R2, and GCS all support this.

Choose storage by where the run can resume, not by raw speed. For kassette’s target workload — 10-100 steps, small journal entries, and waits dominated by LLMs, tools, humans, webhooks, or timeouts — storage latency should not be the bottleneck.

LocalStorage appends are O(1): one fsync, ~100µs per entry. Use it when the retrying process can read the same persistent filesystem.

RemoteStorage / @usekassette/s3 stores each run as one object. Each append reads the current object and uploads the whole journal with one more line. This makes start, resume, and fork cheap because readAll is one object read, but write cost grows with the journal. Total uploaded bytes over a run are O(N²).

That tradeoff is usually fine for agent runs. For large entries, hundreds of steps, or multi-GB journals, prefer LocalStorage when possible.

kassette does not clean up, expire, or snapshot journals so storage will grow across runs. Use lifecycle policies or cleanup jobs if you need retention limits.

See Storage backends and Object storage design.

kassette journals are designed to be easy for agents to inspect. Agentic runs are usually small enough to inspect directly: tens of expensive steps, KB-MB journals, and one append-only JSONL log per run. A debugging agent can read the whole run, answer "what already happened?", "where did this suspend or fail?", and "where can I safely fork?" without a service console, database shell, snapshot protocol, or application-specific inspector.

This repo bundles an agent skill at skills/kassette/SKILL.md. Use it when an agent needs to inspect, debug, resume, or fork a kassette run. It covers storage discovery, journal structure, jq patterns, status interpretation, suspend/resume debugging, and safe forking.

@usekassette/cli ships a kassette command for inspecting and forking journals against either backend. list, status <runId>, dump <runId> [--offset N], fork <srcRunId> --from-offset N. Pass --storage file:<path> or --storage s3://<bucket>[/<prefix>]; @usekassette/s3 is a soft dep, install alongside for s3://.

Get started

Concepts

Going to production

Reference

• agent-loop A minimal durable think-act-observe loop. LLM calls and tool executions are wrapped in step(). After a crash, replay skips finished work and continues at the first unfinished step.
• loan-underwriting Parallel data gathering with human approval gates. parallel() runs the credit check and property appraisal. Each branch can suspend() for reviewer sign-off. fork() can re-run the final decision with earlier analysis replayed.
• deploy-assistant A webhook-driven deployment assistant. Uses suspend() for clarifying replies and production approvals, and sleep() while dispatched jobs take effect.
• coding-agent Speculative branching and backtracking with fork(). Reuse a recorded plan while trying multiple implementations, or backtrack to planning if none pass.
• vercel-ai-sdk Middleware for recording LLM calls through the Vercel AI SDK, including streaming responses. Replay returns the recorded response without calling the provider.
• cloudflare-worker A workflow running inside a Cloudflare Worker and journaling to R2. Any isolate or region with access to R2 can resume the run.
• cloudflare-queue Adds a queue in front of the Worker. Queue redelivery acts as the crash detector, a fresh isolate replays from R2 and continues live.

MIT