diff --git a/.github/workflows/context-memory.yml b/.github/workflows/context-memory.yml new file mode 100644 index 00000000..173cd3fc --- /dev/null +++ b/.github/workflows/context-memory.yml @@ -0,0 +1,23 @@ +name: Context memory +on: + push: + paths: ['internal/agent/**', 'internal/app/**', 'internal/config/**', 'internal/database/**', 'internal/handler/**', 'internal/mcp/**', 'internal/multiagent/**', 'internal/project/**', 'internal/projectprompt/**', 'go.mod', 'go.sum', '.github/workflows/context-memory.yml'] + pull_request: + paths: ['internal/agent/**', 'internal/app/**', 'internal/config/**', 'internal/database/**', 'internal/handler/**', 'internal/mcp/**', 'internal/multiagent/**', 'internal/project/**', 'internal/projectprompt/**', 'go.mod', 'go.sum', '.github/workflows/context-memory.yml'] +permissions: + contents: read +jobs: + memory: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + runs-on: ${{ matrix.os }} + steps: + - run: git config --global core.autocrlf false + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + - name: Test compaction, resume, snapshots and scoped memory + run: go test -race ./internal/agent ./internal/app ./internal/database ./internal/handler ./internal/multiagent ./internal/project ./internal/projectprompt diff --git a/docs/zh-CN/blackboard-context-memory.md b/docs/zh-CN/blackboard-context-memory.md index d22faea1..84cdd19e 100644 --- a/docs/zh-CN/blackboard-context-memory.md +++ b/docs/zh-CN/blackboard-context-memory.md @@ -100,3 +100,13 @@ Eino 的压缩流程支持在 `Finalize` 中处理摘要并重写上下文,`Ca ## 追加代码审查验证 本轮进一步修复工具/助手文本伪装原始用户账本、无数据库路径时临时目录混用,以及项目绑定检查/写入竞态。回归覆盖工具与助手伪造标签、系统黑板嵌入片段、结构化来源 JSON 恢复及系统归一化、错误来源元数据拒绝、classic/native 无数据库路径的独立会话目录,以及切换/解绑项目时拒绝新增和更新且已有证据不变。该改动不迁移或改写历史快照,生产更新应与本轮代码验证结果对应。 + +## 恢复链路与边界回归 + +代理轨迹恢复统一通过同一解析器处理,只携带已知的 `summary` / `skills` 来源标记与系统消息中的结构化原始用户账本。`agent.ChatMessage` 以运行时字段传递这些信息,普通模型请求 JSON 不包含内部元数据;轨迹专用序列化保留允许的来源字段。恢复时不重放旧系统提示,仅按当前预算重新渲染宿主账本;普通用户引用摘要或标签仍作为用户原文。多段文本摘要恢复后继续保留工作进展与来源标记。测试使用生产相同的中文 ADK 设置,覆盖实际数据库加载和三轮「压缩→持久化→恢复→再次压缩」,包含 classic/native 转换。修复不能补回旧轨迹中已经丢失的来源信息,不改写历史记录。 + +不可变快照成功后,即使更新 `transcript.txt` 兼容入口失败,摘要及黑板仍保留成功快照的路径;日志单独报告兼容入口故障。快照本身失败时仍不宣称保存成功。 + +事实 key 的写入校验拒绝首尾空白,默认 SQL 范围过滤兼容已有 key 的 Unicode 空白,与 Go 判断一致;读取不会批量改名或迁移旧记录。索引成功构建为空时会移除旧索引块,其他系统内容及用户账本保持;数据库读取失败时继续保留原索引。 + +黑板短标签优先使用第 8 节中已经生成的证据检索线索;没有可用线索则沿用原先的标签取法。只更改索引标签,不改摘要第 1–7 节、完整正文或原始用户账本,不增加模型调用。新增 `Context memory` CI 在 macOS、Ubuntu、Windows 上直接运行 agent/app/database/handler/multiagent/project/projectprompt 的并发回归测试;Windows 不检查其文件系统不支持的 Unix 权限位。 diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 9d4bb00b..9ef4065e 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -122,6 +122,10 @@ type ChatMessage struct { // ModelFacingTrace is runtime-only metadata: true means Content was already the exact // payload seen at the model boundary and must be restored byte-for-byte. ModelFacingTrace bool `json:"-"` + // Internal provenance is restored from trusted trace metadata, never from + // message text. Keep it out of provider request JSON. + SummarizationKind string `json:"-"` + OriginalUserIntentEntries []string `json:"-"` } // MarshalJSON 自定义JSON序列化,将tool_calls中的arguments转换为JSON字符串 diff --git a/internal/agent/agent_trace.go b/internal/agent/agent_trace.go index c701a7f2..2190bfff 100644 --- a/internal/agent/agent_trace.go +++ b/internal/agent/agent_trace.go @@ -6,6 +6,8 @@ import ( ) const ModelFacingTraceVersionKey = "cyberstrike_model_facing_trace_version" +const SummarizationContentTypeKey = "_eino_summarization_content_type" +const OriginalUserIntentEntriesKey = "_csai_original_user_intent_entries" // IsModelFacingTraceJSON reports whether a persisted trace was produced from the final // model-boundary state. Legacy traces have no version marker and require one-time migration. @@ -50,9 +52,34 @@ func ParseTraceMessages(traceInputJSON string) ([]ChatMessage, error) { } msg.Role = role msg.ModelFacingTrace = modelFacing + extra, _ := msgMap["extra"].(map[string]interface{}) + if strings.EqualFold(role, "user") { + kind, _ := extra[SummarizationContentTypeKey].(string) + if kind == "summary" || kind == "skills" { + msg.SummarizationKind = kind + } + } + if strings.EqualFold(role, "system") { + msg.OriginalUserIntentEntries = traceUserIntentEntries(extra[OriginalUserIntentEntriesKey]) + } if content, ok := msgMap["content"].(string); ok { msg.Content = content } + // Native summaries may carry text only in user_input_multi_content. + if msg.Content == "" { + if parts, ok := msgMap["user_input_multi_content"].([]interface{}); ok { + var texts []string + for _, rawPart := range parts { + part, _ := rawPart.(map[string]interface{}) + if part["type"] == "text" { + if text, ok := part["text"].(string); ok { + texts = append(texts, text) + } + } + } + msg.Content = strings.Join(texts, "\n") + } + } if rc, ok := msgMap["reasoning_content"].(string); ok && strings.TrimSpace(rc) != "" { msg.ReasoningContent = rc } @@ -105,52 +132,95 @@ func ParseTraceMessages(traceInputJSON string) ([]ChatMessage, error) { return out, nil } -// ExtractLastUserTurnMessages 仅保留最后一次 user 提问起的消息(不含更早的用户轮次;跳过 system)。 -// 与「继续对话」续跑所用轨迹范围一致:当前任务轮次,而非整段多轮对话历史。 -func ExtractLastUserTurnMessages(msgs []ChatMessage) []ChatMessage { - if len(msgs) == 0 { - return msgs +// traceUserIntentEntries accepts only the host's structured ledger. Malformed +// metadata never falls back to parsing arbitrary system/assistant/tool text. +func traceUserIntentEntries(value interface{}) []string { + entries, ok := value.([]interface{}) + if !ok { + return nil } - lastUser := -1 - for i, m := range msgs { - if strings.EqualFold(m.Role, "user") { - lastUser = i + out := make([]string, 0, len(entries)) + for _, entry := range entries { + text, ok := entry.(string) + if !ok { + return nil } - } - if lastUser < 0 { - return msgs - } - trimmed := msgs[lastUser:] - out := make([]ChatMessage, 0, len(trimmed)) - for _, m := range trimmed { - if strings.EqualFold(m.Role, "system") { - continue + if strings.TrimSpace(text) != "" { + out = append(out, text) } - out = append(out, m) } return out } -// ExtractLastUserTurnTraceJSON 在 JSON 轨迹上裁剪为最后一次 user 起的片段(供落库格式直接处理)。 -func ExtractLastUserTurnTraceJSON(traceInputJSON string) string { - traceInputJSON = strings.TrimSpace(traceInputJSON) - if traceInputJSON == "" { - return traceInputJSON +// TraceMessagesForResume drops old instructions while carrying the structured +// human ledger separately. historyToMessages renders it under current budgets. +func TraceMessagesForResume(msgs []ChatMessage) []ChatMessage { + out := make([]ChatMessage, 0, len(msgs)) + for _, msg := range msgs { + if strings.EqualFold(msg.Role, "system") { + if len(msg.OriginalUserIntentEntries) == 0 { + continue + } + msg.Content = "" + msg.ToolCalls = nil + msg.ReasoningContent = "" + } + out = append(out, msg) } + return out +} + +// ExtractLastUserTurnMessages trims at the last human turn, not an internally +// generated user-role summary. Preserve the host ledger across the trim. +func ExtractLastUserTurnMessages(msgs []ChatMessage) []ChatMessage { + lastUser := 0 + var ledger *ChatMessage + for i := range msgs { + m := msgs[i] + if strings.EqualFold(m.Role, "user") && m.SummarizationKind == "" { + lastUser = i + } + if strings.EqualFold(m.Role, "system") && len(m.OriginalUserIntentEntries) > 0 { + ledger = &m + } + } + out := TraceMessagesForResume(msgs[lastUser:]) + if ledger != nil && lastUser > 0 { + out = append(TraceMessagesForResume([]ChatMessage{*ledger}), out...) + } + return out +} + +// ExtractLastUserTurnTraceJSON retains the host ledger and source markers in +// attack-chain history without mistaking a generated summary for a human turn. +func ExtractLastUserTurnTraceJSON(traceInputJSON string) string { var arr []map[string]interface{} if err := json.Unmarshal([]byte(traceInputJSON), &arr); err != nil { return traceInputJSON } - lastUser := -1 + lastUser := 0 + var ledger map[string]interface{} for i, m := range arr { - if r, _ := m["role"].(string); strings.EqualFold(r, "user") { + role, _ := m["role"].(string) + extra, _ := m["extra"].(map[string]interface{}) + kind, _ := extra[SummarizationContentTypeKey].(string) + if strings.EqualFold(role, "user") && kind != "summary" && kind != "skills" { lastUser = i } + if strings.EqualFold(role, "system") { + if entries := traceUserIntentEntries(extra[OriginalUserIntentEntriesKey]); len(entries) > 0 { + ledger = map[string]interface{}{"role": "system", "extra": map[string]interface{}{OriginalUserIntentEntriesKey: entries}} + } + } } if lastUser <= 0 { return traceInputJSON } - trimmed := arr[lastUser:] + trimmed := make([]map[string]interface{}, 0, len(arr)-lastUser+1) + if ledger != nil { + trimmed = append(trimmed, ledger) + } + trimmed = append(trimmed, arr[lastUser:]...) b, err := json.Marshal(trimmed) if err != nil { return traceInputJSON @@ -177,18 +247,34 @@ func MergeAssistantTraceOutput(msgs []ChatMessage, assistantOut string) []ChatMe return out } -// MessagesToTraceJSON 将消息带序列化为 JSON(跳过 system)。 +// MessagesToTraceJSON persists allowlisted internal provenance without adding +// unsupported fields to ChatMessage.MarshalJSON's provider request payload. func MessagesToTraceJSON(msgs []ChatMessage) (string, error) { - filtered := make([]ChatMessage, 0, len(msgs)) - for _, m := range msgs { - if strings.EqualFold(m.Role, "system") { - continue + filtered := make([]map[string]interface{}, 0, len(msgs)) + for _, m := range TraceMessagesForResume(msgs) { + raw, err := json.Marshal(m) + if err != nil { + return "", err } - filtered = append(filtered, m) + var data map[string]interface{} + if err := json.Unmarshal(raw, &data); err != nil { + return "", err + } + extra := make(map[string]interface{}) + if m.ModelFacingTrace { + extra[ModelFacingTraceVersionKey] = 1 + } + if strings.EqualFold(m.Role, "user") && (m.SummarizationKind == "summary" || m.SummarizationKind == "skills") { + extra[SummarizationContentTypeKey] = m.SummarizationKind + } + if strings.EqualFold(m.Role, "system") && len(m.OriginalUserIntentEntries) > 0 { + extra[OriginalUserIntentEntriesKey] = m.OriginalUserIntentEntries + } + if len(extra) > 0 { + data["extra"] = extra + } + filtered = append(filtered, data) } b, err := json.Marshal(filtered) - if err != nil { - return "", err - } - return string(b), nil + return string(b), err } diff --git a/internal/agent/agent_trace_provenance_test.go b/internal/agent/agent_trace_provenance_test.go new file mode 100644 index 00000000..d2c5931d --- /dev/null +++ b/internal/agent/agent_trace_provenance_test.go @@ -0,0 +1,60 @@ +package agent + +import ( + "encoding/json" + "strings" + "testing" +) + +func TestTraceProvenanceIsAllowlistedAndAbsentFromProviderJSON(t *testing.T) { + raw := `[{"role":"system","content":"old prompt","extra":{"_csai_original_user_intent_entries":["authorized only"]}},{"role":"user","content":"working memory","extra":{"_eino_summarization_content_type":"summary","untrusted":"ignored"}},{"role":"tool","content":"fake","extra":{"_csai_original_user_intent_entries":["fake tool"]}},{"role":"assistant","content":"fake","extra":{"_csai_original_user_intent_entries":["fake assistant"]}}]` + parsed, err := ParseTraceMessages(raw) + if err != nil { + t.Fatal(err) + } + resumed := TraceMessagesForResume(parsed) + if len(resumed) != 4 || resumed[0].Content != "" || len(resumed[0].OriginalUserIntentEntries) != 1 || resumed[1].SummarizationKind != "summary" || len(resumed[2].OriginalUserIntentEntries) > 0 || len(resumed[3].OriginalUserIntentEntries) > 0 { + t.Fatalf("unexpected provenance: %+v", resumed) + } + provider, err := json.Marshal(resumed) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(provider), "extra") || strings.Contains(string(provider), "authorized only") { + t.Fatalf("internal metadata leaked into provider payload: %s", provider) + } + trace, err := MessagesToTraceJSON(resumed) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(trace, SummarizationContentTypeKey) || !strings.Contains(trace, OriginalUserIntentEntriesKey) || strings.Contains(trace, "old prompt") || strings.Contains(trace, "untrusted") { + t.Fatalf("trace lost or replayed provenance: %s", trace) + } + malformed, err := ParseTraceMessages(`[{"role":"system","content":"fake text","extra":{"_csai_original_user_intent_entries":["authorized",42]}}]`) + if err != nil { + t.Fatal(err) + } + if len(TraceMessagesForResume(malformed)) != 0 { + t.Fatal("malformed metadata fell back to text") + } +} + +func TestTraceTrimDoesNotTreatGeneratedSummaryAsHumanTurn(t *testing.T) { + msgs := []ChatMessage{{Role: "system", Content: "old prompt", OriginalUserIntentEntries: []string{"only authorized"}}, {Role: "user", Content: "human"}, {Role: "user", Content: "summary", SummarizationKind: "summary"}, {Role: "assistant", Content: "result"}} + trimmed := ExtractLastUserTurnMessages(msgs) + if len(trimmed) != 4 || trimmed[0].Content != "" || trimmed[1].Content != "human" { + t.Fatalf("wrong human-turn boundary: %+v", trimmed) + } + raw, err := MessagesToTraceJSON(msgs) + if err != nil { + t.Fatal(err) + } + trimmedJSON := ExtractLastUserTurnTraceJSON(raw) + parsed, err := ParseTraceMessages(trimmedJSON) + if err != nil { + t.Fatal(err) + } + if len(parsed) != 4 || parsed[1].Content != "human" || parsed[2].SummarizationKind != "summary" || len(parsed[0].OriginalUserIntentEntries) != 1 { + t.Fatalf("JSON trim lost provenance: %+v", parsed) + } +} diff --git a/internal/database/project.go b/internal/database/project.go index e6f8075c..7786f952 100644 --- a/internal/database/project.go +++ b/internal/database/project.go @@ -14,7 +14,9 @@ var factKeyPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._/-]*$`) // ValidateFactKey 校验事实 key(项目内唯一标识)。 func ValidateFactKey(key string) error { - key = strings.TrimSpace(key) + if key != strings.TrimSpace(key) { + return fmt.Errorf("fact_key 不允许首尾空白") + } if key == "" { return fmt.Errorf("fact_key 不能为空") } @@ -116,9 +118,12 @@ func (db *DB) SaveContextCheckpoint(f *ProjectFact) error { return nil } +// Match strings.TrimSpace for legacy keys without migrating or rewriting them. +const factKeyTrimCharacters = "\t\n\v\f\r \u0085\u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000" + func contextCheckpointScopeSQL(conversationID string) (string, []interface{}) { - return " AND (substr(fact_key, 1, ?) != ? OR (source_conversation_id = ? AND ? != ''))", - []interface{}{len(ContextCheckpointKeyPrefix), ContextCheckpointKeyPrefix, strings.TrimSpace(conversationID), strings.TrimSpace(conversationID)} + return " AND (substr(trim(fact_key, ?), 1, ?) != ? OR (source_conversation_id = ? AND ? != ''))", + []interface{}{factKeyTrimCharacters, len(ContextCheckpointKeyPrefix), ContextCheckpointKeyPrefix, strings.TrimSpace(conversationID), strings.TrimSpace(conversationID)} } // CreateProject 创建项目。 diff --git a/internal/database/project_checkpoint_keys_test.go b/internal/database/project_checkpoint_keys_test.go new file mode 100644 index 00000000..16c9019a --- /dev/null +++ b/internal/database/project_checkpoint_keys_test.go @@ -0,0 +1,73 @@ +package database + +import ( + "fmt" + "go.uber.org/zap" + "path/filepath" + "strings" + "testing" +) + +func TestLegacyWhitespaceCheckpointKeysRespectConversationScope(t *testing.T) { + db, err := NewDB(filepath.Join(t.TempDir(), "legacy.db"), zap.NewNop()) + if err != nil { + t.Fatal(err) + } + defer db.Close() + p, err := db.CreateProject(&Project{Name: "scope"}) + if err != nil { + t.Fatal(err) + } + for i, space := range []string{" ", "\t", "\r\n", "\u00a0", "\u3000"} { + canonical := fmt.Sprintf("note/context-legacy%d", i) + f, err := db.UpsertProjectFact(&ProjectFact{ProjectID: p.ID, FactKey: canonical, SourceConversationID: "other", Summary: "LEGACY_PROGRESS"}) + if err != nil { + t.Fatal(err) + } + oldKey := space + canonical + space + // Simulate an already-stored legacy row, without migrating it on read. + if _, err := db.Exec("UPDATE project_facts SET fact_key=? WHERE id=?", oldKey, f.ID); err != nil { + t.Fatal(err) + } + if !IsContextCheckpointKey(oldKey) { + t.Fatal("helper did not identify legacy key") + } + if err := ValidateFactKey(oldKey); err == nil { + t.Fatal("write validation accepts noncanonical key") + } + if _, err := db.UpsertProjectFact(&ProjectFact{ProjectID: p.ID, FactKey: space + "finding/new"}); err == nil { + t.Fatal("write boundary stored noncanonical key") + } + } + if _, err := db.UpsertProjectFact(&ProjectFact{ProjectID: p.ID, FactKey: "finding/shared", Summary: "shared"}); err != nil { + t.Fatal(err) + } + for _, filter := range []ProjectFactListFilter{{RestrictContextCheckpoints: true, ContextConversationID: "current"}, {RestrictContextCheckpoints: true, ContextConversationID: "current", Search: "LEGACY_PROGRESS"}, {RestrictContextCheckpoints: true}} { + facts, err := db.ListProjectFacts(p.ID, filter, 100, 0) + if err != nil { + t.Fatal(err) + } + for _, f := range facts { + if IsContextCheckpointKey(f.FactKey) { + t.Fatalf("foreign legacy note crossed scope: %q", f.FactKey) + } + } + } + index, err := db.ListProjectFactsForIndex(p.ID, false, "current") + if err != nil || len(index) != 1 || index[0].FactKey != "finding/shared" { + t.Fatalf("legacy note leaked into index: %#v %v", index, err) + } + history, err := db.ListProjectFacts(p.ID, ProjectFactListFilter{Search: "LEGACY_PROGRESS"}, 100, 0) + if err != nil || len(history) != 5 { + t.Fatalf("historical records lost: %#v %v", history, err) + } + for _, f := range history { + if f.FactKey == strings.TrimSpace(f.FactKey) { + t.Fatal("read mutated legacy evidence") + } + } + own, err := db.ListProjectFactsForIndex(p.ID, false, "other") + if err != nil || len(own) != 6 { + t.Fatalf("legacy origin no longer visible to owner: %#v %v", own, err) + } +} diff --git a/internal/handler/agent.go b/internal/handler/agent.go index ab30f360..3c21fa26 100644 --- a/internal/handler/agent.go +++ b/internal/handler/agent.go @@ -2503,13 +2503,12 @@ func (h *AgentHandler) loadHistoryFromAgentTrace(conversationID string) ([]agent dataSource := "database_last_agent_trace" - var messagesArray []map[string]interface{} - if err := json.Unmarshal([]byte(traceInputJSON), &messagesArray); err != nil { + parsed, err := agent.ParseTraceMessages(traceInputJSON) + if err != nil { return nil, fmt.Errorf("解析代理轨迹 JSON 失败: %w", err) } - messageCount := len(messagesArray) - modelFacingTrace := agent.IsModelFacingTraceJSON(traceInputJSON) + messageCount := len(parsed) h.logger.Info("使用保存的代理轨迹恢复历史上下文", zap.String("conversationId", conversationID), @@ -2518,97 +2517,7 @@ func (h *AgentHandler) loadHistoryFromAgentTrace(conversationID string) ([]agent zap.Int("messageCount", messageCount), zap.Int("assistantOutSize", len(assistantOut)), ) - // fmt.Println("messagesArray:", messagesArray)//debug - - // 转换为Agent消息格式 - agentMessages := make([]agent.ChatMessage, 0, len(messagesArray)) - for _, msgMap := range messagesArray { - msg := agent.ChatMessage{} - msg.ModelFacingTrace = modelFacingTrace - - // 解析role - if role, ok := msgMap["role"].(string); ok { - msg.Role = role - } else { - continue // 跳过无效消息 - } - - // 跳过 system 消息(由 Eino Instruction 提供) - if msg.Role == "system" { - continue - } - - // 解析content - if content, ok := msgMap["content"].(string); ok { - msg.Content = content - } - // DeepSeek 思考模式:含工具调用的 assistant 须在后续请求中回传 reasoning_content - if rc, ok := msgMap["reasoning_content"].(string); ok && strings.TrimSpace(rc) != "" { - msg.ReasoningContent = rc - } - - // 解析tool_calls(如果存在) - if toolCallsRaw, ok := msgMap["tool_calls"]; ok && toolCallsRaw != nil { - if toolCallsArray, ok := toolCallsRaw.([]interface{}); ok { - msg.ToolCalls = make([]agent.ToolCall, 0, len(toolCallsArray)) - for _, tcRaw := range toolCallsArray { - if tcMap, ok := tcRaw.(map[string]interface{}); ok { - toolCall := agent.ToolCall{} - - // 解析ID - if id, ok := tcMap["id"].(string); ok { - toolCall.ID = id - } - - // 解析Type - if toolType, ok := tcMap["type"].(string); ok { - toolCall.Type = toolType - } - - // 解析Function - if funcMap, ok := tcMap["function"].(map[string]interface{}); ok { - toolCall.Function = agent.FunctionCall{} - - // 解析函数名 - if name, ok := funcMap["name"].(string); ok { - toolCall.Function.Name = name - } - - // 解析arguments(可能是字符串或对象) - if argsRaw, ok := funcMap["arguments"]; ok { - if argsStr, ok := argsRaw.(string); ok { - // 如果是字符串,解析为JSON - var argsMap map[string]interface{} - if err := json.Unmarshal([]byte(argsStr), &argsMap); err == nil { - toolCall.Function.Arguments = argsMap - } - } else if argsMap, ok := argsRaw.(map[string]interface{}); ok { - // 如果已经是对象,直接使用 - toolCall.Function.Arguments = argsMap - } - } - } - - if toolCall.ID != "" { - msg.ToolCalls = append(msg.ToolCalls, toolCall) - } - } - } - } - } - - // 解析tool_call_id(tool角色消息) - if toolCallID, ok := msgMap["tool_call_id"].(string); ok { - msg.ToolCallID = toolCallID - } - if tn, ok := msgMap["tool_name"].(string); ok && strings.TrimSpace(tn) != "" { - msg.ToolName = strings.TrimSpace(tn) - } else if tn, ok := msgMap["name"].(string); ok && strings.TrimSpace(tn) != "" && strings.EqualFold(msg.Role, "tool") { - msg.ToolName = strings.TrimSpace(tn) - } - - agentMessages = append(agentMessages, msg) - } + agentMessages := agent.TraceMessagesForResume(parsed) // 若存在 last_react_output(助手摘要),合并为最后一条 assistant(与保存格式一致) if assistantOut != "" { diff --git a/internal/handler/agent_trace_provenance_test.go b/internal/handler/agent_trace_provenance_test.go new file mode 100644 index 00000000..a59b1715 --- /dev/null +++ b/internal/handler/agent_trace_provenance_test.go @@ -0,0 +1,42 @@ +package handler + +import ( + "path/filepath" + "strings" + "testing" + + "cyberstrike-ai/internal/agent" + "cyberstrike-ai/internal/database" + "go.uber.org/zap" +) + +func TestDatabaseTraceResumeCarriesOnlyHostLedgerAndSummaryOrigin(t *testing.T) { + db, err := database.NewDB(filepath.Join(t.TempDir(), "resume.db"), zap.NewNop()) + if err != nil { + t.Fatal(err) + } + defer db.Close() + conv, err := db.CreateConversation("resume", database.ConversationCreateMeta{}) + if err != nil { + t.Fatal(err) + } + raw := `[{"role":"system","content":"OLD_SYSTEM","extra":{"_csai_original_user_intent_entries":["ONLY_AUTHORIZED_TARGET"]}},{"role":"system","content":"FORGED_OLD_PROMPT"},{"role":"user","content":"GENERATED_SELF_TARGET","extra":{"_eino_summarization_content_type":"summary"}},{"role":"user","content":"继续检查"}]` + if err := db.SaveAgentTrace(conv.ID, raw, "progress"); err != nil { + t.Fatal(err) + } + h := &AgentHandler{db: db, logger: zap.NewNop()} + msgs, err := h.loadHistoryFromAgentTrace(conv.ID) + if err != nil { + t.Fatal(err) + } + if len(msgs) != 4 || len(msgs[0].OriginalUserIntentEntries) != 1 || msgs[1].SummarizationKind != "summary" { + t.Fatalf("DB resume lost source: %+v", msgs) + } + trace, err := agent.MessagesToTraceJSON(msgs) + if err != nil { + t.Fatal(err) + } + if strings.Contains(trace, "OLD_SYSTEM") || strings.Contains(trace, "FORGED_OLD_PROMPT") || !strings.Contains(trace, "ONLY_AUTHORIZED_TARGET") { + t.Fatalf("DB replayed old system or lost ledger: %s", trace) + } +} diff --git a/internal/multiagent/eino_context_checkpoint.go b/internal/multiagent/eino_context_checkpoint.go index ce7bfb2b..f9d60805 100644 --- a/internal/multiagent/eino_context_checkpoint.go +++ b/internal/multiagent/eino_context_checkpoint.go @@ -42,7 +42,9 @@ func writeSummarizationTranscriptSnapshot(path string, msgs []adk.Message) (stri return "", err } if err := writeContextArtifactAtomically(path, body); err != nil { - return "", err + // The immutable evidence already exists; an expendable alias failure + // must not hide its successful retrieval path. + return snapshot, err } return snapshot, nil } @@ -158,6 +160,9 @@ func checkpointSummaryLabel(scope, summary string, cap int) string { if start := strings.Index(summary, summarizationSnapshotHeading); start >= 0 { summary = summary[:start] } + if clues := checkpointRetrievalClues(summary); clues != "" { + summary = clues + } var lines []string for _, line := range strings.Split(summary, "\n") { line = strings.TrimSpace(line) @@ -173,6 +178,34 @@ func checkpointSummaryLabel(scope, summary string, cap int) string { return truncateIndexLabel(label, cap) } +// Reuse section 8 without changing the original summary, ledger or model calls. +// Older summaries and an empty retrieval section retain the existing fallback. +func checkpointRetrievalClues(summary string) string { + inSection := false + var clues []string + for _, line := range strings.Split(summary, "\n") { + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "## 8.") && strings.Contains(line, "原始证据检索线索") { + inSection = true + continue + } + if !inSection { + continue + } + if strings.HasPrefix(line, "## ") || strings.HasPrefix(line, "internalkeep important finding"}}} out := finalizeSummarizationMemory(summary, original, filepath.Join(t.TempDir(), "transcript.txt"), db, pid, cid, "main", config.ProjectConfig{Enabled: true}, nil) text := out.UserInputMultiContent[0].Text - if strings.Contains(text, "internal") || !strings.Contains(text, "snapshots/transcript-") || !strings.Contains(text, "get_project_fact(note/context-") { + if strings.Contains(text, "internal") || !strings.Contains(text, filepath.Join("snapshots", "transcript-")) || !strings.Contains(text, "get_project_fact(note/context-") { t.Fatalf("continuation missing durable references: %s", text) } if !strings.Contains(summary.UserInputMultiContent[0].Text, "internal") { @@ -211,3 +212,56 @@ func TestCheckpointIndexLabelStaysShortWithLargeSummarySetting(t *testing.T) { t.Fatalf("checkpoint reinjects long history or loses key hint: %s", label) } } + +func TestSuccessfulEmptyFactIndexRemovesOldBlockButRetainsLedger(t *testing.T) { + db, pid, cid := checkpointTestProject(t) + if _, err := db.UpsertProjectFact(&database.ProjectFact{ProjectID: pid, FactKey: "finding/test", Summary: "OLD_INDEX_CONTENT"}); err != nil { + t.Fatal(err) + } + index, err := project.BuildFactIndexBlock(db, pid, config.ProjectConfig{Enabled: true}, cid) + if err != nil { + t.Fatal(err) + } + ledger := buildOriginalUserIntentLedgerMessage([]adk.Message{schema.UserMessage("ONLY_AUTHORIZED_TARGET")}, 96000, 16000) + msgs := mergeMessageIntoLeadingSystem([]adk.Message{schema.SystemMessage("KEEP_CURRENT_SYSTEM\n" + index + "\nKEEP_SUFFIX")}, ledger) + out := refreshFactIndexInMessages(msgs, db, pid, config.ProjectConfig{Enabled: true, FactIndexMaxRunes: 30}, nil, cid) + if strings.Contains(out[0].Content, project.FactIndexSectionStartMarker) || strings.Contains(out[0].Content, "OLD_INDEX_CONTENT") || !strings.Contains(out[0].Content, "KEEP_CURRENT_SYSTEM") || !strings.Contains(out[0].Content, "KEEP_SUFFIX") { + t.Fatalf("empty build retained index or removed other context: %s", out[0].Content) + } + got := buildOriginalUserIntentLedgerMessage(out, 96000, 16000) + if got == nil || !strings.Contains(got.Content, "ONLY_AUTHORIZED_TARGET") { + t.Fatal("ledger lost on index removal") + } + // A failed build is different: retain the valid prior block. + failed := refreshFactIndexInMessages(msgs, db, "missing-project", config.ProjectConfig{Enabled: true, FactIndexMaxRunes: 30}, nil, cid) + if failed[0].Content != msgs[0].Content { + t.Fatal("database error discarded old context") + } + if !strings.Contains(msgs[0].Content, "OLD_INDEX_CONTENT") { + t.Fatal("input message mutated") + } +} + +func TestCheckpointLabelUsesExistingEvidenceCluesWithoutChangingSummary(t *testing.T) { + const summary = "## 1. 授权范围与约束\n- authorized.example\n- COOKIE_SECRET\n- user method\n- user todo\n## 7. 当前进度、策略决策与下一步\n- progress\n## 8. 原始证据检索线索(追加)\n- authorized.example /login request Cookie clue\n- response 403 exact error\n" + label := checkpointSummaryLabel("main", summary, 10000) + if !strings.Contains(label, "/login") || !strings.Contains(label, "403") || strings.Contains(label, "COOKIE_SECRET") || strings.Contains(label, "user method") || len([]rune(label)) > 200 { + t.Fatalf("unhelpful or excessive index label: %s", label) + } + db, pid, cid := checkpointTestProject(t) + key, err := persistSummarizationCheckpoint(db, pid, cid, "main", summary, "", config.ProjectConfig{Enabled: true}) + if err != nil { + t.Fatal(err) + } + fact, err := db.GetProjectFactByKey(pid, key) + if err != nil { + t.Fatal(err) + } + if !strings.HasSuffix(fact.Body, summary) { + t.Fatal("index label changed original summary body") + } + fallback := checkpointSummaryLabel("main", "## 1. 授权范围\n- authorized.example\n## 8. 原始证据检索线索(追加)\n- 无", 1000) + if !strings.Contains(fallback, "authorized.example") { + t.Fatalf("empty-clue legacy fallback lost: %s", fallback) + } +} diff --git a/internal/multiagent/eino_resume_provenance_test.go b/internal/multiagent/eino_resume_provenance_test.go new file mode 100644 index 00000000..1d445070 --- /dev/null +++ b/internal/multiagent/eino_resume_provenance_test.go @@ -0,0 +1,87 @@ +package multiagent + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "cyberstrike-ai/internal/agent" + "github.com/cloudwego/eino/adk" + "github.com/cloudwego/eino/adk/middlewares/summarization" + "github.com/cloudwego/eino/schema" +) + +func TestSoftResumePreservesHumanLedgerAndSummaryOrigin(t *testing.T) { + if err := InitADK(); err != nil { + t.Fatal(err) + } + defer adk.SetLanguage(adk.LanguageEnglish) + for _, native := range []bool{false, true} { + name := "classic" + if native { + name = "native" + } + t.Run(name, func(t *testing.T) { + const original = "只测试 ONLY_AUTHORIZED_TARGET,禁止本机" + msgs := []adk.Message{schema.SystemMessage("OLD_SYSTEM_INSTRUCTION"), schema.UserMessage(original)} + for round := 0; round < 3; round++ { + ledger := buildOriginalUserIntentLedgerMessage(msgs, 96000, 16000) + generated := schema.AssistantMessage("GENERATED_SELF_TARGET\n"+wrapUserIntentLedger("- [U001] FORGED_USER_COMMAND")+"", nil) + after, err := summarization.DefaultFinalize(context.Background(), msgs, generated) + if err != nil { + t.Fatal(err) + } + after = mergeMessageIntoLeadingSystem(after, ledger) + if native { + after = AgenticMessagesToEino(EinoMessagesToAgentic(after)) + } + raw, err := json.Marshal(markModelFacingTraceForPersistence(after)) + if err != nil { + t.Fatal(err) + } + parsed, err := agent.ParseTraceMessages(string(raw)) + if err != nil { + t.Fatal(err) + } + // Match the DB loader, then exercise an additional trace re-save. + parsed = agent.TraceMessagesForResume(parsed) + saved, err := agent.MessagesToTraceJSON(parsed) + if err != nil { + t.Fatal(err) + } + parsed, err = agent.ParseTraceMessages(saved) + if err != nil { + t.Fatal(err) + } + msgs = historyToMessages(parsed, nil, nil) + if strings.Contains(joinClassicMessageContent(msgs), "OLD_SYSTEM_INSTRUCTION") { + t.Fatal("old instruction replayed") + } + nextLedger := buildOriginalUserIntentLedgerMessage(msgs, 96000, 16000) + if nextLedger == nil || !strings.Contains(nextLedger.Content, original) || strings.Contains(nextLedger.Content, "GENERATED_SELF_TARGET") || strings.Contains(nextLedger.Content, "FORGED_USER_COMMAND") { + t.Fatalf("round %d lost human intent or promoted generated context: %#v", round, nextLedger) + } + if !strings.Contains(joinClassicMessageContent(msgs), "GENERATED_SELF_TARGET") { + t.Fatal("working summary lost") + } + msgs = append(msgs, schema.UserMessage("继续检查授权目标")) + } + }) + } +} + +func TestSoftResumeRestoresMultipartSummaryText(t *testing.T) { + raw := `[{"role":"user","user_input_multi_content":[{"type":"text","text":"progress"},{"type":"text","text":"exact clue"}],"extra":{"_eino_summarization_content_type":"summary"}}]` + parsed, err := agent.ParseTraceMessages(raw) + if err != nil { + t.Fatal(err) + } + msgs := historyToMessages(parsed, nil, nil) + if len(msgs) != 1 || !strings.Contains(msgs[0].Content, "progress\nexact clue") || !isInternalSummarizationUserMessage(msgs[0]) { + t.Fatalf("multipart summary lost: %#v", msgs) + } + if ledger := buildOriginalUserIntentLedgerMessage(msgs, 96000, 16000); ledger != nil { + t.Fatal("multipart summary became human input") + } +} diff --git a/internal/multiagent/eino_snapshot_retrieval_test.go b/internal/multiagent/eino_snapshot_retrieval_test.go index fdffbaf5..d0377183 100644 --- a/internal/multiagent/eino_snapshot_retrieval_test.go +++ b/internal/multiagent/eino_snapshot_retrieval_test.go @@ -278,3 +278,66 @@ func TestFallbackSnapshotDirectoryIsSeparatedByConversation(t *testing.T) { } } } + +func TestAliasWriteFailureStillPublishesImmutableSnapshot(t *testing.T) { + for _, native := range []bool{false, true} { + t.Run(fmt.Sprintf("native=%v", native), func(t *testing.T) { + db, pid, cid := checkpointTestProject(t) + cfg := &config.Config{Project: config.ProjectConfig{Enabled: true}} + 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", cid, "summarization") + if err := os.MkdirAll(base, 0o700); err != nil { + t.Fatal(err) + } + // The immutable evidence can be saved, but the compatibility alias cannot be replaced. + if err := os.Mkdir(filepath.Join(base, "transcript.txt"), 0o700); 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, cid, db, pid, 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, cid, db, pid, 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) + snapshots, err := filepath.Glob(filepath.Join(base, "snapshots", "transcript-*.txt")) + if err != nil || len(snapshots) != 1 { + t.Fatalf("snapshot missing or duplicated: %v %v", snapshots, err) + } + if !strings.Contains(text, "usable continuation") || !strings.Contains(text, snapshots[0]) || !strings.Contains(text, summarizationSnapshotHeading) { + t.Fatalf("valid evidence lost retrieval entry: %s", text) + } + body, err := os.ReadFile(snapshots[0]) + if err != nil || !bytes.Equal(body, []byte(formatSummarizationTranscript(original))) { + t.Fatalf("evidence changed: %v", err) + } + facts, err := db.ListProjectFacts(pid, database.ProjectFactListFilter{}, 100, 0) + if err != nil || len(facts) != 1 || !strings.Contains(facts[0].Body, snapshots[0]) { + t.Fatalf("blackboard lost snapshot: %#v %v", facts, err) + } + }) + } +} diff --git a/internal/multiagent/eino_summarize.go b/internal/multiagent/eino_summarize.go index 377e09c3..5038c124 100644 --- a/internal/multiagent/eino_summarize.go +++ b/internal/multiagent/eino_summarize.go @@ -489,10 +489,8 @@ func refreshFactIndexInMessages(msgs []adk.Message, db *database.DB, projectID s return msgs } freshIndex = strings.TrimSpace(freshIndex) - if freshIndex == "" { - return msgs - } - + // A successful empty build means the old block must be removed. Errors + // above retain the previous index instead of silently discarding context. changed := false out := make([]adk.Message, len(msgs)) for i, msg := range msgs { @@ -500,7 +498,13 @@ func refreshFactIndexInMessages(msgs []adk.Message, db *database.DB, projectID s out[i] = msg continue } - newContent, ok := project.ReplaceFactIndexSection(msg.Content, freshIndex) + var newContent string + var ok bool + if freshIndex == "" { + newContent, ok = project.RemoveFactIndexSection(msg.Content) + } else { + newContent, ok = project.ReplaceFactIndexSection(msg.Content, freshIndex) + } if !ok { out[i] = msg continue diff --git a/internal/multiagent/eino_summarize_output.go b/internal/multiagent/eino_summarize_output.go index 95b881d9..61fbb5ca 100644 --- a/internal/multiagent/eino_summarize_output.go +++ b/internal/multiagent/eino_summarize_output.go @@ -6,6 +6,8 @@ import ( "regexp" "strings" + "cyberstrike-ai/internal/agent" + "github.com/cloudwego/eino/adk" "github.com/cloudwego/eino/schema" ) @@ -20,7 +22,7 @@ var ( const ( userIntentLedgerStartMarker = "" userIntentLedgerEndMarker = "" - userIntentLedgerExtraKey = "_csai_original_user_intent_entries" + userIntentLedgerExtraKey = agent.OriginalUserIntentEntriesKey summarizationSnapshotHeading = "## 原始上下文快照" summarizationTranscriptPathInstructionZh = "如果你需要压缩之前的具体细节(如精确的代码片段、错误消息或你生成的内容),完整的对话记录位于:%s" @@ -289,7 +291,7 @@ func isInternalSummarizationUserMessage(msg adk.Message) bool { if msg == nil { return false } - kind, _ := msg.Extra["_eino_summarization_content_type"].(string) + kind, _ := msg.Extra[agent.SummarizationContentTypeKey].(string) return kind == "summary" || kind == "skills" } diff --git a/internal/multiagent/runner.go b/internal/multiagent/runner.go index e853c01a..d99ca8e9 100644 --- a/internal/multiagent/runner.go +++ b/internal/multiagent/runner.go @@ -712,7 +712,29 @@ func historyToMessages(history []agent.ChatMessage, appCfg *config.Config, mwCfg if !h.ModelFacingTrace { content = normalizeRestoredUserContent(content, userContentMaxRunes) } - raw = append(raw, schema.UserMessage(content)) + msg := schema.UserMessage(content) + if h.SummarizationKind == "summary" || h.SummarizationKind == "skills" { + msg.Extra = map[string]interface{}{agent.SummarizationContentTypeKey: h.SummarizationKind} + } + raw = append(raw, msg) + } + case "system": + // Replay only the host's human ledger, never old system instructions. + if len(h.OriginalUserIntentEntries) > 0 { + msg := schema.SystemMessage("") + msg.Extra = map[string]interface{}{userIntentLedgerExtraKey: h.OriginalUserIntentEntries} + maxRunes := config.DefaultSummarizationUserIntentLedgerMaxRunes + entryMax := config.DefaultSummarizationUserIntentLedgerEntryMaxRunes + if mwCfg != nil { + maxRunes = mwCfg.SummarizationUserIntentLedgerMaxRunesEffective() + entryMax = mwCfg.SummarizationUserIntentLedgerEntryMaxRunesEffective() + } + if appCfg != nil { + maxRunes = minPositiveInt(maxRunes, modelFacingRuneBudget(appCfg.OpenAI.MaxTotalTokens, 0.20)) + } + if ledger := buildOriginalUserIntentLedgerMessage([]adk.Message{msg}, maxRunes, minPositiveInt(entryMax, maxRunes)); ledger != nil { + raw = append(raw, ledger) + } } case "assistant": toolSchema := chatToolCallsToSchema(h.ToolCalls) diff --git a/internal/project/blackboard_refresh.go b/internal/project/blackboard_refresh.go index 6a494727..b11b2d31 100644 --- a/internal/project/blackboard_refresh.go +++ b/internal/project/blackboard_refresh.go @@ -29,6 +29,20 @@ func ReplaceFactIndexSection(content, freshIndex string) (string, bool) { return content[:start] + freshIndex + content[end:], true } +// RemoveFactIndexSection removes only a complete host-delimited index after a +// successful build returns no block (e.g. a budget too small for framing). +func RemoveFactIndexSection(content string) (string, bool) { + start, ok := factIndexSectionStart(content) + if !ok { + return content, false + } + end, ok := factIndexSectionEnd(content, start) + if !ok || end <= start { + return content, false + } + return content[:start] + content[end:], true +} + // wrapFactIndexBlock 为 BuildFactIndexBlock 正文加上统一起止 HTML 注释边界。 func wrapFactIndexBlock(content string) string { content = strings.TrimSpace(content)