From 5a1a994cc20281c62561187589b9a752d81e16de Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 29 Jun 2026 12:38:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(hitl):=20=E5=A2=9E=E9=87=8F2a=20=E2=80=94?= =?UTF-8?q?=E2=80=94=20board=20=E5=8F=AF=E5=BA=8F=E5=88=97=E5=8C=96?= =?UTF-8?q?=EF=BC=88compose=20checkpoint=20=E7=9A=84=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E8=BD=BD=E4=BD=93=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit compose checkpoint 把图执行态（含 GenLocalState 的 board）序列化进 store，但 Eino 序列化器只认导出字段，而 board 字段全未导出 → 直接持久化会落成空、resume 丢全部状态。Eino 对同时实现 json.Marshaler+Unmarshaler 的类型改走自定义 JSON （internal/serialization checkMarshaler），故给 *board 实现一对 JSON 方法映射到 导出 DTO，无需把 board 字段全导出（牵连几十处调用点）。 - board_serde.go: *board 的 MarshalJSON/UnmarshalJSON ↔ boardSnapshot；丢弃 fatalErr（transient error，中断点必为 nil）；schema.RegisterName[*board] 注册 类型名供 checkpoint 的 State(any) 还原。 - 测试: 快照往返无损 + 编译期断言实现 json.Marshaler/Unmarshaler + fatalErr 不被持久化。 零行为变更（纯新增）。go test ./... 全绿。 下一步 2b：审批节点 compose.Interrupt + 编译挂 checkpoint store + orchestrator 识别 InterruptInfo 置 waiting 并释放 goroutine。 Co-Authored-By: Claude Opus 4.8 --- .../internal/eino/board_serde.go | 75 +++++++++++++++++++ .../internal/eino/board_serde_test.go | 64 ++++++++++++++++ 2 files changed, 139 insertions(+) create mode 100644 sundynix-dispatcher/internal/eino/board_serde.go create mode 100644 sundynix-dispatcher/internal/eino/board_serde_test.go diff --git a/sundynix-dispatcher/internal/eino/board_serde.go b/sundynix-dispatcher/internal/eino/board_serde.go new file mode 100644 index 0000000..8bba6d6 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/board_serde.go @@ -0,0 +1,75 @@ +package eino + +import ( + "encoding/json" + + "github.com/cloudwego/eino/schema" +) + +// board 的字段全是未导出的，而 Eino 的 checkpoint 序列化器只认导出字段——若直接让 compose +// 持久化 board，会落成空、resume 丢全部执行态。Eino 对「同时实现 json.Marshaler+Unmarshaler」 +// 的类型改走自定义 JSON（见 internal/serialization checkMarshaler），故给 *board 实现一对 JSON +// 方法、映射到导出 DTO，即可把整个黑板按需持久化，而无需把 board 字段全导出（牵连几十处调用点）。 +// +// 不持久化 fatalErr（error，transient——中断点处必为 nil）：审批中断时图尚未失败， +// 它是单次 Invoke 内的致命错标志，跨 resume 无意义。rejected 同理在中断点为 false， +// 但仍带上以求快照完整、零歧义。 +type boardSnapshot struct { + UID string `json:"uid,omitempty"` + SID string `json:"sid,omitempty"` + Query string `json:"query,omitempty"` + Profile string `json:"profile,omitempty"` + History []*schema.Message `json:"history,omitempty"` + KB string `json:"kb,omitempty"` + Refs []string `json:"refs,omitempty"` + ToolOut []string `json:"tool_out,omitempty"` + Sections []reportSection `json:"sections,omitempty"` + Answer string `json:"answer,omitempty"` + AgentOut []string `json:"agent_out,omitempty"` + Rejected bool `json:"rejected,omitempty"` +} + +// MarshalJSON 把黑板序列化为快照（Eino checkpoint 经此持久化整图执行态）。 +func (b *board) MarshalJSON() ([]byte, error) { + return json.Marshal(boardSnapshot{ + UID: b.uid, + SID: b.sid, + Query: b.query, + Profile: b.profile, + History: b.history, + KB: b.kb, + Refs: b.refs, + ToolOut: b.toolOut, + Sections: b.sections, + Answer: b.answer, + AgentOut: b.agentOut, + Rejected: b.rejected, + }) +} + +// UnmarshalJSON 从快照还原黑板（resume 时 Eino 据此重建中断前的执行态）。 +func (b *board) UnmarshalJSON(data []byte) error { + var s boardSnapshot + if err := json.Unmarshal(data, &s); err != nil { + return err + } + b.uid = s.UID + b.sid = s.SID + b.query = s.Query + b.profile = s.Profile + b.history = s.History + b.kb = s.KB + b.refs = s.Refs + b.toolOut = s.ToolOut + b.sections = s.Sections + b.answer = s.Answer + b.agentOut = s.AgentOut + b.rejected = s.Rejected + return nil +} + +// 注册 *board 的序列化类型名：Eino checkpoint 的 State 是 any，持久化时记类型名、 +// resume 时据名还原具体类型再走 UnmarshalJSON。名字一经上线不可改（旧 checkpoint 据此解码）。 +func init() { + schema.RegisterName[*board]("sundynix_board") +} diff --git a/sundynix-dispatcher/internal/eino/board_serde_test.go b/sundynix-dispatcher/internal/eino/board_serde_test.go new file mode 100644 index 0000000..617e747 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/board_serde_test.go @@ -0,0 +1,64 @@ +package eino + +import ( + "encoding/json" + "reflect" + "testing" + + "github.com/cloudwego/eino/schema" +) + +// board 必须同时实现 json.Marshaler+Unmarshaler，Eino checkpoint 才会走自定义 JSON 路径 +// 持久化未导出字段；任一缺失都会让序列化器回退到「只存导出字段」=空 board。 +var ( + _ json.Marshaler = (*board)(nil) + _ json.Unmarshaler = (*board)(nil) +) + +// TestBoardSnapshotRoundtrip 钉死黑板快照往返：中断点的全部执行态都能无损还原（resume 不丢状态）。 +func TestBoardSnapshotRoundtrip(t *testing.T) { + orig := &board{ + uid: "u42", + sid: "s7", + query: "介绍杭州西湖", + profile: "用户偏好简洁", + history: []*schema.Message{schema.UserMessage("上一轮"), schema.AssistantMessage("上一轮回答", nil)}, + kb: "u42/travel", + refs: []string{"西湖十景…", "苏堤春晓…"}, + toolOut: []string{"[wiki] 杭州…"}, + sections: []reportSection{ + {Heading: "概况", Body: "西湖位于杭州。"}, + }, + answer: "西湖是…", + agentOut: []string{"草稿一", "草稿二"}, + // fatalErr 故意置非空：快照应丢弃它（transient），不应影响往返。 + fatalErr: errRejected, + } + + data, err := json.Marshal(orig) + if err != nil { + t.Fatalf("marshal: %v", err) + } + + var got board + if err := json.Unmarshal(data, &got); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + // 持久化字段逐一比对。 + if got.uid != orig.uid || got.sid != orig.sid || got.query != orig.query || + got.profile != orig.profile || got.kb != orig.kb || got.answer != orig.answer { + t.Fatalf("标量字段未还原: %+v", got) + } + if !reflect.DeepEqual(got.refs, orig.refs) || !reflect.DeepEqual(got.toolOut, orig.toolOut) || + !reflect.DeepEqual(got.agentOut, orig.agentOut) || !reflect.DeepEqual(got.sections, orig.sections) { + t.Fatalf("切片字段未还原: %+v", got) + } + if len(got.history) != len(orig.history) || got.history[0].Content != "上一轮" { + t.Fatalf("history 未还原: %+v", got.history) + } + // fatalErr 是 transient，快照不持久化 → 还原后必为 nil。 + if got.fatalErr != nil { + t.Fatalf("fatalErr 不应被持久化，got %v", got.fatalErr) + } +}