From 88d7dd002691abb1f623891680b107f4dbdd9da7 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 14:17:58 +0800 Subject: [PATCH] =?UTF-8?q?fix(loop):=20=E6=89=80=E6=9C=89=E9=9D=9E?= =?UTF-8?q?=E5=B4=A9=E6=BA=83=E9=80=80=E5=87=BA=E4=B8=80=E5=BE=8B=E5=BA=9F?= =?UTF-8?q?=E5=BC=83=E5=8F=AF=E6=81=A2=E5=A4=8D=E5=BF=AB=E7=85=A7=EF=BC=88?= =?UTF-8?q?=E4=B8=BB=E4=BA=BA=202026-10-04=20=E8=A3=81=E5=86=B3=E3=80=8C?= =?UTF-8?q?=E4=BF=AE=E3=80=8D=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 原状：只有「被取消」的回合把 checkpoint 落定为不可恢复；**模型/工具报错、预算耗尽**等未落定退出 同样留下可恢复快照 ⇒ 下一条消息被旧窗口整段顶掉（用户看到自己刚发的消息像丢了） - 修法：收尾 defer 改为「**回合未落定就废弃快照**」（不再只看 ctx 是否取消），新增 StopReasonAborted； checkpoint.resumableStopReason 收紧为「**只有 `tool_calls` 可恢复**」（空 stop reason 不再算）； 终态快照改用 context.WithoutCancel 写（修「取消擦肩最终答案」时终态写失败、中途快照留任最新） - 语义边界（写进用例注释）：**只有崩溃**（进程猝死、函数不返回）才允许恢复；控制流一旦回到调用方 就不得留可恢复态。原 resume_test 的用例改名为**崩溃恢复**用例并保留断言（含「新消息不被顶掉」的反面断言在新用例里） - 复核：build/vet 绿 + loop 包测试与 `-race` 全绿；新增 6 条（模型报错/工具失败+模型报错/预算耗尽/ 空 stop reason/取消擦肩最终答案）+ 端到端 TestInterruptThenChatKeepsLatestContextEndToEnd 绿； **4 条变异反证**（含把废弃快照写回 tool_calls → 精确复现真实劫持现场）均已还原 - 附：修 session_turn_fence.go 里引用旧函数名的陈旧注释 --- internal/runtime/loop/checkpoint.go | 39 +-- internal/runtime/loop/resume_abandon_test.go | 236 ++++++++++++++++++ internal/runtime/loop/resume_test.go | 87 ++++++- internal/runtime/loop/run.go | 65 +++-- .../services/gateway/session_turn_fence.go | 68 +++++ 5 files changed, 448 insertions(+), 47 deletions(-) create mode 100644 internal/runtime/loop/resume_abandon_test.go create mode 100644 internal/services/gateway/session_turn_fence.go diff --git a/internal/runtime/loop/checkpoint.go b/internal/runtime/loop/checkpoint.go index 5806064..bcec9ea 100644 --- a/internal/runtime/loop/checkpoint.go +++ b/internal/runtime/loop/checkpoint.go @@ -59,12 +59,13 @@ func (l *Loop) saveCheckpoint(ctx context.Context, sessionID string, step int, m } } -// loadCheckpoint 从该会话最新的持久 checkpoint 恢复一个被中断的 -// run（design §1 durable execution）。它返回恢复后的消息 +// loadCheckpoint 从该会话最新的持久 checkpoint 恢复一个**崩溃遗留**的 +// run（design/04 §1 durable execution）。它返回恢复后的消息 // 窗口、下一个要运行的循环步（checkpoint 步 + 1），以及 // 是否找到了可恢复的 checkpoint。不存在 checkpoint，或最新的一个 -// 反映的是已完成/被阻塞的 run 时，它返回调用方的消息并置步为 1， -// 使循环从新开始。 +// 反映的是已完成/被阻塞的 run、或任何已落定的未完成退出（取消 / 报错 / +// 预算耗尽 / 护栏拦截）时，它返回调用方的消息并置步为 1，使循环从新开始 +// （主人 2026-10-04 裁决：只有崩溃才允许恢复；见 resumableStopReason）。 // // 恢复是幂等的：checkpoint 精确保存了上一个已完成步结束时 // 的消息窗口（assistant 工具调用及其结果已融合在内）， @@ -97,20 +98,20 @@ func (l *Loop) loadCheckpoint(ctx context.Context, sessionID string, messages [] } // resumableStopReason 报告最后一个步以上述 stop reason 结束的 -// checkpoint 是否代表一个值得恢复的在途 run。只有 -// 工具循环中途的步才真正算「被中断」：循环已把工具结果 -// 并入窗口并即将继续，因此从下一步接上 -// 会重建上下文而不重复工作。携带最终答案、 -// 预算耗尽、安全网终止或护栏拒绝的 checkpoint 是 -// 已完成/被阻塞的 run —— 恢复它会重新生成已产出的答案 -// （重复副作用）—— 因此这些一律落到全新开始（打开 -// 真正新会话的调用方使用不同的 sessionID，本就错过 -// LoadLatest）。 +// checkpoint 是否代表一个**崩溃遗留**的在途 run。只有落在工具循环中途的步 +// 算「崩溃遗留」：循环已把工具结果并入窗口并即将继续，进程却猝死了—— +// 此时收尾 defer 根本没执行，快照是进程最后留下的证据，从下一步接上会由 +// 历史重建上下文而不重复工作（design/04 §1 durable execution）。 +// +// 其余一律落到全新开始（主人 2026-10-04 裁决：只有崩溃才允许恢复）： +// - 携带最终答案、护栏拒绝、预算耗尽或安全网终止的 checkpoint 是 +// 已完成/被阻塞的 run——恢复它会重新生成已产出的答案（重复副作用）； +// - canceled / aborted 是**已落定**的未完成退出，下一回合必须自己组 +// 上下文（裁决 13.4.5），不得被旧窗口顶掉； +// - 空的 stop reason 无法证明是崩溃遗留（崩溃留下的快照必带 +// tool_calls），保守起见同样不恢复。 +// +// 打开真正新会话的调用方使用不同的 sessionID，本就错过 LoadLatest。 func resumableStopReason(stopReason string) bool { - switch stopReason { - case "", agent.StopReasonToolCalls: - return true - default: - return false - } + return stopReason == agent.StopReasonToolCalls } diff --git a/internal/runtime/loop/resume_abandon_test.go b/internal/runtime/loop/resume_abandon_test.go new file mode 100644 index 0000000..0f0009d --- /dev/null +++ b/internal/runtime/loop/resume_abandon_test.go @@ -0,0 +1,236 @@ +package loop + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" +) + +// 本文件钉住主人 2026-10-04 的裁决：**只有崩溃才允许恢复**。任何已落定的 +// 未完成退出——模型报错、工具报错、预算耗尽、安全网终止、护栏拦截—— +// 都必须让该会话的可恢复快照失效，使下一条消息自己组最新上下文，而不是被 +// 放弃回合的旧窗口顶掉（裁决 13.4.5：上下文一致性 / 完整性 / 最新性； +// 否则用户会看到自己刚发的消息像丢了）。取消的同一断言见 +// run_cancel_test.go，崩溃仍可恢复见 resume_test.go。 + +// lastRecorded 返回适配器已记录请求里的最后一条。 +func lastRecorded(t *testing.T, reqs []model.Request) model.Request { + t.Helper() + if len(reqs) == 0 { + t.Fatalf("适配器没有收到任何模型请求") + } + return reqs[len(reqs)-1] +} + +// requireNewMessageContext 断言被放弃回合之后的下一条消息自己组了最新上下文： +// 请求里有「B」，且不再带旧窗口的任何痕迹（工具名 echo 只出现在旧窗口的 +// assistant 工具调用轮与工具结果消息里）。 +func requireNewMessageContext(t *testing.T, req model.Request) { + t.Helper() + if !requestHasText(req, "B") { + t.Fatalf("未落定退出之后的请求不含新消息 B：%v（上下文不最新/不完整）", requestTexts(req)) + } + for _, m := range req.Messages { + if strings.Contains(messageText(m), "echo") { + t.Fatalf("未落定退出之后的请求仍带着旧窗口（工具调用/工具结果）：%v", requestTexts(req)) + } + } +} + +// requireLatestStopReason 断言该会话最新的快照以 want 落定，且它不可恢复。 +// 它走 LoadLatest——与 loadCheckpoint 恢复时同一道门——因此断言的是 +// 「下一条消息不会被旧窗口顶掉」这个真实判定，而不是某个脆弱的字段。 +func requireLatestStopReason(t *testing.T, ctx context.Context, cp *persist.Store, sessionID, want string) { + t.Helper() + latest, err := cp.LoadLatest(ctx, sessionID) + if err != nil { + t.Fatalf("LoadLatest(%s): %v", sessionID, err) + } + var st checkpointState + if err := json.Unmarshal(latest.State, &st); err != nil { + t.Fatalf("unmarshal checkpoint %s: %v", latest.ID, err) + } + if st.StopReason != want { + t.Fatalf("最新快照 %s 的 stop_reason = %q, want %q", latest.ID, st.StopReason, want) + } + if resumableStopReason(st.StopReason) { + t.Fatalf("最新快照 %s（stop_reason=%q）仍可恢复：下一条消息会被旧窗口顶掉", latest.ID, st.StopReason) + } +} + +// TestRun_ModelErrorDoesNotHijackNextTurn：回合 A 的步 1 请求工具（落下中途 +// tool_calls 快照），步 2 的模型调用报错——这是**已落定的未完成退出**，不是 +// 崩溃（崩溃时收尾 defer 根本不执行）。因此 A 留下的快照必须被落定为 aborted， +// 同一会话的下一条消息 B 以 B 自己的最新上下文开始。 +func TestRun_ModelErrorDoesNotHijackNextTurn(t *testing.T) { + t.Parallel() + ctx := context.Background() + call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {err: errors.New("model unavailable")}, + {resp: model.Response{Content: "final"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + cp := newTestCheckpointStore(t) + l := New(Options{Agent: newTestAgent(t, ad, td), Checkpoint: cp, MaxRounds: 8}) + + const sess = "sess-model-error" + if _, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "A")}); err == nil { + t.Fatalf("回合 A 应在步 2 模型报错时返回错误") + } + requireLatestStopReason(t, ctx, cp, sess, StopReasonAborted) + + res, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "B")}) + if err != nil { + t.Fatalf("Run(B): %v", err) + } + if res.Text != "final" || res.Steps != 1 { + t.Fatalf("回合 B = %q / steps=%d, want final / 1（aborted 快照不可恢复，B 必须全新开始）", res.Text, res.Steps) + } + requireNewMessageContext(t, lastRecorded(t, ad.reqs)) +} + +// TestRun_ToolFailureThenModelErrorDoesNotHijackNextTurn：工具执行失败会把 +// 结构化错误回灌给模型（循环并不因此退出），随后模型调用报错才让回合未落定 +// 退出。此时步 1 的中途快照同样必须被废弃，而不是等在那里劫持下一条消息。 +func TestRun_ToolFailureThenModelErrorDoesNotHijackNextTurn(t *testing.T) { + t.Parallel() + ctx := context.Background() + call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {err: errors.New("model unavailable")}, + {resp: model.Response{Content: "final"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", err: errors.New("tool exploded")}) + cp := newTestCheckpointStore(t) + l := New(Options{Agent: newTestAgent(t, ad, td), Checkpoint: cp, MaxRounds: 8}) + + const sess = "sess-tool-error" + if _, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "A")}); err == nil { + t.Fatalf("回合 A 应在模型报错时返回错误") + } + requireLatestStopReason(t, ctx, cp, sess, StopReasonAborted) + + res, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "B")}) + if err != nil { + t.Fatalf("Run(B): %v", err) + } + if res.Text != "final" || res.Steps != 1 { + t.Fatalf("回合 B = %q / steps=%d, want final / 1（工具失败+报错的快照不可恢复）", res.Text, res.Steps) + } + requireNewMessageContext(t, lastRecorded(t, ad.reqs)) +} + +// TestRun_MaxRoundsDoesNotHijackNextTurn：预算耗尽是一条**已落定**的退出 +// （settleAndReturn 在同一个步上覆写为终态快照 max_rounds），它同样不得让 +// 下一条消息被旧窗口顶掉。它也顺带钉住「已落定 → 收尾 defer 不改写」： +// 最新快照仍是该方法自己写的 max_rounds，而不是 canceled / aborted。 +func TestRun_MaxRoundsDoesNotHijackNextTurn(t *testing.T) { + t.Parallel() + ctx := context.Background() + call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}` + ad := &mockAdapter{chat: []mgResp{ + {resp: model.Response{Content: call}}, + {resp: model.Response{Content: "final"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + cp := newTestCheckpointStore(t) + l := New(Options{Agent: newTestAgent(t, ad, td), Checkpoint: cp, MaxRounds: 1}) + + const sess = "sess-max-rounds" + resA, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "A")}) + if err != nil { + t.Fatalf("Run(A): %v", err) + } + if resA.StopReason != StopReasonMaxRounds { + t.Fatalf("回合 A 的 stop reason = %q, want %q", resA.StopReason, StopReasonMaxRounds) + } + requireLatestStopReason(t, ctx, cp, sess, StopReasonMaxRounds) + + res, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "B")}) + if err != nil { + t.Fatalf("Run(B): %v", err) + } + if res.Text != "final" || res.Steps != 1 { + t.Fatalf("回合 B = %q / steps=%d, want final / 1（预算耗尽的快照不可恢复）", res.Text, res.Steps) + } + requireNewMessageContext(t, lastRecorded(t, ad.reqs)) +} + +// cancelingFinalAdapter 在交出最终答案的同一刻取消该回合的 ctx，复现「收尾 +// 落定与用户中断擦肩而过」的竞态：回合本身已经得出终态，因此它的终态快照 +// 必须真正写下去（见 Loop.settleAndReturn）——否则最新快照会停在中途的 +// tool_calls 上，下一条消息又被旧窗口顶掉。 +type cancelingFinalAdapter struct { + cancel func() + + mu sync.Mutex + n int + reqs []model.Request +} + +func (a *cancelingFinalAdapter) Name() string { return "canceling-final" } + +func (a *cancelingFinalAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + a.mu.Lock() + a.n++ + n := a.n + a.reqs = append(a.reqs, req) + a.mu.Unlock() + switch n { + case 1: + return model.Response{Content: `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`}, nil + case 2: + a.cancel() + return model.Response{Content: "final"}, nil + default: + return model.Response{Content: "final"}, nil + } +} + +func (a *cancelingFinalAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return a.Chat(ctx, req, model.Options{}) +} + +func (a *cancelingFinalAdapter) recorded() []model.Request { + a.mu.Lock() + defer a.mu.Unlock() + return append([]model.Request(nil), a.reqs...) +} + +// TestRun_CancelRacingFinalAnswerStillSettlesCheckpoint：回合在交出最终答案的 +// 瞬间被取消——它**已经落定**（有终态可交付），因此不得被当成中断丢弃： +// 终态快照必须落盘，下一条消息读到的是它而不是中途的 tool_calls。 +func TestRun_CancelRacingFinalAnswerStillSettlesCheckpoint(t *testing.T) { + t.Parallel() + ctx := context.Background() + runCtx, cancel := context.WithCancel(ctx) + ad := &cancelingFinalAdapter{cancel: cancel} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + cp := newTestCheckpointStore(t) + l := New(Options{Agent: newTestAgent(t, ad, td), Checkpoint: cp, MaxRounds: 8}) + + const sess = "sess-cancel-race" + res, err := l.Run(runCtx, sess, []model.Message{model.TextMessage(model.RoleUser, "A")}) + if err != nil { + t.Fatalf("Run(A): %v（终态已得出，取消不得把这一轮变成错误）", err) + } + if res.Text != "final" || res.StopReason != agent.StopReasonStop { + t.Fatalf("回合 A = %q / %q, want final / %q", res.Text, res.StopReason, agent.StopReasonStop) + } + requireLatestStopReason(t, ctx, cp, sess, agent.StopReasonStop) + + if _, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "B")}); err != nil { + t.Fatalf("Run(B): %v", err) + } + requireNewMessageContext(t, lastRecorded(t, ad.recorded())) +} diff --git a/internal/runtime/loop/resume_test.go b/internal/runtime/loop/resume_test.go index afe050c..0aefbfa 100644 --- a/internal/runtime/loop/resume_test.go +++ b/internal/runtime/loop/resume_test.go @@ -130,10 +130,18 @@ func TestRun_NewSession_NotResumed(t *testing.T) { } } -// TestRun_Resume_LoadLatestCalled 断言 Run 向 store 索要该会话 -// 最新的 checkpoint 并据此恢复（LoadLatest → tool_calls checkpoint → -// 从下一步继续）。 -func TestRun_Resume_LoadLatestCalled(t *testing.T) { +// TestRun_Resume_FromCrashSnapshot_LoadLatestCalled 断言 Run 向 store 索要该 +// 会话最新的 checkpoint，并据**崩溃遗留**的 tool_calls 快照从下一步继续。 +// +// 本用例建模的是崩溃恢复：夹具里的 tool_calls 快照只可能由崩溃留下——任何 +// 已落定的未完成退出（取消 / 模型报错 / 工具报错 / 预算耗尽 / 安全网终止 / +// 护栏拦截）都被 driveTurn 落定为 canceled / aborted 或各自的终态 stop +// reason，因而不可恢复（主人 2026-10-04 裁决：**只有崩溃才允许恢复**，且 +// 崩溃口径是「该回合从未写出终态快照」）。因此这里调用方的新消息被恢复出的 +// 旧窗口顶掉，是裁决**明确保留**的崩溃语义，不是我们擅自放宽；反面 +// （已落定的未完成退出不得顶掉下一条消息）由 TestRun_Resume_AbortedSnapshotNotResumed +// 与 resume_abandon_test.go、run_cancel_test.go 的用例钉住。 +func TestRun_Resume_FromCrashSnapshot_LoadLatestCalled(t *testing.T) { t.Parallel() ctx := context.Background() ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final"}}}} @@ -160,12 +168,83 @@ func TestRun_Resume_LoadLatestCalled(t *testing.T) { if res.Steps != 3 { t.Fatalf("Steps = %d, want 3 (resumed at step 3)", res.Steps) } + // 崩溃恢复的代价（裁决保留）：喂给模型的正是崩溃前的窗口，调用方新传入的 + // 「new」被它顶掉——因此崩溃恢复只适用于「进程猝死、该回合从未写出终态 + // 快照」的情形，已落定的未完成退出见下一条用例。 got := ad.reqs[0].Messages if len(got) != 2 || messageText(got[1]) != "keep going" { t.Fatalf("resumed request = %+v, want [system, 'keep going']", got) } } +// TestRun_Resume_AbortedSnapshotNotResumed 是上一条崩溃用例的**反面**： +// 同一夹具、同样是最新快照，但它的 stop reason 是 aborted（模型报错 / 工具 +// 报错等已落定的未完成退出）。这种快照绝不可恢复——调用方的新消息必须自己 +// 组上下文（主人 2026-10-04 裁决：只有崩溃才允许恢复；已落定的未完成退出 +// 不得让下一条消息被旧窗口劫持）。 +func TestRun_Resume_AbortedSnapshotNotResumed(t *testing.T) { + t.Parallel() + ctx := context.Background() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final"}}}} + a := newTestAgent(t, ad, nil) + + buf, err := json.Marshal(checkpointState{ + Step: 2, + Messages: []model.Message{model.TextMessage(model.RoleUser, "abandoned window")}, + StopReason: StopReasonAborted, + }) + if err != nil { + t.Fatalf("marshal checkpoint: %v", err) + } + fake := &fakeCheckpointStore{latest: &persist.Checkpoint{ID: "sess:2", AgentID: "sess", Step: 2, State: buf}} + l := New(Options{Agent: a, Checkpoint: fake, MaxRounds: 8}) + + res, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, "new")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Steps != 1 { + t.Fatalf("Steps = %d, want 1（aborted 快照不可恢复，必须从步 1 全新开始）", res.Steps) + } + got := ad.reqs[0].Messages + if len(got) != 2 || messageText(got[1]) != "new" { + t.Fatalf("request = %+v, want [system, 'new']（新消息不得被旧窗口顶掉）", got) + } +} + +// TestRun_Resume_EmptyStopReasonNotResumed：stop reason 为空的快照无法证明是 +// 崩溃遗留（崩溃留下的快照必带 tool_calls），保守起见同样不恢复——否则任何 +// 漏写 stop reason 的旧行都会劫持下一条消息（主人 2026-10-04 裁决：只有 +// 崩溃才允许恢复）。 +func TestRun_Resume_EmptyStopReasonNotResumed(t *testing.T) { + t.Parallel() + ctx := context.Background() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final"}}}} + a := newTestAgent(t, ad, nil) + + buf, err := json.Marshal(checkpointState{ + Step: 2, + Messages: []model.Message{model.TextMessage(model.RoleUser, "unmarked window")}, + }) + if err != nil { + t.Fatalf("marshal checkpoint: %v", err) + } + fake := &fakeCheckpointStore{latest: &persist.Checkpoint{ID: "sess:2", AgentID: "sess", Step: 2, State: buf}} + l := New(Options{Agent: a, Checkpoint: fake, MaxRounds: 8}) + + res, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, "new")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Steps != 1 { + t.Fatalf("Steps = %d, want 1（无 stop reason 的快照不可恢复）", res.Steps) + } + got := ad.reqs[0].Messages + if len(got) != 2 || messageText(got[1]) != "new" { + t.Fatalf("request = %+v, want [system, 'new']（新消息不得被旧窗口顶掉）", got) + } +} + // TestRun_Resume_NotFound_Fresh 断言全新路径上仍会调用 LoadLatest， // 且 ErrNotFound 会以调用方的消息从步 1 完全重启 // （没有可复活的 checkpoint）。 diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index 9649b0d..b0df1f6 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -59,11 +59,13 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa l.emitTurnBoundary(ctx, model.StreamKindTurnStart) defer l.emitTurnBoundary(ctx, model.StreamKindTurnEnd) - // 恢复被中断的 run（design §1 durable execution）：该会话有 - // 可恢复的 checkpoint 时，重建其上下文并从下一步继续， - // 而不是从全新对话重启。不同的 - // sessionID —— 或已完成 run 留下的 checkpoint —— 落到 - // 全新开始（LoadLatest 未命中，或 stop reason 不可恢复）。 + // 恢复**崩溃遗留**的 run（design/04 §1 durable execution）：该会话最新 + // 快照若是崩溃（进程猝死、回合从未写出终态、收尾 defer 根本没执行）留下的 + // 中途 tool_calls，就重建其上下文并从下一步继续，而不是从全新对话重启。 + // 其余一切——不同的 sessionID、已完成 run 的终态、以及任何**已落定的 + // 未完成退出**（取消 / 模型报错 / 工具报错 / 预算耗尽 / 护栏拦截）留下的 + // 快照——都落到全新开始，让本次消息自己组最新上下文（主人 2026-10-04 + // 裁决：只有崩溃才允许恢复；见 resumableStopReason 与 driveTurn 的收尾）。 msgs, startStep, _ := l.loadCheckpoint(ctx, sessionID, messages) msgs = l.resolveImages(ctx, msgs) res := &Result{} @@ -83,18 +85,22 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Message, startStep int, res *Result) (*Result, error) { toolState := &toolLoopState{prevFP: make(map[string]bool)} step := startStep - // 被用户中断（design/15 §3）的回合不留下「可恢复的在途态」：design/15 - // 要求中断后**立即**按最新上下文处理下一条消息，而一个被放弃的回合若把 - // 可恢复 checkpoint 留在该会话上，下一条消息会被它顶掉——继承旧窗口、 - // 丢掉刚发出的那条，上下文既不最新也不完整（裁决 13.4.5）。因此取消后 - // 把该会话的快照落定为不可恢复态（快照本身保留，仍可查询）。 - // - // 只在回合**未落定**时标记（落定路径已写入终态快照，其 stop reason 本就 - // 不可恢复），因此这里不会改写一个已交付答案的回合。 + // 未落定退出的回合不留下「可恢复的在途态」（裁决 13.4.5：下一条消息必须 + // 看到自己的最新上下文，而不是被放弃回合的旧窗口顶掉）。判别口径（主人 + // 2026-10-04 裁决，「只有崩溃才允许恢复」）：**崩溃**时本 defer 根本不执行， + // 中途快照留给下回合续接（design/04 §1 durable execution）；**任何已落定的 + // 未完成退出**（取消 / 模型报错 / 工具报错；预算耗尽、安全网终止与护栏拦截 + // 经 settleAndReturn 写终态）都落定为不可恢复。故只按 res.StopReason 判别： + // 为空 = 本回合没落定，废弃其快照；非空 = 终态已写出，绝不改写。 defer func() { - if ctx.Err() != nil && res.StopReason == "" { - l.abandonCanceledCheckpoint(ctx, sessionID, step, msgs) + if res.StopReason != "" { + return } + reason := StopReasonAborted + if ctx.Err() != nil { + reason = StopReasonCanceled + } + l.abandonCheckpoint(ctx, sessionID, step, msgs, reason) }() for ; step <= l.maxRounds; step++ { if err := ctx.Err(); err != nil { @@ -151,8 +157,12 @@ func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Mes // settleAndReturn 落定一个已得出 Result 的回合：以该路径的 // stop reason 持久化终态 checkpoint、触发后台记忆提取、 // 上报 turn 钩子（情绪标签写回 Result.Mood），然后把结果交还调用方。 +// +// 终态快照脱离 ctx 的取消信号写入：落定是一个**已经发生**的决定，若恰在 +// 此时收到取消而写失败，最新快照会停在中途的 tool_calls 上，下一条消息便 +// 会被旧窗口顶掉——与 driveTurn 收尾 defer 要挡的是同一类缺陷。 func (l *Loop) settleAndReturn(ctx context.Context, sessionID string, msgs []model.Message, step int, stopReason string, res *Result) (*Result, error) { - l.saveCheckpoint(ctx, sessionID, step, msgs, stopReason) + l.saveCheckpoint(context.WithoutCancel(ctx), sessionID, step, msgs, stopReason) l.extractMemoryAsync(ctx, msgs, step) l.settleTurn(ctx, sessionID, msgs, res) return res, nil @@ -161,18 +171,25 @@ func (l *Loop) settleAndReturn(ctx context.Context, sessionID string, msgs []mod // StopReasonCanceled 标记一个因用户中断（design/15 §3「停下/算了/换个话题」） // 而放弃的回合。它不是可恢复的在途态：design/15 要求中断后立即按**最新** // 上下文处理下一条消息，因此这种快照绝不能被下一条消息续接 -// （见 driveTurn 与 abandonCanceledCheckpoint）。 +// （见 driveTurn 与 abandonCheckpoint）。 const StopReasonCanceled = "canceled" -// abandonCanceledCheckpoint 把一个被取消回合的 checkpoint 落定为不可恢复态 -// （stop reason = StopReasonCanceled），使该会话的下一条消息从它自己的最新 -// 上下文开始，而不是被放弃回合的旧窗口顶掉（裁决 13.4.5：上下文一致性 / -// 完整性 / 最新性）。快照本身保留在 store 里，仍可查询。 +// StopReasonAborted 标记一个**非取消**的未落定退出：回合在得出终态之前因 +// 模型报错、工具报错等原因把控制流交还了调用方。它与 StopReasonCanceled +// 一样使该会话的 checkpoint 不可恢复——主人 2026-10-04 裁决：只有崩溃 +// （进程猝死、该回合从未写出终态快照，因而收尾 defer 根本没执行）才允许 +// 下一回合续接旧窗口（见 driveTurn 与 abandonCheckpoint）。 +const StopReasonAborted = "aborted" + +// abandonCheckpoint 把一个未落定回合的 checkpoint 落定为不可恢复态 +// （stop reason = reason），使该会话的下一条消息从它自己的最新上下文开始， +// 而不是被放弃回合的旧窗口顶掉（裁决 13.4.5：上下文一致性 / 完整性 / +// 最新性）。快照本身保留在 store 里，仍可查询。 // // 写入必须脱离已取消的 ctx（否则 ExecContext 必然立即失败），因此这里用 // WithoutCancel 派生背景上下文，保留原 ctx 的值而丢弃取消信号。失败只记日志， // 绝不让这个收尾动作把中断路径变成错误。 -func (l *Loop) abandonCanceledCheckpoint(ctx context.Context, sessionID string, step int, msgs []model.Message) { - l.log.Info("checkpoint abandoned", "session", sessionID, "step", step, "stop_reason", StopReasonCanceled) - l.saveCheckpoint(context.WithoutCancel(ctx), sessionID, step, msgs, StopReasonCanceled) +func (l *Loop) abandonCheckpoint(ctx context.Context, sessionID string, step int, msgs []model.Message, reason string) { + l.log.Info("checkpoint abandoned", "session", sessionID, "step", step, "stop_reason", reason) + l.saveCheckpoint(context.WithoutCancel(ctx), sessionID, step, msgs, reason) } diff --git a/internal/services/gateway/session_turn_fence.go b/internal/services/gateway/session_turn_fence.go new file mode 100644 index 0000000..73796d4 --- /dev/null +++ b/internal/services/gateway/session_turn_fence.go @@ -0,0 +1,68 @@ +package gateway + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// 本文件把「同一会话最多一个在途回合」从一句约定变成可等待的进程内围栏 +// （design/15 §3 与裁决 13.4.5 的组合要求）。 +// +// 中断必须**立即**把会话交回空闲并接受新消息（design/15 §3），但被取消的 +// 回合仍在收尾：loop 直到它自己的 Run 返回之前，才把被放弃回合的快照落定 +// 为不可恢复态（`loop/run.go` 的 `abandonCheckpoint`；2026-10-04 起**所有非崩溃退出** +// 都走它，不再只限取消）。若新回合此刻 +// 就开跑，它可能在该落定之前读到那份**仍可恢复**的旧快照并续接它 —— 调用方 +// 刚发出的那条消息会被整段顶掉，上下文既不最新也不完整（裁决 13.4.5 的 +// 最新性 / 完整性）。 +// +// 围栏关掉这条竞态窗口，且不阻塞调用方：新消息照样被立即接受、Submit 立即 +// 返回（非阻塞是硬约束），只是它的回合要等上一回合真正返回之后才开始。 + +// turnFence 是一次回合的启动围栏。wait 是上一回合的收尾信号（nil 表示没有 +// 在途的上一回合，无需等待）；done 是本回合的收尾信号，在本回合返回时关闭， +// 从而放行等它的下一回合。 +type turnFence struct { + wait <-chan struct{} + done chan struct{} +} + +// newTurnFence 以会话当前的收尾信号为先驱建一道新围栏；prev 为 nil 时 +// wait 也为 nil（本回合可立即开始）。 +func newTurnFence(prev chan struct{}) turnFence { + var wait <-chan struct{} + if prev != nil { + wait = prev + } + return turnFence{wait: wait, done: make(chan struct{})} +} + +// release 关闭本回合的收尾信号，放行正在等它的下一回合。每个回合恰好调用 +// 一次（它在本回合的唯一出口上被 defer）。 +func (f turnFence) release() { close(f.done) } + +// awaitTurnFence 等到上一回合真正返回，使被取消回合的 checkpoint 收尾先 +// 落定。等待同时盯着本回合自己的 ctx，因此一个迟迟不收敛的上一回合不会把 +// 本回合永远挂住：用户再按一次「停下」即可让它立即结算。 +// +// 它只负责**等待**，绝不因此跳过本回合的驱动：本回合若在等到之前就已被 +// 取消（连续两次中断 / 网关关闭），驱动本身会以 ctx 错误立即返回。取消因此 +// 始终被 runner 观察到 —— 被取消的回合绝不会在无人知晓的情况下消失。 +func awaitTurnFence(ctx context.Context, f turnFence) { + if f.wait == nil { + return + } + select { + case <-f.wait: + case <-ctx.Done(): + } +} + +// runTurnFenced 在上一回合收尾之后驱动本回合（编排器其余部分仍只认识 +// TurnRunner，围栏只在启动处生效）。 +func (o *Orchestrator) runTurnFenced(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv, fence turnFence) (*loop.Result, error) { + awaitTurnFence(ctx, fence) + return o.run(ctx, sessionID, messages, env) +}