fix: preserve evidence across truncated indexes and failed writes

docs/zh-CN/blackboard-context-memory.md

@@ -54,6 +54,8 @@ project:
 
 黑板不需要再次调用模型压缩：同一会话/代理的自动续接正文更新为本次摘要，短标签最多 200 字符并保留可检索的证据类别，不自动注入完整正文或快照索引。每次生成新证据索引时跳过历史中已有的程序索引块，防止把检索提示本身重复当作证据。历史原文及其索引留在文件中，已确认事实不因续接记录更新而被改写。
 
+跳过和替换仅处理同一消息内带完整起止标记的索引块；模型复述时遗漏结束标记，不会导致后续消息中的原始证据被跳过或摘要正文被误删。快照写入失败时，不注入框架默认的 `transcript.txt` 路径提示，避免 Agent 读取不存在或属于上一轮的文件。
+
 遗留超长 summary 在展示索引时缩短为单行，库中的完整 summary/body 不改动。关系速览仅使用实际列入索引的事实，避免预算外事实继续通过路径视图占用上下文。置顶与事实类型的原有优先级保持不变。
 
 ## 实现依据与验证

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 {

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 {

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 = "<summary>usable continuation</summary>"
+			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)
+	}
+}

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 {