Skip to content

feat(shared): persist stream event log for cross-restart replay (#75) - #76

Open
sjr666666 wants to merge 16 commits into
helsome:mainfrom
sjr666666:feat/75-stream-event-persistence
Open

sjr666666 wants to merge 16 commits into
helsome:mainfrom
sjr666666:feat/75-stream-event-persistence

Conversation

@sjr666666

@sjr666666 sjr666666 commented Sep 11, 2026

Copy link
Copy Markdown
Collaborator

背景与动机

ADR 0001 把"事件日志持久化"留成了 open question #1#43 的 replay 数据源只有内存 StreamEventHistory,进程重启后历史即丢,replayStream 也就无法跨重启补发。本 PR 补上磁盘层(issue #75)。

完成了什么

  • packages/shared/src/kernel/stream-event-log.ts(新增):append-only JSONL(一行一条 StreamEvent,与内存缓冲并行落盘,放在 kernel store 旁的 stream-events.jsonl);load() 容忍坏行、按 (runId, sequence) 幂等丢弃重复/倒退行、只保留最近 maxRuns(默认 64)个 run,超限时收缩重写让磁盘占用有界;任何磁盘读/写失败都被吞掉并计入 failures(),实时事件链路永不阻塞(降级为内存-only)。
  • StreamEventHistory恢复与追加彻底分离 —— restore() 只把磁盘历史装回内存、绝不落盘;只有实时 append() 才写日志;同 run 内只接受严格递增 sequence,重复/倒退 seq 直接拒绝。recoverable / atEnd 契约不变。
  • RunManager 接线 streamLogDirAgentKernelstorageDir 传入,打包版与开发版行为一致。
  • 测试:stream-event-log.test.ts(往返、缺失目录、坏行跳过、maxRuns 收缩、写失败降级、连续 3 次重启不增长已有重复 seq 夹具)+ stream-history.test.ts(append 幂等)+ stream-replay.e2e.test.ts(含连续 3 次仓储重载的真实 runtime E2E)。

影响范围

未配置 streamLogDir 时行为与改动前完全一致(纯内存)。落盘失败不影响实时流(仅诊断计数)。

关联 Issue

Fixes #75(ADR 0001 open question #1 由本 PR 关闭)。依赖 #41#42#43

验证

环境:Bun 1.4.2 / Windows 11 (NT 10.0.26200)

CI(本 PR 的 checks):Typecheck / Focused tests / Full unit tests (advisory) / Secret scan。

本地(rebase 到最新 main 后在栈顶复跑)

bun test packages/shared/src/kernel --isolate   -> 93 passed / 0 failed(含 stream-event-log 8 例与 8 例 replay E2E)
bun test packages/shared --isolate              -> 967 passed / 0 failed
bun test packages/core packages/ui apps/electron --isolate -> 393 passed / 0 failed
bun run typecheck                               -> core/i18n/shared/ui/electron 全部 exit 0

跨重启相关用例(stream-replay.e2e.test.ts)在本地实际执行并全绿:

  • 连续 3 次仓储重载:只恢复不写回,文件不增长且 replay 保持连续
  • 仓储重载:磁盘日志恢复历史,replay 对重载前的 run 仍可用
  • 中途断线:按已收 lastSequence 补发剩余段,拼接与全量一致
  • 取消路径:显式 cancelled 事件入历史并可补发,atEnd=true
  • 带工具的 run:tool 事件入历史并可全量补发
  • 未知 run 与伪造游标:明确不可恢复

说明:以上是重载集成测试——同一临时目录下重建仓储 / RunManager 实例,验证磁盘恢复语义;它不是启动多个 OS 进程的跨进程 E2E。原"跨进程重启"表述已按评审意见更正为"仓储重载"。

评审修复说明

针对维护者 2026-09-17 的 blocker(StreamEventHistory 构造器把 log.load() 的历史又 append 回写同一个 JSONL,导致每次启动翻倍、第二次重启起 replay 连续性失败):

  1. 恢复与追加分离:新增 restore(),构造器只走内存装载,完全不触发磁盘写入;load() 的唯一写路径只剩超限 run 收缩。
  2. 重复 seq 处理:内存侧 push() 拒绝非严格递增 seq;磁盘侧 load() 按 (runId, sequence) 幂等去重并计入 duplicates,旧版本污染的文件可正常恢复。
  3. 回归测试:真实临时目录下连续 3 次新建 log/history(不发新事件)后文件行数不增长、从 0 与中间游标重放均连续、终态不变;并覆盖已有 [1,2,1,2] 重复 seq 夹具。

Rebase 说明

已随协议栈一起 clean rebase 到最新 main8d0ac3a),#41/#42/#43/#76 四个栈顶均已 force-push,零冲突。

