English | 简体中文

Replay

让每一次事故 (incident) 都逐字节可复现。一个面向受治理事件流的确定性测试台 (determinism harness) —— 它冻结时钟、随机性与外部 IO,使处理器 (handler) 回放 (replay) 出逐字节一致的转录 (transcript)。

Octopus Core 的一部分 —— 受治理 AI 的开源基础设施栈。 每个仓库只做一件事，沿 agent 生命周期组合：Scout · Observe · Experience · Blackboard · Runtime · Replay —— Inspect 横贯每一环做治理。

本仓库 —— Replay · 验证： 逐字节复现每一次 agent 事故。

Scenario (日志 / fixtures) → record → Recording ── replay ──▶ Transcript
                                          └────── verify ─────▶ 通过 / 分歧

从一次真实事故中捕获一份黄金录制 (recording) —— 事件流,加上处理器从时钟、 RNG 与网络读到的每一个值。在 CI 中重跑它,任何改变受治理行为的变更都会失败,并给出 指向第一处不同输出的精确定位。相同录制、相同处理器 → 相同的转录哈希,在任何机器 上、永远如此。

边界 (Boundaries)

Replay 是一个确定性测试台,而非 JSONL 读取器。日志适配器只是便利;本质在于它 冻结每一个不确定性来源。一个只经由 ReplayContext 读取时钟、随机性与外部 IO 的处理器会精确回放。一个直接去调用 Date.now()、Math.random() 或真实网络的处理器 则逃离了测试台,并会产生分歧 (diverge) —— 这份纪律正是全部要点所在。

Replay 不会替换全局对象、拦截 fetch,也不会为你的代码提供沙箱。它不判断某个 输出是否正确 —— 只判断它是否可复现。它宿主 (host) 一个处理器;它不是该处理器的 依赖。

它通过让处理器包裹受治理核心 (Observe / Runtime / Blackboard) 并回放事故来验证 它们 —— 但它对这些核心零依赖。边界是 Handler 签名与 JSON 事件形态,而非任何 包的 SDK。Replay 是测试与证明 (proof) 基础设施。

它具有零第三方依赖:唯一的运行时依赖是一方的 octopus-evidence 原语(其本身零依赖),它提供全栈共享的规范化哈希。除此之外本仓库可完全独立使用。

安装与构建 (Install & build)

npm install
npm run typecheck   # tsc --noEmit
npm test            # node --test
npm run build       # emit dist/
npm run example     # 捕获一份录制并校验它(见下)

需要 Node ≥ 22。

快速上手 (Quickstart)

处理器是事件及其上下文的确定性函数:

import type { Handler } from "octopus-replay";

// 一个微型受治理处理器:盖一个时间戳、做一次策略掷点、执行一次 IO 调用 ——
// 每一个不确定性来源都只经由 `ctx` 读取。
const handler: Handler = async (event, ctx) => {
  const receivedAt = ctx.now();               // 冻结的时钟
  const sampled = ctx.random() < 0.2;         // 冻结的 RNG(20% 抽样)
  const balance = await ctx.respond(          // 冻结的 IO:record 时录带,
    "account.balance",                         // replay 时回放
    () => 1000 - (event as { amount: number }).amount,
  );
  return { receivedAt, sampled, approved: balance >= 0, balance };
};

构建一个 scenario、捕获一份黄金录制,然后校验:

import {
  fromEvents,
  record,
  seededRandom,
  steppingClock,
  verifyDeterminism,
} from "octopus-replay";

const scenario = fromEvents([{ amount: 300 }, { amount: 1000 }, { amount: 50 }], {
  id: "incident-4211",
  note: "three charges captured from production",
});

// 在确定性来源下捕获一份黄金录制。
const recording = await record(scenario, handler, {
  sources: { clock: steppingClock(Date.parse("2026-07-03T00:00:00Z"), 1000), random: seededRandom(99) },
});

// 相同处理器 → 逐字节可复现。
const good = await verifyDeterminism(recording, handler);
good.ok; // true —— actualHash === expectedHash

当行为漂移 (drift) 时,同一次调用会捕获它,并指向第一处分歧:

