From e073709e8fa605c1cc6c7ec93ef3a74960d8f19e Mon Sep 17 00:00:00 2001 From: Sycun <165354365+Sycun@users.noreply.github.com> Date: Sun, 11 Oct 2026 23:58:07 +0800 Subject: [PATCH] fix: preserve evidence across truncated indexes and failed writes --- docs/zh-CN/blackboard-context-memory.md | 2 + internal/multiagent/eino_agentic_summarize.go | 3 +- internal/multiagent/eino_evidence_index.go | 57 ++++++--- .../multiagent/eino_evidence_index_test.go | 108 ++++++++++++++++++ internal/multiagent/eino_summarize.go | 4 +- 5 files changed, 154 insertions(+), 20 deletions(-) diff --git a/docs/zh-CN/blackboard-context-memory.md b/docs/zh-CN/blackboard-context-memory.md index 64df2c10..fe174d84 100644 --- a/docs/zh-CN/blackboard-context-memory.md +++ b/docs/zh-CN/blackboard-context-memory.md @@ -54,6 +54,8 @@ project: 黑板不需要再次调用模型压缩：同一会话/代理的自动续接正文更新为本次摘要，短标签最多 200 字符并保留可检索的证据类别，不自动注入完整正文或快照索引。每次生成新证据索引时跳过历史中已有的程序索引块，防止把检索提示本身重复当作证据。历史原文及其索引留在文件中，已确认事实不因续接记录更新而被改写。 +跳过和替换仅处理同一消息内带完整起止标记的索引块；模型复述时遗漏结束标记，不会导致后续消息中的原始证据被跳过或摘要正文被误删。快照写入失败时，不注入框架默认的 `transcript.txt` 路径提示，避免 Agent 读取不存在或属于上一轮的文件。 + 遗留超长 summary 在展示索引时缩短为单行，库中的完整 summary/body 不改动。关系速览仅使用实际列入索引的事实，避免预算外事实继续通过路径视图占用上下文。置顶与事实类型的原有优先级保持不变。 ## 实现依据与验证 diff --git a/internal/multiagent/eino_agentic_summarize.go b/internal/multiagent/eino_agentic_summarize.go index 0526beb1..049579be 100644 --- a/internal/multiagent/eino_agentic_summarize.go +++ b/internal/multiagent/eino_agentic_summarize.go @@ -170,7 +170,8 @@ func newEinoAgenticSummarizationMiddleware( TokenCounter: agenticTokenCounter, UserInstruction: einoSummarizeUserInstruction, EmitInternalEvents: emitInternalEvents, - TranscriptFilePath: transcriptPath, + // Finalize advertises only a successfully written immutable snapshot. + TranscriptFilePath: "", Retry: &summarization.TypedRetryConfig[*schema.AgenticMessage]{ MaxRetries: &retryMax, ShouldRetry: func(_ context.Context, _ *schema.AgenticMessage, err error) bool { diff --git a/internal/multiagent/eino_evidence_index.go b/internal/multiagent/eino_evidence_index.go index 0608a3f0..9a5b94a1 100644 --- a/internal/multiagent/eino_evidence_index.go +++ b/internal/multiagent/eino_evidence_index.go @@ -29,18 +29,17 @@ func buildTranscriptEvidenceIndex(snapshot, transcript string, limit int) string var out strings.Builder fmt.Fprintf(&out, "%s\n快照: %s\n历史快照目录: %s\n索引文件: %s.index.md\n", evidenceIndexMarker, snapshot, filepath.Dir(snapshot), snapshot) lines := strings.Split(transcript, "\n") + skip := make([]bool, len(lines)) + for _, span := range completeEvidenceIndexLineSpans(lines) { + for i := span[0]; i <= span[1]; i++ { + skip[i] = true + } + } for _, topic := range evidenceTopics { var locations []string count := 0 - inIndex := false for i, line := range lines { - if strings.HasPrefix(line, evidenceIndexMarker) { - inIndex = true - } - if inIndex { - if strings.TrimSpace(line) == evidenceIndexEnd { - inIndex = false - } + if skip[i] { continue } if topic.pattern.MatchString(line) { @@ -66,18 +65,40 @@ func buildTranscriptEvidenceIndex(snapshot, transcript string, limit int) string func appendEvidenceIndex(msgText, section string) string { // Replace a previously generated section rather than nesting evidence indexes. - for { - start := strings.Index(msgText, evidenceIndexMarker) - if start < 0 { - break + lines := strings.SplitAfter(msgText, "\n") + spans := completeEvidenceIndexLineSpans(lines) + var out strings.Builder + next := 0 + for _, span := range spans { + for _, line := range lines[next:span[0]] { + out.WriteString(line) } - end := strings.Index(msgText[start:], evidenceIndexEnd) - if end < 0 { - break - } - msgText = msgText[:start] + msgText[start+end+len(evidenceIndexEnd):] + next = span[1] + 1 } - return appendSummarizationSection(msgText, section) + for _, line := range lines[next:] { + out.WriteString(line) + } + return appendSummarizationSection(out.String(), section) +} + +// Only complete blocks within a single transcript turn are generated indexes. +// A truncated or nested heading must never swallow subsequent conversation data. +func completeEvidenceIndexLineSpans(lines []string) [][2]int { + var spans [][2]int + start := -1 + for i, line := range lines { + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "--- [") && strings.HasSuffix(line, "] ---") { + start = -1 + } + if line == evidenceIndexMarker { + start = i + } else if line == evidenceIndexEnd && start >= 0 { + spans = append(spans, [2]int{start, i}) + start = -1 + } + } + return spans } func appendEvidenceIndexToMessage(msg adk.Message, section string) adk.Message { diff --git a/internal/multiagent/eino_evidence_index_test.go b/internal/multiagent/eino_evidence_index_test.go index d3154d14..c6e50739 100644 --- a/internal/multiagent/eino_evidence_index_test.go +++ b/internal/multiagent/eino_evidence_index_test.go @@ -1,10 +1,14 @@ package multiagent import ( + "context" + "crypto/sha256" + "fmt" "os" "path/filepath" "strconv" "strings" + "sync" "testing" "cyberstrike-ai/internal/config" @@ -61,6 +65,90 @@ func TestEvidenceIndexLocatesSerializedEvidenceWithoutRepeatingSecrets(t *testin } } +func TestFailedTranscriptWriteDoesNotLeaveFrameworkAliasHint(t *testing.T) { + for _, native := range []bool{false, true} { + t.Run(fmt.Sprintf("native=%v", native), func(t *testing.T) { + cfg := &config.Config{} + cfg.Database.Path = filepath.Join(t.TempDir(), "runtime.db") + cfg.OpenAI.Model = "gpt-4o" + cfg.OpenAI.MaxTotalTokens = 5000 + emit := false + mwCfg := &config.MultiAgentEinoMiddlewareConfig{SummarizationEmitInternalEvents: &emit, SummarizationOutputReserveTokens: 1024} + ctx := context.Background() + base := filepath.Join(filepath.Dir(cfg.Database.Path), "conversation_artifacts", "failure", "summarization") + if err := os.MkdirAll(base, 0o700); err != nil { + t.Fatal(err) + } + // Make snapshot creation fail while the middleware can still create its base directory. + if err := os.WriteFile(filepath.Join(base, "snapshots"), []byte("blocked"), 0o600); err != nil { + t.Fatal(err) + } + original := []adk.Message{schema.SystemMessage("system"), schema.UserMessage(strings.Repeat("history ", 20000)), schema.UserMessage("continue")} + const summary = "usable continuation" + var after []adk.Message + if native { + mw, err := newEinoAgenticSummarizationMiddleware(ctx, &capturingAgenticChatModel{output: agenticAssistantTextMessage(summary)}, cfg, mwCfg, "failure", nil, "", nil) + if err != nil { + t.Fatal(err) + } + _, state, err := mw.BeforeModelRewriteState(ctx, &adk.TypedChatModelAgentState[*schema.AgenticMessage]{Messages: EinoMessagesToAgentic(original)}, nil) + if err != nil { + t.Fatal(err) + } + after = AgenticMessagesToEino(state.Messages) + } else { + mw, err := newEinoSummarizationMiddleware(ctx, &capturingClassicChatModel{output: &schema.Message{Role: schema.Assistant, Content: summary, ResponseMeta: &schema.ResponseMeta{FinishReason: "stop"}}}, cfg, mwCfg, "failure", nil, "", nil) + if err != nil { + t.Fatal(err) + } + _, state, err := mw.BeforeModelRewriteState(ctx, &adk.ChatModelAgentState{Messages: original}, nil) + if err != nil { + t.Fatal(err) + } + after = state.Messages + } + text := joinClassicMessageContent(after) + if !strings.Contains(text, "usable continuation") || strings.Contains(text, "transcript.txt") || strings.Contains(text, evidenceIndexMarker) { + t.Fatalf("failed snapshot has a misleading retrieval hint: %s", text) + } + }) + } +} + +func TestConcurrentSnapshotWritesRemainCompleteAndDeduplicated(t *testing.T) { + path := filepath.Join(t.TempDir(), "transcript.txt") + var wg sync.WaitGroup + errors := make(chan error, 20) + for i := 0; i < 20; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + msgs := []adk.Message{schema.UserMessage(fmt.Sprintf("Cookie: fixture-%d", i%2))} + snapshot, err := writeSummarizationTranscriptSnapshot(path, msgs) + if err == nil { + var body []byte + body, err = os.ReadFile(snapshot) + if err == nil && filepath.Base(snapshot) != fmt.Sprintf("transcript-%x.txt", sha256.Sum256(body)) { + err = fmt.Errorf("partial or mismatched snapshot: %s", snapshot) + } + } + errors <- err + }(i) + } + wg.Wait() + close(errors) + for err := range errors { + if err != nil { + t.Fatal(err) + } + } + files, _ := filepath.Glob(filepath.Join(filepath.Dir(path), "snapshots", "*.txt")) + indexes, _ := filepath.Glob(filepath.Join(filepath.Dir(path), "snapshots", "*.index.md")) + if len(files) != 2 || len(indexes) != 2 { + t.Fatalf("concurrent retries not deduplicated: %v %v", files, indexes) + } +} + func TestEvidenceIndexIsBoundedAndSkipsEarlierGeneratedIndexes(t *testing.T) { old := buildTranscriptEvidenceIndex("/old/transcript.txt", "Cookie: earlier", 3) transcript := old + "\n" + strings.Repeat("Cookie: current\n", 100) @@ -142,3 +230,23 @@ func TestEvidenceHintFitsBlackboardLabelBudget(t *testing.T) { t.Fatalf("unbounded label or lost evidence categories: %s", label) } } + +func TestTruncatedEvidenceIndexDoesNotHideLaterConversationEvidence(t *testing.T) { + transcript := formatSummarizationTranscript([]adk.Message{ + schema.UserMessage("old summary\n" + evidenceIndexMarker + "\nindex interrupted"), + {Role: schema.Tool, Content: "Cookie: sid=new-fixture\nError: later request failed"}, + }) + index := buildTranscriptEvidenceIndex("/new/snapshot.txt", transcript, 3) + if !strings.Contains(index, "Cookie / Set-Cookie：行") || !strings.Contains(index, "错误 / 失败：行") { + t.Fatalf("truncated old index suppressed later evidence: %s", index) + } +} + +func TestIncompleteEvidenceIndexDoesNotConsumeLaterSummaryWhenReplaced(t *testing.T) { + complete := buildTranscriptEvidenceIndex("/old/snapshot.txt", "Cookie: fixture", 3) + input := "summary\n" + evidenceIndexMarker + "\nincomplete\nkeep this progress\n" + complete + updated := appendEvidenceIndex(input, "new index") + if !strings.Contains(updated, "keep this progress") || strings.Contains(updated, "/old/snapshot.txt") { + t.Fatalf("replacing a completed index consumed unrelated summary: %s", updated) + } +} diff --git a/internal/multiagent/eino_summarize.go b/internal/multiagent/eino_summarize.go index 180a2347..439cee0f 100644 --- a/internal/multiagent/eino_summarize.go +++ b/internal/multiagent/eino_summarize.go @@ -220,7 +220,9 @@ func newEinoSummarizationMiddleware( TokenCounter: tokenCounter, UserInstruction: einoSummarizeUserInstruction, EmitInternalEvents: emitInternalEvents, - TranscriptFilePath: transcriptPath, + // Publish the immutable snapshot only after the host successfully writes it + // in Finalize; the framework otherwise advertises even a failed/stale alias. + TranscriptFilePath: "", Retry: &summarization.RetryConfig{ MaxRetries: &retryMax, ShouldRetry: func(_ context.Context, _ adk.Message, err error) bool {