本 PR 相对 #43 栈顶(e67969b)的 diff 仍为 persistence-onlystream-event-log.ts(新增)+ StreamEventHistory 的 log/persisted 接线 + RunManager.streamLogDir / AgentKernelstorageDir 传入 + 8 个 StreamEventLog 用例 + 跨重载 E2E,无协议或 renderer 改动。

@helsome helsome left a comment

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

持久化 replay 的测试方向够了,当前有两个 merge 前置项:

  1. 测试报告补齐 Bun 版本 + OS,并把 typecheck green across shared/ui/electron 改成实际执行命令与结果;现有 kernel 80 pass / cross-restart E2E 不需要扩大。
  2. 这条把 #41#42#43 的整套 stack 一起带进来了,而 #43 当前已经审核通过但仍在等基于最新 main 的 clean rebase。请先把协议栈整理到最新 main,再让 #76 只保留 persistence 的最终增量,避免重复历史/后续 squash 冲突。

不要求新增更多功能;重点是测试报告可复现 + 干净依赖历史。

@sjr666666
sjr666666 force-pushed the feat/75-stream-event-persistence branch from bc90a51 to d69964c Compare September 11, 2026 11:50
@sjr666666

Copy link
Copy Markdown
Collaborator Author

已处理两点评审意见:

  1. 测试报告可复现:PR 描述已更新,补 Bun v1.4.2 + OS(Windows 11 10.0.26200),并给出实际执行命令与结果(kernel 80 pass / 0 fail,shared/ui/electron typecheck 均 exit 0)。

  2. 干净依赖历史feat(core): add Stream Event Protocol v1 types (ADR 0001) #41feat(shared): add parallel Stream Event v1 channel to RunManager #42feat: wire Stream Event v1 transport and renderer log #43 整个协议栈已 rebase 到最新 main(f9309ac,含 feat: add canonical instrument identity and symbol resolution #59/fix(tests): 修复 Windows 上的单元测试失败(路径断言/symlink 权限/locale 依赖) #70/feat: 接入 Langfuse 追踪与评测写回 (#14) #74),三个 PR 分支均已 force-push(13 个 commit 重放无冲突)。feat(shared): persist stream event log for cross-restart replay (#75) #76 已 --onto 重放到新栈顶端,只保留 persistence 的最终增量。等 feat: wire Stream Event v1 transport and renderer log #43 合并后,feat(shared): persist stream event log for cross-restart replay (#75) #76 的 diff 会自动收窄为纯 persistence 改动,不产生重复历史或 squash 冲突。

helsome commented Sep 11, 2026

Copy link
Copy Markdown
Owner

复查后,上一轮第 1 个 blocker 已解除:PR 描述现在包含 Bun 1.4.2、Windows 平台、实际 kernel test/typecheck 命令与结果,d69964c 的 PR checks 也已 success。

当前只剩依赖历史这一项。因为 #81 刚合入 main,#43 stack 又需要基于 6a9a288 做一次 clean rebase。请先保持 #76 在新的 #43 栈顶之上;等 #43 合入后,再把 #76 重放成只含 StreamEventLog persistence 的最终 delta。无需新增功能或扩大测试范围。

@sjr666666
sjr666666 force-pushed the feat/75-stream-event-persistence branch 2 times, most recently from 0a62cd5 to db80eb0 Compare September 13, 2026 13:05
@sjr666666

Copy link
Copy Markdown
Collaborator Author

已按你的要求保持 #76 在新的 #43 栈顶之上:

  • #41 → #42 → #43 栈已 clean rebase 到最新 main046b5d3);
  • #76 随之重放到新的 feat: wire Stream Event v1 transport and renderer log #43 顶部(eb75fc7),当前只含 persistence delta:StreamEventLog + StreamEventHistory 的 log/persisted 接线 + RunManager.streamLogDir / AgentKernelstorageDir 传入 + 5 个 StreamEventLog 用例与跨重启 E2E,没有夹带协议或 renderer 改动;
  • 未新增功能、未扩大测试范围;PR checks 全部 pass(Typecheck / Focused tests / Full unit tests (advisory) / Secret scan)。

#43 合入后,本 PR 的 diff 会自动收窄为纯 persistence 改动,不产生重复历史或 squash 冲突。