// 一个回归:有人把审批边界翻成了 `> 0`。
const changed: Handler = async (event, ctx) => {
  const receivedAt = ctx.now();
  const sampled = ctx.random() < 0.2;
  const balance = await ctx.respond("account.balance", () => 1000 - (event as { amount: number }).amount);
  return { receivedAt, sampled, approved: balance > 0, balance }; // >= 0 变成了 > 0
};

const bad = await verifyDeterminism(recording, changed);
bad.ok;                    // false
bad.divergence?.seq;       // 1 —— { amount: 1000 } 那笔,余额为 0
bad.divergence?.expected;  // { …, approved: true,  … }
bad.divergence?.actual;    // { …, approved: false, … }

这就是端到端的 examples/replay-incident.ts; 用 npm run example 运行它。

在 CI 中使用

把一份录制作为 fixture 提交,让 CI 捕获行为漂移:

1. 录制一次 —— 从一次真实事故(或手工构造的 scenario)用 record 录制,并用 serializeRecording 写成一个提交进仓库的 *.json fixture。
2. 每次变更都校验 —— 在 CI 中运行 octopus-replay verify recording.json --handler ./handler.js,或在一个 node --test 文件里用 verifyDeterminism(recording, handler) 断言 result.ok。

任何改变受治理输出的变更都会让 verify 失败,并给出确切的事件与期望/实际值 —— 在你 捕获事故的那一刻就免费获得的一个回归测试。

裸 (bare) 与冻结 (frozen) scenario

裸 scenario 只有事件 —— 源适配器(fromEvents、fromJsonl、 fromWebhookFixtures)返回的就是它。record 会用真实来源(默认 Date.now / Math.random,或你注入的 sources)运行它,并把处理器消费的每个值录带下来, 产出一份完全冻结的录制。

冻结 scenario 已携带自己的 clock、random 与 responses 带子(例如 toScenario(recording))。record 会从它自己的带子重新物化 (re-materialize) —— 不触碰任何真实来源。replay 始终以冻结方式运行:它是纯的,不触碰真实时钟、RNG 或 网络。

ReplayContext 契约

处理器所需的一切不确定性都必须经由上下文:

| 成员 | record 时 | replay 时 | | ---- | --------- | --------- | | ctx.now() | 读取真实/注入的时钟并录带该值 | 返回下一个录带的时间戳 | | ctx.random() | 读取真实/注入的 RNG 并录带该值 | 返回下一个录带的值 | | ctx.respond(key, produce) | 运行 produce() 并把其结果按 key 录带 | 返回录带结果 —— produce 不会被调用 | | ctx.seq | 当前事件的 0 基下标 | 同左 |

replay 时 produce 永不运行 —— 因此一个在生产中会失败、有副作用或很慢的 IO 调用, 在回放时是无操作 (no-op)。若处理器过度读取带子(now()/random() 调用次数超过录制, 或对一个已耗尽的 key 调用 respond),冻结上下文会抛出 DeterminismError, verifyDeterminism 会把它呈现为一处分歧,而非直接抛出。这正是测试台在告诉你:行为 以录制无法复现的方式漂移了。

源适配器 (Source adapters)

把捕获的生产数据转成裸 scenario:

import { fromJsonl, fromEvents, fromWebhookFixtures } from "octopus-replay";

// NDJSON / JSONL 事件日志(每行一个事件):
const a = fromJsonl(logText, { select: (line) => (line as { event: unknown }).event });

// 一个普通事件数组:
const b = fromEvents([{ amount: 300 }, { amount: 50 }]);

// 捕获的 webhook 投递 —— 按 `receivedAt` 确定性排序:
const c = fromWebhookFixtures(fixtures, { wrap: true }); // wrap → { id?, headers?, payload }

确定性辅助 (Determinism helpers)

用于可复现地驱动 record(在测试、示例与 CLI 中):

• seededRandom(seed) —— 一个 mulberry32 PRNG;相同种子产出相同序列,独立于平台 RNG。
• steppingClock(start, step = 1000) —— 一个从 start 起、每次调用步进 step 毫秒 的时钟。
• sequenceClock(times) —— 依次返回 times 中的每个时间戳,耗尽时抛错,使过度读取的 测试大声失败。

