Add JournalOnly sessions and persist one-time history seeds (N3 WP2)
Why: the product conversation path needs the intent journal and recovery pairing, but making the transcript the authoritative context would freeze the assembled system prefix (memory/plans/injections go stale). JournalOnly keeps the session as journal-only: recovery runs against it, the context always comes from History, and the recovered outcomes are injected as a system note. Separately, the one-time History seed was memory-only and silently lost from the second run on (authority mode); it is now persisted except in JournalOnly mode. Evidence: GOWORK=off go vet + go test -count=1 ./... green; TestJournalOnlyRecoversWithoutTranscriptAuthority (journal gets the replayed pair exactly once, history never enters the journal, the recovery note precedes the new user message, seeds stay out); persistedProjection honors JournalOnly; all pre-existing durable and harness tests pass unchanged (default semantics untouched). Impact: new opt-in Config.JournalOnly; default (false) behavior is byte-identical. Refs: Agent-System-Documents#43 (N3 plan WP2)
This commit is contained in:
+80
-2
@@ -121,7 +121,7 @@ func TestStoredSafeCurrentNeverDoesNotRerun(t *testing.T) {
|
||||
tool := &replayTool{countingTool: countingTool{name: "read_file"}, replay: tools.ReplayNever}
|
||||
reg := tools.NewRegistry(dir)
|
||||
reg.Register(tool)
|
||||
if err := recoverPending(context.Background(), Config{Session: st, Tools: reg}); err != nil {
|
||||
if _, err := recoverPending(context.Background(), Config{Session: st, Tools: reg}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tool.calls != 0 {
|
||||
@@ -354,7 +354,7 @@ func TestRepairOrphanTailNoopOnHealthyTranscript(t *testing.T) {
|
||||
}
|
||||
}
|
||||
before, _ := st.Load()
|
||||
if err := repairOrphanTail(Config{Session: st}); err != nil {
|
||||
if _, err := repairOrphanTail(Config{Session: st}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
after, _ := st.Load()
|
||||
@@ -362,3 +362,81 @@ func TestRepairOrphanTailNoopOnHealthyTranscript(t *testing.T) {
|
||||
t.Fatalf("健康转录被改写：%d → %d", len(before), len(after))
|
||||
}
|
||||
}
|
||||
|
||||
// ---- N3 WP2：JournalOnly（journal 只作日志，上下文由 History 组装）----
|
||||
|
||||
// JournalOnly：journal 有杀窗口尾巴 → 恢复配对写入 journal；上下文来自
|
||||
// History（不含 journal 内容），恢复摘要以系统消息注入；种子不进 journal。
|
||||
func TestJournalOnlyRecoversWithoutTranscriptAuthority(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
st, err := session.Open(filepath.Join(dir, "s.jsonl"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.Append(llm.Message{
|
||||
Role: llm.RoleAssistant, StopReason: llm.StopToolCalls,
|
||||
ToolCalls: []llm.ToolCall{{ID: "c1", Name: "read_file", Arguments: `{}`}},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.AppendToolIntent(session.ToolIntent{CallID: "c1", Name: "read_file", Arguments: `{}`, Replay: tools.ReplaySafe}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reg := tools.NewRegistry(dir)
|
||||
reg.Register(&replayTool{countingTool: countingTool{name: "read_file"}, replay: tools.ReplaySafe})
|
||||
var gotReq []llm.Message
|
||||
res, err := Run(context.Background(), Config{
|
||||
Model: "fake", Session: st, JournalOnly: true, Tools: reg, MaxTurns: 2,
|
||||
History: []llm.Message{
|
||||
{Role: llm.RoleSystem, Content: "sys"},
|
||||
{Role: llm.RoleUser, Content: "历史一"},
|
||||
{Role: llm.RoleAssistant, Content: "回复一"},
|
||||
},
|
||||
Stream: func(ctx context.Context, model string, msgs []llm.Message, _ []llm.ToolDecl, _ llm.StreamOptions) <-chan llm.Event {
|
||||
gotReq = append([]llm.Message{}, msgs...)
|
||||
ch := make(chan llm.Event, 1)
|
||||
ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{Role: llm.RoleAssistant, Content: "继续", StopReason: llm.StopEnd}}
|
||||
close(ch)
|
||||
return ch
|
||||
},
|
||||
}, llm.Message{Role: llm.RoleUser, Content: "最新"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// journal 里 c1 被 safe 重放配对（一次真实结果）。
|
||||
journal, _ := st.Load()
|
||||
var journaled bool
|
||||
for _, m := range journal {
|
||||
if m.Role == llm.RoleTool && m.ToolCallID == "c1" && m.Content == "ok" {
|
||||
journaled = true
|
||||
}
|
||||
if m.Content == "历史一" {
|
||||
t.Fatal("种子历史不得进 journal")
|
||||
}
|
||||
}
|
||||
if !journaled {
|
||||
t.Fatalf("journal 缺恢复配对：%#v", journal)
|
||||
}
|
||||
// 上下文：History 内容在；恢复摘要在；且在新 user 消息之前。
|
||||
var sawHist, sawNote int
|
||||
noteIdx, userIdx := -1, -1
|
||||
for i, m := range gotReq {
|
||||
if m.Content == "历史一" {
|
||||
sawHist++
|
||||
}
|
||||
if m.Role == llm.RoleSystem && strings.Contains(m.Content, "已完成恢复") && strings.Contains(m.Content, "c1") {
|
||||
sawNote++
|
||||
noteIdx = i
|
||||
}
|
||||
if m.Content == "最新" {
|
||||
userIdx = i
|
||||
}
|
||||
}
|
||||
if sawHist != 1 || sawNote != 1 {
|
||||
t.Fatalf("历史=%d 摘要=%d：%#v", sawHist, sawNote, gotReq)
|
||||
}
|
||||
if noteIdx < 0 || userIdx < 0 || noteIdx > userIdx {
|
||||
t.Fatalf("摘要必须在新 user 消息之前：note=%d user=%d", noteIdx, userIdx)
|
||||
}
|
||||
_ = res
|
||||
}
|
||||
|
||||
+73
-16
@@ -111,6 +111,11 @@ type Config struct {
|
||||
// ResultsDir 非空时，被清理的旧工具结果完整内容落盘到该目录
|
||||
// （占位写路径）；空则占位指向会话记录（旧行为）。
|
||||
ResultsDir string
|
||||
// JournalOnly 为真且 Session 非空时：Session 只作意图日志与恢复配对
|
||||
// （不恢复的完整权威上下文），请求上下文始终由 History 组装；
|
||||
// 本次恢复写入的配对摘要以系统消息注入（不持久化，已在 journal 里）。
|
||||
// 为假（默认）保持既有语义：Session 为权威上下文。
|
||||
JournalOnly bool
|
||||
// Hooks 是进程内钩子。空总线不改变停止、提醒和拒绝次数。
|
||||
Hooks *hooks.Bus
|
||||
// PermissionAsk 在钩子对一次工具调用表态 ask 时向宿主请求裁决。
|
||||
@@ -174,37 +179,61 @@ func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) {
|
||||
endRun := startSpan(cfg, "invoke_agent", map[string]string{"model": cfg.Model})
|
||||
defer endRun()
|
||||
var messages []llm.Message
|
||||
if cfg.Session != nil {
|
||||
if cfg.Session != nil && !cfg.JournalOnly {
|
||||
loaded, err := cfg.Session.Load()
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
messages = loaded
|
||||
}
|
||||
if err := recoverPending(ctx, cfg); err != nil {
|
||||
recovered, err := recoverPending(ctx, cfg)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
// 尾部配对兜底：无论进程死在 persist(assistant) 与 commitToolIntent
|
||||
// 之间的哪个窗口，补全尾部 assistant 的无结果调用，转录配对恒成立。
|
||||
if err := repairOrphanTail(cfg); err != nil {
|
||||
repaired, err := repairOrphanTail(cfg)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
if cfg.Session != nil {
|
||||
recovered = append(recovered, repaired...)
|
||||
if cfg.Session != nil && !cfg.JournalOnly {
|
||||
loaded, err := cfg.Session.Load()
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
messages = loaded
|
||||
}
|
||||
seeded := false
|
||||
if len(messages) == 0 && len(cfg.History) > 0 {
|
||||
messages = append([]llm.Message{}, cfg.History...)
|
||||
seeded = true
|
||||
}
|
||||
prependedSys := false
|
||||
if len(messages) == 0 || messages[0].Role != llm.RoleSystem {
|
||||
sys := llm.Message{Role: llm.RoleSystem, Content: systemText(cfg)}
|
||||
messages = append([]llm.Message{sys}, messages...)
|
||||
if err := persist(cfg, sys); err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
prependedSys = true
|
||||
}
|
||||
// 种子落盘（N3 WP2）：转录为权威上下文时，History 只作一次性种子——
|
||||
// 不落盘则第二跑起种子静默丢失。已落盘的前置系统消息不重复写；
|
||||
// JournalOnly 时种子不进 journal（journal 只收本轮起的消息）。
|
||||
if seeded && !cfg.JournalOnly {
|
||||
for i, m := range messages {
|
||||
if i == 0 && prependedSys {
|
||||
continue
|
||||
}
|
||||
if err := persist(cfg, m); err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
}
|
||||
}
|
||||
if cfg.JournalOnly && len(recovered) > 0 {
|
||||
// 恢复摘要以系统消息注入上下文（不持久化——内容已在 journal 里）。
|
||||
messages = append(messages, llm.Message{Role: llm.RoleSystem, Content: recoveredNoteText(recovered)})
|
||||
}
|
||||
if err := persist(cfg, user); err != nil {
|
||||
return Result{}, err
|
||||
@@ -699,7 +728,8 @@ func prepareRequest(cfg Config, messages []llm.Message, turn int, repeats *repea
|
||||
// persistedProjection 在有会话存储时只采用已经落盘的消息。
|
||||
// steer 与 follow-up 必须先写入存储，下一轮请求才会看见它们。
|
||||
func persistedProjection(cfg Config, messages []llm.Message) []llm.Message {
|
||||
if cfg.Session == nil {
|
||||
// JournalOnly：journal 不作权威上下文，每轮请求都用内存态（History 组装）。
|
||||
if cfg.Session == nil || cfg.JournalOnly {
|
||||
return messages
|
||||
}
|
||||
loaded, err := cfg.Session.Load()
|
||||
@@ -715,14 +745,17 @@ const unknownSideEffect = "副作用未知，未重试"
|
||||
// commitToolIntent 之间的窗口时使用）。
|
||||
const crashInterrupted = "进程中断，该调用未执行"
|
||||
|
||||
func recoverPending(ctx context.Context, cfg Config) error {
|
||||
// recoverPending 把未配对的意图补齐（safe 双声明重放，否则「副作用未知」
|
||||
// 配对），返回本次写入的配对消息（JournalOnly 时供恢复摘要使用）。
|
||||
func recoverPending(ctx context.Context, cfg Config) ([]llm.Message, error) {
|
||||
if cfg.Session == nil {
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
pending, err := cfg.Session.PendingIntents()
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
var out []llm.Message
|
||||
for _, in := range pending {
|
||||
current := tools.ReplayNever
|
||||
if cfg.Tools != nil {
|
||||
@@ -738,10 +771,11 @@ func recoverPending(ctx context.Context, cfg Config) error {
|
||||
msg.IsError = res.IsError
|
||||
}
|
||||
if err := persist(cfg, msg); err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, msg)
|
||||
}
|
||||
return nil
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// repairOrphanTail 兜底配对：找到转录里最后一个带 tool_calls 的 assistant，
|
||||
@@ -749,13 +783,14 @@ func recoverPending(ctx context.Context, cfg Config) error {
|
||||
// 结果并落盘。与意图日志正交（按 CallID 判重，已配对的跳过；意图路径
|
||||
// 已处理的执行中死亡不受影响）。历史深处、其后已有新消息的孤立调用不在
|
||||
// 此处修补（追加只落在尾部，位置语义不破）——那种形态由毒化轮转覆盖。
|
||||
func repairOrphanTail(cfg Config) error {
|
||||
// repairOrphanTail 返回本次补写的配对消息（JournalOnly 时供恢复摘要使用）。
|
||||
func repairOrphanTail(cfg Config) ([]llm.Message, error) {
|
||||
if cfg.Session == nil {
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
msgs, err := cfg.Session.Load()
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
tail := -1
|
||||
for i := len(msgs) - 1; i >= 0; i-- {
|
||||
@@ -770,7 +805,7 @@ func repairOrphanTail(cfg Config) error {
|
||||
break
|
||||
}
|
||||
if tail < 0 {
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
results := map[string]bool{}
|
||||
for _, m := range msgs {
|
||||
@@ -778,6 +813,7 @@ func repairOrphanTail(cfg Config) error {
|
||||
results[m.ToolCallID] = true
|
||||
}
|
||||
}
|
||||
var out []llm.Message
|
||||
for _, call := range msgs[tail].ToolCalls {
|
||||
if results[call.ID] {
|
||||
continue
|
||||
@@ -787,11 +823,32 @@ func repairOrphanTail(cfg Config) error {
|
||||
Content: crashInterrupted, IsError: true,
|
||||
}
|
||||
if err := persist(cfg, pair); err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
results[call.ID] = true
|
||||
out = append(out, pair)
|
||||
}
|
||||
return nil
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// recoveredNoteText 把本次恢复写入的配对摘成一段系统说明（注入上下文用；
|
||||
// 内容截断防爆，工具名与调用 ID 全列）。
|
||||
func recoveredNoteText(recovered []llm.Message) string {
|
||||
var b strings.Builder
|
||||
b.WriteString("上次会话中断的调用已完成恢复（这是系统说明，不是用户发言）：")
|
||||
for _, m := range recovered {
|
||||
content := m.Content
|
||||
if len([]rune(content)) > 120 {
|
||||
content = string([]rune(content)[:120]) + "…"
|
||||
}
|
||||
b.WriteString("\n- ")
|
||||
b.WriteString(m.Name)
|
||||
b.WriteString("（")
|
||||
b.WriteString(m.ToolCallID)
|
||||
b.WriteString("）：")
|
||||
b.WriteString(content)
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func commitToolIntent(cfg Config, call llm.ToolCall) error {
|
||||
|
||||
Reference in New Issue
Block a user