@sjr666666
sjr666666 force-pushed the feat/75-stream-event-persistence branch from db80eb0 to 418d0b8 Compare September 16, 2026 18:33
@sjr666666

Copy link
Copy Markdown
Collaborator Author

已按你的两项前置要求执行完毕:

1. 测试报告可复现#76 描述已用实际执行命令 + pass/fail 重写(Bun 1.4.2 / Windows 11 NT 10.0.26200)。这次本地依赖按 --linker=isolated 装全,因此在 rebase 后的新栈顶做了完整复跑,不再是"仅供参考"的旧结果:

bun test packages/shared/src/kernel --isolate  -> 84 passed / 0 failed
bun test packages/shared --isolate             -> 936 passed / 0 failed
bun test packages/core --isolate               -> 49 passed / 0 failed
bun test packages/ui --isolate                 -> 305 passed / 0 failed
bun test apps/electron --isolate               -> 29 passed / 0 failed
bun run typecheck                              -> core/i18n/shared/ui/electron 全部 exit 0

跨重启 E2E 的 5 条用例(含"重启后 replay 对旧 run 仍可用")在本地实际执行且全绿,未扩大测试范围。如实说明:受限于受限环境时 disposeSession 用例曾因 /tmp 不可写报 EPERM(非断言失败、与持久化无关),换到有写权限的同一工作区复跑即 936/0。

2. 干净的依赖历史#41#42#43 已 clean rebase 到最新 main6405b7b,零冲突),#76 随之重放到新栈顶:

本 PR 相对 #43 栈顶仍是 persistence-only:8 文件 / +282 −4(stream-event-log.ts 新增 + StreamEventHistory 接线 + RunManager.streamLogDir / AgentKernel 传参 + 5 个 StreamEventLog 用例与跨重启 E2E),没有夹带协议或 renderer 改动。

PR checks 在新 commit 上全部 pass(Typecheck / Focused tests / Full unit tests (advisory) / Secret scan),显示 MERGEABLE / CLEAN。等 #43 合入后本 PR 的 diff 会自动收窄为纯 persistence 改动。方便的话请帮忙复查一下,把之前那条 Changes Requested 收敛掉。

@helsome
helsome dismissed their stale review September 17, 2026 03:53

当前 head 已提供 Bun/OS、实际 focused/workspace/typecheck 结果,并整理为 #43 之上的 persistence-only 增量。原测试报告/stack 说明 blocker 已解决;本轮发现的恢复写回缺陷另行给出精确 review,不再沿用旧 blocker。

@helsome helsome left a comment

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

测试报告和 clean stack 的旧 blocker 已解除。审查当前栈顶发现一个实际持久化/replay blocker,只需修这一处,不要求性能专项或新存储框架:

StreamEventHistory 构造器先设置 this.log,随后对每条 persisted event 调 this.append(event);而 append 又调用 this.log?.append(event)。RunManager 同时传入 loglog.load().events,因此每次启动都会把刚读出的历史再写回同一个 JSONL。

最小场景:磁盘为 seq [1,2] → 第一次恢复后磁盘变成 [1,2,1,2] → 第二次恢复读入重复序列,replay(runId,0) 的 contiguous 检查返回 false;磁盘还会继续翻倍。这不是优化问题,而是第二次重启就破坏跨重启恢复。

请把“加载到内存”与“追加新事件并落盘”分离,恢复不触发写回;并处理/拒绝已有重复记录,避免恢复出的重复 seq 被当作新事件。补真实临时目录的 focused 回归:连续至少 3 次新建 log/history(不发新事件)后,文件行数不增加、从 0/中间游标重放均连续、终态不变;已有 seq 重复夹具也要覆盖。现有 kernel/typecheck 重跑并更新报告即可,不需要真实模型/金融 API。

另外请将“跨进程 E2E”准确标注为实际执行形式:只重建仓储/RunManager 实例的测试是重载集成测试,不等于启动了多个 OS 进程。测试本身有价值,这点不扩大验收范围。