规范哈希 (Canonical hashing)

"逐字节"是针对规范形态定义的,因此键序、空白与数字格式绝不影响相等性:

• stableStringify(value) —— 每一层键都排序的确定性 JSON。拒绝 JSON 无法忠实往返的 内容:非有限数值、undefined、函数与环 (cycle) —— 一次"成功"的哈希绝不掩盖静默的 数据丢失。
• canonicalHash(value) —— 该编码的 SHA-256(即转录哈希)。
• canonicalEqual(a, b) —— 经由同一编码的深度相等,因此 diff 绝不会与哈希不匹配相 矛盾。
• cloneJson(value) —— 一次深克隆,同时断言该值为可规范化的 JSON。

录制格式 (Recording format)

一份录制序列化为美化打印的 JSON(RECORDING_VERSION = "1"),适合作为 fixture 提交:

{
  "meta": { "recordingVersion": "1", "id": "incident-4211", "createdAt": "2026-07-03T00:00:00.000Z", "note": "…" },
  "events": [ { "seq": 0, "event": { "amount": 300 } }, … ],
  "clock":  [ 1751500800000, 1751500801000, 1751500802000 ],   // ctx.now()  的值,按调用顺序
  "random": [ 0.63…, 0.11…, 0.94… ],                            // ctx.random() 的值,按调用顺序
  "responses": { "account.balance": [ 700, 0, 950 ] },          // ctx.respond() 的结果,按 key、按顺序
  "transcript": {
    "entries": [ { "seq": 0, "output": { "approved": true, … } }, … ],
    "hash": "a1b2c3…"                                            // 覆盖规范化 entries 的 SHA-256
  }
}

serializeRecording / parseRecording / assertRecording 分别写入、读取并校验 该形态;畸形文件(版本错误、字段类型错误)会抛出 RecordingFormatError。

CLI

octopus-replay record <scenario> --handler <mod> [-o out.json] [--seed N] [--clock-start MS] [--clock-step MS]
octopus-replay verify <recording.json> --handler <mod>
octopus-replay run    <recording.json> --handler <mod>
octopus-replay diff   <a.json> <b.json>

• record —— 在一个 scenario 上运行处理器,并打印(或 -o)一份黄金录制。 scenario 输入可以是 .jsonl/.ndjson 日志、.json 事件数组,或一个 scenario 对象。--seed 确定性地驱动 ctx.random();--clock-start/--clock-step 驱动 ctx.now()。
• verify —— 回放一份录制并检查它是否仍然匹配;打印 ✔ 或第一处分歧。
• run —— 回放一份录制并打印所得的转录 JSON。
• diff —— 比较两份录制的转录。

--handler <mod> 模块默认导出 (default-export) (event, ctx) => output 函数(命名导出 handler 亦可)。相对/绝对路径相对 cwd 解析;裸标识符按 node 模块 解析。

退出码: 0 正常 · 1 分歧 / 校验失败 · 2 用法或 IO 错误。

API 一览

• 测试台 (Harness) —— record(scenario, handler, opts?)、 replay(recording, handler)、toScenario(recording)。
• 校验 (Verify) —— verifyDeterminism(recording, handler) → { ok, expectedHash, actualHash, divergence? }、diffTranscripts、 diffRecordings、formatDivergence。
• 源 (Sources) —— fromJsonl、fromEvents、fromWebhookFixtures。
• 确定性 (Determinism) —— seededRandom、steppingClock、sequenceClock。
• 哈希 (Hashing) —— stableStringify、canonicalHash、canonicalEqual、 cloneJson。
• 格式 (Format) —— serializeRecording、parseRecording、assertRecording、 RECORDING_VERSION。
• 错误 (Errors) —— DeterminismError、RecordingFormatError。

设计 (Design)

权威的架构与契约文档位于 docs/DESIGN.zh-CN.md —— 确定性测试台论点、Scenario → record → Recording → replay/verify 流水线、 ReplayContext 纪律,以及录制格式。在做出更改之前请先阅读它;代码是依照该规范编写的。

许可证 (License)

Apache-2.0 © Octoryn。