石敬荣 added 16 commits September 17, 2026 13:49
Introduce the pure-type foundation for the structured streaming event protocol (issue helsome#27): versioned envelope, 12 typed events, and the idempotency/cancel/reconnect contracts as types. Zero runtime change.

Adds ADR 0001 documenting context, decision, migration path and open questions for maintainer review.
Emit Stream Event Protocol v1 events alongside the existing AgentEvent stream (issue helsome#27, ADR 0001 migration step 2). Adds toStreamEvents mapping (8 AgentEvent types -> 12 protocol events, cancel normalized to 'cancelled') and RunManager.subscribeStream. Existing AgentEvent consumers are untouched; the parallel channel only activates when a stream subscriber is registered.
StreamEvent was a single indexed-union instantiation (Tagged StreamEventEnvelope<StreamEventType>), so payload could not be narrowed by type in switch/if. Rewrite as a distributive mapped union; on-disk type shape is unchanged. Adjusts adapter unit-test helper accordingly.
KernelHost subscribes RunManager.subscribeStream (issue helsome#27) and forwards { sessionId, event } over IPC channel 'agent:stream'; preload exposes electronAPI.kernel.onStreamEvent. Legacy 'agent:event' delivery untouched. Transport only; renderer consumption follows.
Renderer-side data layer for issue helsome#27: KernelBridge subscribes client.kernel.onStreamEvent into a parallel StreamEvent log. reduceStreamLog keeps per-run events ordered by sequence, dedupes replays (drops), and flags gaps/out-of-order (anomalies) as a protocol health signal. FinagentClient and preload.cjs wire onStreamEvent; FinagentClient adds onStreamEvent contract with fallback noop. No visual change: existing AgentEvent rendering untouched.
Addresses helsome#43 review: messageId is no longer fused with runId (issue helsome#34 - message/generation/run are distinct identities). message-level events carry the real assistant messageId, pre-assigned per run in RunManager; run-level events omit it; idempotency key is now runId + sequence.

Adds StreamEventHistory: in-memory per-run tail used by RunManager.replayStream(runId, lastSequence) for reconnect resume, with an explicit recoverable:false path when the run is unknown or the tail is non-contiguous (eviction).
Adds 'runs:stream-replay' handler (KernelHost.streamReplay -> RunManager.replayStream), preload (ts + cjs) streamReplay, finagentClient wiring and FinagentClient contract with fallback noop.
…plementation

Envelope messageId is now optional (message-level events only); idempotency key runId+sequence; Reconnect row documents implemented StreamEventHistory + IPC with explicit unrecoverable path; open question 3 resolved.
run_started(seq 1)与 runtime 自产的首个事件(同样 seq 1)冲突,
导致 replay(runId, 0) 被判为不连续(recoverable: false),renderer 的
幂等去重也会误丢事件。RunManager 现为每个 run 持有 RunProtocol 计数器,
扇出前把所有 AgentEvent 统一重排为 1..N;messageId 随该对象传递,
崩溃/取消兜底合成的 terminal 事件也保住 helsome#34 的身份契约。replay 游标
超出已知最大 sequence 时,由静默视为已同步改为明确不可恢复。
kernel 级(stream-replay.e2e.test.ts):真实 LocalRuntimeAdapter +
RunManager + 持久化。六个用例:完整 run 后全量补发、无实时订阅者时
事件仍入历史、按 lastSequence 中途断线补发、取消路径产生可补发的显式
cancelled 事件、带工具 run 的全量流、未知 run 与伪造游标返回明确的
不可恢复路径。

app 级(e2e/stream-replay.mjs):真实 Electron + preload IPC(CDP)。
四个用例:replay(runId, 0) 与实时投递逐字节一致、断线补发拼接还原
完整流、未知 run 跨 IPC 返回不可恢复、非法 lastSequence 被拒以
INVALID_ARGUMENT。
AgentKernelHost.attach wires the Stream Event v1 channel through
kernel.runs.subscribeStream; the fake kernel in kernelHost.test.ts lacked
the method, so the transport test crashed with TypeError before asserting.
Align the fake with the kernel surface (subscribeStream + replayStream)
so focused CI runs green again.
The Stream Event v1 IPC surface (streamReplay / onStreamEvent) landed in
the kernel channel types; the test kernel client was not updated, which
broke the ui + i18n + electron typecheck gates. Align the stub with the
channel so typecheck is green again.
streamReplay returned an inferred StreamReplayResult that TS could not
name portably across the @finagent/shared boundary (error TS2742). Export
the type from the shared package and annotate the kernelHost surface so
every package typecheck gate passes.
- stream-event-adapter:run_failed 按 error.code 归一为 cancelled(user/budget/runtime),
  并把已生成的 partial 文本带进 cancelled.partial.text(对齐 ADR 0001)
- run-manager:run 全程保留 assistant 文本快照,供 cancelled 事件使用
- kernelHost:streamReplay 游标限定为非负整数;dispose 时一并清理 stream 订阅
- ui/streamAtoms:幂等去重仅在 sequence 回退时扫描,避免长 run 下的 O(n²)
- 补齐 adapter / streamAtoms / kernelHost 边界测试,electron E2E 增加非法游标用例
…ome#75)

ADR 0001 kept replay data in memory (32 runs / 2000 events), so a restart
or buffer eviction silently destroyed the reconnect data source. Add an
append-only JSONL log under the kernel store dir:

- StreamEventLog: synchronous append + tolerant load (skips bad lines),
  bounded by maxRuns (compaction rewrites the file to stay bounded);
  any disk failure is swallowed and only counted, so the live stream is
  never blocked (in-memory fallback).
- StreamEventHistory accepts a log and persisted events; RunManager wires
  streamLogDir and restores history on boot, keeping the existing
  recoverable/atEnd contract intact.
- AgentKernel supplies streamLogDir from its storageDir.
修复 helsome#76 评审指出的持久化回写缺陷:StreamEventHistory 构造器把
log.load() 恢复出的历史又逐条 append 回同一个 JSONL,导致每次启动文件
翻倍([1,2] -> [1,2,1,2]),第二次重启起 replay 连续性检查失败。

- 恢复与追加分离:restore() 只装入内存、绝不落盘;append() 才写日志;
- 幂等去重:同 run 内只接受严格递增 sequence,重复/倒退 seq 直接拒绝,
  磁盘上已有的重复行在 load() 时按 (runId, sequence) 丢弃并计入
  duplicates;旧版本污染的文件因此可自愈;
- load() 写路径仅剩超限 run 收缩,恢复历史不再触发任何写入;
- 回归测试(真实临时目录):stream-event-log.test.ts 覆盖连续 3 次重启
  文件行数不增长、从 0/中间游标重放均连续且终态不变,以及已有
  [1,2,1,2] 重复 seq 夹具的幂等恢复;stream-replay.e2e.test.ts 覆盖
  连续 3 次仓储重载并走真实 runtime;
- 措辞:把“跨进程重启”改为“仓储重载”,注明这是重建仓储/RunManager
  实例的重载集成测试,不等于启动多个 OS 进程。
@sjr666666
sjr666666 force-pushed the feat/75-stream-event-persistence branch from 418d0b8 to ba8dc3e Compare September 17, 2026 06:00
@sjr666666

Copy link
Copy Markdown
Collaborator Author

已按评审意见完成修复,并把整个协议栈 clean rebase 到最新 main8d0ac3a):

1. 持久化回写 blocker(已修)

  • StreamEventHistory 的恢复与追加彻底分离:新增 restore(),构造器只把 log.load() 的历史装入内存、不触发任何磁盘写入;只有实时 append() 才落盘。load() 的唯一写路径只剩超限 run 收缩。
  • 重复 seq 处理:内存侧 push() 只接受同 run 严格递增的 sequence,重复/倒退直接拒绝;磁盘侧 load()(runId, sequence) 幂等去重并计入新增的 duplicates,旧版本污染过的 [1,2,1,2] 文件可正常恢复。
  • 回归测试(真实临时目录):
    • stream-event-log.test.ts:连续 3 次新建 log/history(不发新事件)后文件行数不增长、从 0 与中间游标重放均连续、终态不变;另有已有重复 seq 夹具。
    • stream-replay.e2e.test.ts:连续 3 次仓储重载同上,并走真实 runtime 全链路。

2. 措辞(已改)
把"跨进程重启"改为"仓储重载",并在用例处注明这是重建仓储/RunManager 实例的重载集成测试,不等于启动多个 OS 进程。

本地复跑(Bun 1.4.2 / Windows 11 NT 10.0.26200):

bun test packages/shared/src/kernel --isolate                      -> 93 passed / 0 failed
bun test packages/shared --isolate                                 -> 967 passed / 0 failed
bun test packages/core packages/ui apps/electron --isolate         -> 393 passed / 0 failed
bun test --isolate                                                 -> 1484 passed / 8 skipped / 0 failed
bun run typecheck                                                  -> core/i18n/shared/ui/electron 全部 exit 0

Rebase#41/#42/#43/#76 均已 force-push 到 main 之上,四个 PR 现在都是 MERGEABLE。本 PR 相对 #43 栈顶仍是 persistence-only,无协议或 renderer 改动。

补充说明:本次 CI 的 Full unit tests (advisory) 出现了 ExperimentService 成串 5s 超时(该 job 为 continue-on-error,不阻塞合并)。本地全量单测 1484/0 全绿,已重跑该 job 复核,判断为 CI 环境 flake。

方便时请复查,帮忙把这条 Changes Requested 收敛掉。

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Reliability] Persist stream event log so replay survives restarts

2 participants