diff --git a/internal/agent/audio/model_provider_synthesizer.go b/internal/agent/audio/model_provider_synthesizer.go index e2f45b8da7..c2e66dea4b 100644 --- a/internal/agent/audio/model_provider_synthesizer.go +++ b/internal/agent/audio/model_provider_synthesizer.go @@ -98,7 +98,7 @@ func (m *modelProviderSynthesizer) Synthesize(ctx context.Context, req Synthesiz // Resolve tenant from canvas state when not on the request. tenantID := "" if state, err := runtime.GetStateFromContext(ctx); err == nil && state != nil { - if uid, ok := state.Sys["user_id"].(string); ok { + if uid, ok := state.GetSysString("user_id"); ok { tenantID = uid } } diff --git a/internal/agent/canvas/parallel_subgraph.go b/internal/agent/canvas/parallel_subgraph.go index 73c94cd294..2477f05581 100644 --- a/internal/agent/canvas/parallel_subgraph.go +++ b/internal/agent/canvas/parallel_subgraph.go @@ -183,10 +183,9 @@ func buildParallelItemWorkflow( if err != nil || localState == nil { return nil, fmt.Errorf("agent: parallel %q item collector: no canvas state in context", parallelID) } - out := map[string]any{ - "item": localState.Globals["__item__"], - "index": localState.Globals["__index__"], - } + item, _ := localState.GetGlobal("__item__") + index, _ := localState.GetGlobal("__index__") + out := map[string]any{"item": item, "index": index} for cpnID, bucket := range localState.Snapshot() { cp := make(map[string]any, len(bucket)) for k, v := range bucket { @@ -273,8 +272,8 @@ func buildParallelOuterWorkflow( if cloneErr != nil { return nil, fmt.Errorf("agent: parallel %q item %d: clone state: %w", key, index, cloneErr) } - localState.Globals["__item__"] = itemMap["item"] - localState.Globals["__index__"] = index + localState.SetGlobal("__item__", itemMap["item"]) + localState.SetGlobal("__index__", index) return runtime.WithState(ctx, localState), nil })) diff --git a/internal/agent/component/agent.go b/internal/agent/component/agent.go index fd73eda402..18dddb3bb4 100644 --- a/internal/agent/component/agent.go +++ b/internal/agent/component/agent.go @@ -1582,7 +1582,7 @@ func stringFromState(state *runtime.CanvasState, name string) (string, bool) { if state == nil { return "", false } - v, ok := state.Sys[name].(string) + v, ok := state.GetSysString(name) if !ok || strings.TrimSpace(v) == "" { return "", false } diff --git a/internal/agent/component/agent_mcp.go b/internal/agent/component/agent_mcp.go index d05aad18ca..d83869b9ba 100644 --- a/internal/agent/component/agent_mcp.go +++ b/internal/agent/component/agent_mcp.go @@ -46,12 +46,12 @@ func buildAgentMCPTools(ctx context.Context, db *gorm.DB, raw any, timeout time. if err != nil || state == nil || db == nil { return nil, fmt.Errorf("MCP server %q: missing runtime tenant or database", selection.ID) } - userID, _ := state.Sys["user_id"].(string) - canvasID, _ := state.Sys["canvas_id"].(string) + userID, _ := state.GetSysString("user_id") + canvasID, _ := state.GetSysString("canvas_id") if strings.TrimSpace(userID) == "" || strings.TrimSpace(canvasID) == "" { return nil, fmt.Errorf("MCP server %q: missing runtime user or canvas", selection.ID) } - tenantID, _ := state.Sys["tenant_id"].(string) + tenantID, _ := state.GetSysString("tenant_id") checker := permission.NewDatabaseChecker(db) err = checker.CheckDependency( ctx, diff --git a/internal/agent/component/begin.go b/internal/agent/component/begin.go index 18c9884058..f9393078aa 100644 --- a/internal/agent/component/begin.go +++ b/internal/agent/component/begin.go @@ -81,13 +81,13 @@ func (b *BeginComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map[str // Query: required to drive downstream components. query, _ := inputs["query"].(string) - state.Sys["query"] = query + state.SetSys("query", query) // Optional user_id โ€” present in interactive chat flows, absent in // background jobs. Always a string when set; cast failure silently // drops the value (mirrors Python's getattr fallback). if uid, ok := inputs["user_id"].(string); ok && uid != "" { - state.Sys["user_id"] = uid + state.SetSys("user_id", uid) } // Webhook payload injection. The webhook HTTP handler sets @@ -97,7 +97,7 @@ func (b *BeginComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map[str // same way they read sys.query / sys.user_id. The chat path never // sets this key, so existing tests stay green. if payload, ok := inputs["webhook_payload"].(map[string]any); ok && len(payload) > 0 { - state.Sys["webhook_payload"] = payload + state.SetSys("webhook_payload", payload) } // Passthrough: a shallow copy keeps the caller's map un-aliased. diff --git a/internal/agent/component/browser.go b/internal/agent/component/browser.go index 6f5dcd4086..5fab1fab65 100644 --- a/internal/agent/component/browser.go +++ b/internal/agent/component/browser.go @@ -203,7 +203,7 @@ func (b *BrowserComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map[s return nil, errors.New("browser: nil canvas state") } - tenantID, _ := state.Sys["tenant_id"].(string) + tenantID, _ := state.GetSysString("tenant_id") if tenantID == "" { return nil, errors.New("browser: tenant_id missing from canvas state (state.Sys[\"tenant_id\"])") } diff --git a/internal/agent/component/docs_generator.go b/internal/agent/component/docs_generator.go index 6c7902fce5..461372acd9 100644 --- a/internal/agent/component/docs_generator.go +++ b/internal/agent/component/docs_generator.go @@ -369,7 +369,7 @@ func storeAgentAttachment(ctx context.Context, docID string, payload []byte) (bo if err != nil || state == nil { return false, nil } - tenantID, _ := state.Sys["tenant_id"].(string) + tenantID, _ := state.GetSysString("tenant_id") tenantID = strings.TrimSpace(tenantID) if tenantID == "" { return false, nil diff --git a/internal/agent/component/llm.go b/internal/agent/component/llm.go index 99bd3b3af0..d64a5e21f8 100644 --- a/internal/agent/component/llm.go +++ b/internal/agent/component/llm.go @@ -811,7 +811,7 @@ func (c *LLMComponent) prepareChatRequest( // so a per-model "max_tokens" override in tenant_model.extra is honored. tenantID := "" if state, err := runtime.GetStateFromContext(ctx); err == nil && state != nil { - if tid, ok := state.Sys["tenant_id"].(string); ok { + if tid, ok := state.GetSysString("tenant_id"); ok { tenantID = tid } } @@ -1307,7 +1307,7 @@ func extractDataImages(values []string) []string { // and image data URIs. The caller is responsible for handling any // {sys.files} placeholder replacement in the prompts. func collectSysFiles(state *runtime.CanvasState) (textParts, imageURIs []string) { - files, ok := state.Sys["files"] + files, ok := state.GetSys("files") if !ok { return nil, nil } diff --git a/internal/agent/component/llm_credentials.go b/internal/agent/component/llm_credentials.go index ad7eff3895..f48b3ee547 100644 --- a/internal/agent/component/llm_credentials.go +++ b/internal/agent/component/llm_credentials.go @@ -28,7 +28,7 @@ func resolveTenantLLMConfig(ctx context.Context, db *gorm.DB, driver, modelID, a common.Debug("llm credentials: no canvas state in ctx") return apiKey, baseURL } - tid, _ := state.Sys["tenant_id"].(string) + tid, _ := state.GetSysString("tenant_id") if tid == "" { common.Debug("llm credentials: state.Sys has no tenant_id") return apiKey, baseURL @@ -197,7 +197,7 @@ func resolveTenantModelByID(ctx context.Context, db *gorm.DB, modelRef, apiKey, if err != nil || state == nil { return "", "", apiKey, baseURL, false, nil } - tid, _ := state.Sys["tenant_id"].(string) + tid, _ := state.GetSysString("tenant_id") if tid == "" { return "", "", apiKey, baseURL, false, nil } diff --git a/internal/agent/component/message.go b/internal/agent/component/message.go index 551290fdcc..1307c05fce 100644 --- a/internal/agent/component/message.go +++ b/internal/agent/component/message.go @@ -586,7 +586,7 @@ func stringFromStateSys(state *runtime.CanvasState, key string) string { if state == nil { return "" } - if v, ok := state.Sys[key]; ok { + if v, ok := state.GetSys(key); ok { if s, ok := v.(string); ok { return s } diff --git a/internal/agent/component/universe_a_wrappers.go b/internal/agent/component/universe_a_wrappers.go index 81ccd8cf4d..1e59fccb9b 100644 --- a/internal/agent/component/universe_a_wrappers.go +++ b/internal/agent/component/universe_a_wrappers.go @@ -243,8 +243,8 @@ func (c *retrievalComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map func checkCanvasRetrievalDependencies(ctx context.Context, db *gorm.DB, inputs map[string]any) error { if state, err := runtime.GetStateFromContext(ctx); err == nil && state != nil { - canvasID, _ := state.Sys["canvas_id"].(string) - userID, _ := state.Sys["user_id"].(string) + canvasID, _ := state.GetSysString("canvas_id") + userID, _ := state.GetSysString("user_id") if strings.TrimSpace(canvasID) == "" && strings.TrimSpace(userID) != "" { return fmt.Errorf("%w: canvas id missing from runtime state", permission.ErrInvalidPermission) } @@ -468,7 +468,7 @@ func resolveRetrievalDatasetID(ctx context.Context, db *gorm.DB, kbName string) } if state, err := runtime.GetStateFromContext(ctx); err == nil && state != nil { common.Debug("agent retrieval component: resolve dataset id context") - if tenantID, _ := state.Sys["tenant_id"].(string); tenantID != "" { + if tenantID, _ := state.GetSysString("tenant_id"); tenantID != "" { if kb, lookupErr := dao.NewKnowledgebaseDAO().GetByName(ctx, db, kbName, tenantID); lookupErr == nil && kb != nil { common.Debug("agent retrieval component: resolved dataset id by tenant") return kb.ID @@ -479,7 +479,7 @@ func resolveRetrievalDatasetID(ctx context.Context, db *gorm.DB, kbName string) common.Debug("agent retrieval component: tenant lookup missed") } } - if userID, _ := state.Sys["user_id"].(string); userID != "" { + if userID, _ := state.GetSysString("user_id"); userID != "" { kbs, lookupErr := dao.NewKnowledgebaseDAO().GetByNameInUserTenants(ctx, db, kbName, userID) if lookupErr != nil { common.Warn("agent retrieval component: resolve dataset id by tenant membership failed", diff --git a/internal/agent/component/variable_assigner.go b/internal/agent/component/variable_assigner.go index 11579ced64..cb5c75d2ea 100644 --- a/internal/agent/component/variable_assigner.go +++ b/internal/agent/component/variable_assigner.go @@ -404,16 +404,16 @@ func operate(state *runtime.CanvasState, oldVal any, op string, param any) (any, func writeVar(state *runtime.CanvasState, ref string, v any) error { switch { case ref == "item": - state.Globals["__item__"] = v + state.SetGlobal("__item__", v) return nil case ref == "index": - state.Globals["__index__"] = v + state.SetGlobal("__index__", v) return nil case strings.HasPrefix(ref, "sys."): - state.Sys[strings.TrimPrefix(ref, "sys.")] = v + state.SetSys(strings.TrimPrefix(ref, "sys."), v) return nil case strings.HasPrefix(ref, "env."): - state.Env[strings.TrimPrefix(ref, "env.")] = v + state.SetEnv(strings.TrimPrefix(ref, "env."), v) return nil } idx := strings.Index(ref, "@") diff --git a/internal/agent/component/variable_assigner_concurrency_test.go b/internal/agent/component/variable_assigner_concurrency_test.go new file mode 100644 index 0000000000..ac344364cb --- /dev/null +++ b/internal/agent/component/variable_assigner_concurrency_test.go @@ -0,0 +1,43 @@ +package component + +import ( + "sync" + "testing" + + "ragflow/internal/agent/runtime" +) + +func TestVariableAssignerConcurrentStateWrites(t *testing.T) { + state := runtime.NewCanvasState("race", "race") + refs := []string{"sys.shared", "env.shared", "item", "index"} + const workers = 16 + const writes = 500 + start := make(chan struct{}) + var wg sync.WaitGroup + for worker := 0; worker < workers; worker++ { + ref := refs[worker%len(refs)] + wg.Add(1) + go func(worker int, ref string) { + defer wg.Done() + <-start + for i := 0; i < writes; i++ { + if err := writeVar(state, ref, i); err != nil { + t.Errorf("writeVar(%q): %v", ref, err) + return + } + switch ref { + case "sys.shared": + state.GetSys("shared") + case "env.shared": + state.GetEnv("shared") + case "item": + state.GetGlobal("__item__") + case "index": + state.GetGlobal("__index__") + } + } + }(worker, ref) + } + close(start) + wg.Wait() +} diff --git a/internal/agent/retrievalbridge/memory.go b/internal/agent/retrievalbridge/memory.go index 1f78b4e62a..41d64052e9 100644 --- a/internal/agent/retrievalbridge/memory.go +++ b/internal/agent/retrievalbridge/memory.go @@ -82,7 +82,7 @@ func (a *MemoryAdapter) Search( } actorID := strings.TrimSpace(req.TenantID) if state, stateErr := runtime.GetStateFromContext(ctx); stateErr == nil && state != nil { - if userID, _ := state.Sys["user_id"].(string); strings.TrimSpace(userID) != "" { + if userID, _ := state.GetSysString("user_id"); strings.TrimSpace(userID) != "" { actorID = strings.TrimSpace(userID) } } diff --git a/internal/agent/runtime/state.go b/internal/agent/runtime/state.go index 2881c7ab62..2568b6b2c3 100644 --- a/internal/agent/runtime/state.go +++ b/internal/agent/runtime/state.go @@ -26,8 +26,7 @@ // Concurrency: a single sync.RWMutex guards every map in CanvasState // (plan ยง2.5 โ€” "start simple"). Helper methods (GetVar / SetVar / // ReadVars / Snapshot / etc.) lock internally; callers should not -// acquire OutputsLock unless they have a specific reason to extend a -// critical section. +// access the exported map fields directly from concurrent run paths. package runtime import ( @@ -719,6 +718,64 @@ func (s *CanvasState) SetGlobal(key string, val any) { s.Globals[key] = val } +// GetSys reads a value from the sys namespace. +func (s *CanvasState) GetSys(key string) (any, bool) { + if s == nil { + return nil, false + } + s.mu.RLock() + defer s.mu.RUnlock() + value, ok := s.Sys[key] + return value, ok +} + +// GetSysString reads a string value from the sys namespace. +func (s *CanvasState) GetSysString(key string) (string, bool) { + value, ok := s.GetSys(key) + if !ok { + return "", false + } + text, ok := value.(string) + return text, ok +} + +// SetSys writes a value into the sys namespace. +func (s *CanvasState) SetSys(key string, value any) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.Sys == nil { + s.Sys = make(map[string]any) + } + s.Sys[key] = value +} + +// GetEnv reads a value from the env namespace. +func (s *CanvasState) GetEnv(key string) (any, bool) { + if s == nil { + return nil, false + } + s.mu.RLock() + defer s.mu.RUnlock() + value, ok := s.Env[key] + return value, ok +} + +// SetEnv writes a value into the env namespace. +func (s *CanvasState) SetEnv(key string, value any) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.Env == nil { + s.Env = make(map[string]any) + } + s.Env[key] = value +} + // GetRetrievalChunks returns a snapshot of the chunks recorded in // state.Retrieval["chunks"]. The Retrieval map is the canvas-level // aggregate that the Retrieval tool populates during the ReAct loop; diff --git a/internal/agent/tool/canvas_ctx.go b/internal/agent/tool/canvas_ctx.go index 6b96701c13..a6938aa6c4 100644 --- a/internal/agent/tool/canvas_ctx.go +++ b/internal/agent/tool/canvas_ctx.go @@ -35,8 +35,8 @@ func CheckCanvasDependencies(ctx context.Context, db *gorm.DB, kind permission.R if err != nil || state == nil { return nil } - canvasID, _ := state.Sys["canvas_id"].(string) - userID, _ := state.Sys["user_id"].(string) + canvasID, _ := state.GetSysString("canvas_id") + userID, _ := state.GetSysString("user_id") canvasID = strings.TrimSpace(canvasID) userID = strings.TrimSpace(userID) if canvasID == "" { @@ -73,10 +73,10 @@ func canvasTenantID(ctx context.Context) string { if err != nil || state == nil { return "" } - if tenantID, _ := state.Sys["tenant_id"].(string); tenantID != "" { + if tenantID, _ := state.GetSysString("tenant_id"); tenantID != "" { return tenantID } - userID, _ := state.Sys["user_id"].(string) + userID, _ := state.GetSysString("user_id") return userID } @@ -96,7 +96,7 @@ func canvasDatasetIDs(ctx context.Context, explicit []string) []string { if err != nil || state == nil { return nil } - if id, _ := state.Sys["dataset_id"].(string); id != "" { + if id, _ := state.GetSysString("dataset_id"); id != "" { return []string{id} } return nil diff --git a/internal/agent/tool/retrieval.go b/internal/agent/tool/retrieval.go index 3edda0a4db..fdaf087b88 100644 --- a/internal/agent/tool/retrieval.go +++ b/internal/agent/tool/retrieval.go @@ -506,10 +506,10 @@ func retrievalTenantID(ctx context.Context) string { if err != nil || state == nil { return "" } - if tenantID, _ := state.Sys["tenant_id"].(string); tenantID != "" { + if tenantID, _ := state.GetSysString("tenant_id"); tenantID != "" { return tenantID } - userID, _ := state.Sys["user_id"].(string) + userID, _ := state.GetSysString("user_id") return userID } diff --git a/internal/agent/tool/retrieval_nlp.go b/internal/agent/tool/retrieval_nlp.go index f2b63209d2..1a761d7554 100644 --- a/internal/agent/tool/retrieval_nlp.go +++ b/internal/agent/tool/retrieval_nlp.go @@ -200,7 +200,7 @@ func (a *NLPRetrievalAdapter) Search(ctx context.Context, db *gorm.DB, req Retri // agent run, not req.UserID, which is a configurable per-user memory filter. userID := "" if state, stateErr := runtime.GetStateFromContext(ctx); stateErr == nil && state != nil { - userID, _ = state.Sys["user_id"].(string) + userID, _ = state.GetSysString("user_id") } userID = strings.TrimSpace(userID) checker := permission.NewDatabaseChecker(db) diff --git a/internal/binding/cpp/CMakeLists.txt b/internal/binding/cpp/CMakeLists.txt index cf1ff81a1b..12d206aed2 100644 --- a/internal/binding/cpp/CMakeLists.txt +++ b/internal/binding/cpp/CMakeLists.txt @@ -3,6 +3,8 @@ project(rag_tokenizer) set(CMAKE_CXX_STANDARD 23) +include(CTest) + # macOS dependency discovery โ€” Homebrew installs headers and libs under a # prefix that is NOT on the compiler's default search path (Apple Silicon: # /opt/homebrew, Intel: /usr/local). Linux is left completely untouched: @@ -174,3 +176,17 @@ set_target_properties(rag_analyzer_c_test PROPERTIES CXX_STANDARD 20 CXX_STANDARD_REQUIRED ON ) + +if(BUILD_TESTING) + add_executable(wordnet_lemmatizer_missing_exceptions_test + wordnet_lemmatizer_test.cpp + wordnet_lemmatizer.cpp + ) + target_include_directories(wordnet_lemmatizer_missing_exceptions_test PUBLIC "${CMAKE_SOURCE_DIR}") + set_target_properties(wordnet_lemmatizer_missing_exceptions_test PROPERTIES + CXX_STANDARD 20 + CXX_STANDARD_REQUIRED ON + ) + add_test(NAME wordnet_lemmatizer_missing_exceptions + COMMAND wordnet_lemmatizer_missing_exceptions_test) +endif() diff --git a/internal/binding/cpp/rag_analyzer_c_api.cpp b/internal/binding/cpp/rag_analyzer_c_api.cpp index 3b2252384e..925974e43c 100644 --- a/internal/binding/cpp/rag_analyzer_c_api.cpp +++ b/internal/binding/cpp/rag_analyzer_c_api.cpp @@ -29,7 +29,11 @@ void RAGAnalyzer_Destroy(RAGAnalyzerHandle handle) { int RAGAnalyzer_Load(RAGAnalyzerHandle handle) { if (!handle) return -1; RAGAnalyzer* analyzer = static_cast(handle); - return analyzer->Load(); + try { + return analyzer->Load(); + } catch (...) { + return -1; + } } void RAGAnalyzer_SetFineGrained(RAGAnalyzerHandle handle, bool fine_grained) { @@ -46,47 +50,58 @@ void RAGAnalyzer_SetEnablePosition(RAGAnalyzerHandle handle, bool enable_positio void RAGAnalyzer_SetLanguage(RAGAnalyzerHandle handle, const char* language) { if (!handle || !language) return; - RAGAnalyzer* analyzer = static_cast(handle); - analyzer->SetLanguage(std::string(language)); + try { + RAGAnalyzer* analyzer = static_cast(handle); + analyzer->SetLanguage(std::string(language)); + } catch (...) { + } } int RAGAnalyzer_Analyze(RAGAnalyzerHandle handle, const char* text, RAGTokenCallback callback) { if (!handle || !text || !callback) return -1; - RAGAnalyzer* analyzer = static_cast(handle); + try { + RAGAnalyzer* analyzer = static_cast(handle); - Term input; - input.text_ = std::string(text); + Term input; + input.text_ = std::string(text); - TermList output; - // Use the analyzer's internal state for fine_grained and enable_position - int ret = analyzer->Analyze(input, output, analyzer->fine_grained_, analyzer->enable_position_); + TermList output; + // Use the analyzer's internal state for fine_grained and enable_position + int ret = analyzer->Analyze(input, output, analyzer->fine_grained_, analyzer->enable_position_); - if (ret != 0) { - return ret; + if (ret != 0) { + return ret; + } + + // Call callback for each token + for (const auto& term : output) { + callback(term.text_.c_str(), term.text_.length(), term.word_offset_, term.end_offset_); + } + + return 0; + } catch (...) { + return -1; } - - // Call callback for each token - for (const auto& term : output) { - callback(term.text_.c_str(), term.text_.length(), term.word_offset_, term.end_offset_); - } - - return 0; } char* RAGAnalyzer_Tokenize(RAGAnalyzerHandle handle, const char* text) { if (!handle || !text) return nullptr; - RAGAnalyzer* analyzer = static_cast(handle); + try { + RAGAnalyzer* analyzer = static_cast(handle); - std::string result = analyzer->Tokenize(std::string(text)); + std::string result = analyzer->Tokenize(std::string(text)); - // Allocate memory for C string - char* c_result = static_cast(malloc(result.size() + 1)); - if (c_result) { - std::memcpy(c_result, result.c_str(), result.size() + 1); + // Allocate memory for C string + char* c_result = static_cast(malloc(result.size() + 1)); + if (c_result) { + std::memcpy(c_result, result.c_str(), result.size() + 1); + } + return c_result; + } catch (...) { + return nullptr; } - return c_result; } RAGTokenList* RAGAnalyzer_TokenizeWithPosition(RAGAnalyzerHandle handle, const char* text) { @@ -94,19 +109,27 @@ RAGTokenList* RAGAnalyzer_TokenizeWithPosition(RAGAnalyzerHandle handle, const c RAGAnalyzer* analyzer = static_cast(handle); - auto [tokens, positions] = analyzer->TokenizeWithPosition(std::string(text)); - if (analyzer->fine_grained_) { - std::string joined_tokens; - for (size_t i = 0; i < tokens.size(); ++i) { - if (i > 0) joined_tokens += " "; - joined_tokens += tokens[i]; - } + std::vector tokens; + std::vector> positions; + try { + auto result = analyzer->TokenizeWithPosition(std::string(text)); + tokens = std::move(result.first); + positions = std::move(result.second); + if (analyzer->fine_grained_) { + std::string joined_tokens; + for (size_t i = 0; i < tokens.size(); ++i) { + if (i > 0) joined_tokens += " "; + joined_tokens += tokens[i]; + } - std::vector fine_tokens; - std::vector> fine_positions; - analyzer->FineGrainedTokenizeWithPosition(joined_tokens, positions, fine_tokens, fine_positions); - tokens = std::move(fine_tokens); - positions = std::move(fine_positions); + std::vector fine_tokens; + std::vector> fine_positions; + analyzer->FineGrainedTokenizeWithPosition(joined_tokens, positions, fine_tokens, fine_positions); + tokens = std::move(fine_tokens); + positions = std::move(fine_positions); + } + } catch (...) { + return nullptr; } RAGTokenList* token_list = static_cast(malloc(sizeof(RAGTokenList))); @@ -176,49 +199,61 @@ uint32_t RAGToken_GetEndOffset(void* token) { char* RAGAnalyzer_FineGrainedTokenize(RAGAnalyzerHandle handle, const char* tokens) { if (!handle || !tokens) return nullptr; - RAGAnalyzer* analyzer = static_cast(handle); + try { + RAGAnalyzer* analyzer = static_cast(handle); - std::vector result; - analyzer->FineGrainedTokenize(std::string(tokens), result); + std::vector result; + analyzer->FineGrainedTokenize(std::string(tokens), result); - // Join results with space - std::string result_str; - for (size_t i = 0; i < result.size(); ++i) { - if (i > 0) result_str += " "; - result_str += result[i]; + // Join results with space + std::string result_str; + for (size_t i = 0; i < result.size(); ++i) { + if (i > 0) result_str += " "; + result_str += result[i]; + } + + // Allocate memory for C string + char* c_result = static_cast(malloc(result_str.size() + 1)); + if (c_result) { + std::memcpy(c_result, result_str.c_str(), result_str.size() + 1); + } + return c_result; + } catch (...) { + return nullptr; } - - // Allocate memory for C string - char* c_result = static_cast(malloc(result_str.size() + 1)); - if (c_result) { - std::memcpy(c_result, result_str.c_str(), result_str.size() + 1); - } - return c_result; } int32_t RAGAnalyzer_GetTermFreq(RAGAnalyzerHandle handle, const char* term) { if (!handle || !term) return 0; - RAGAnalyzer* analyzer = static_cast(handle); - return analyzer->Freq(term); + try { + RAGAnalyzer* analyzer = static_cast(handle); + return analyzer->Freq(term); + } catch (...) { + return 0; + } } char* RAGAnalyzer_GetTermTag(RAGAnalyzerHandle handle, const char* term) { if (!handle || !term) return nullptr; - RAGAnalyzer* analyzer = static_cast(handle); - std::string tag_result = analyzer->Tag(term); + try { + RAGAnalyzer* analyzer = static_cast(handle); + std::string tag_result = analyzer->Tag(term); - if (tag_result.empty()) { + if (tag_result.empty()) { + return nullptr; + } + + // Allocate memory for C string + char* c_result = static_cast(malloc(tag_result.size() + 1)); + if (c_result) { + std::memcpy(c_result, tag_result.c_str(), tag_result.size() + 1); + } + return c_result; + } catch (...) { return nullptr; } - - // Allocate memory for C string - char* c_result = static_cast(malloc(tag_result.size() + 1)); - if (c_result) { - std::memcpy(c_result, tag_result.c_str(), tag_result.size() + 1); - } - return c_result; } RAGAnalyzerHandle RAGAnalyzer_Copy(RAGAnalyzerHandle handle) { diff --git a/internal/binding/cpp/wordnet_lemmatizer.cpp b/internal/binding/cpp/wordnet_lemmatizer.cpp index f02cf37321..bd96644d8b 100644 --- a/internal/binding/cpp/wordnet_lemmatizer.cpp +++ b/internal/binding/cpp/wordnet_lemmatizer.cpp @@ -105,13 +105,14 @@ void WordNetLemmatizer::LoadExceptions() { const std::string &pos_name = pair.second; fs::path exc_path(root / (pos_name + ".exc")); + // Exception files are optional. Keep an empty map for each POS so + // Morphy can still apply substitutions when a file is absent. + exceptions_[pos_abbrev] = {}; std::ifstream file(exc_path.string()); if (!file.is_open()) { continue; } - exceptions_[pos_abbrev] = {}; - std::string line; while (std::getline(file, line)) { std::istringstream stream(line); @@ -131,7 +132,11 @@ void WordNetLemmatizer::LoadExceptions() { } std::vector WordNetLemmatizer::CollectSubstitutions(const std::vector &forms, const std::string &pos) { - const auto &substitutions = MORPHOLOGICAL_SUBSTITUTIONS.at(pos); + const auto substitutionsIt = MORPHOLOGICAL_SUBSTITUTIONS.find(pos); + if (substitutionsIt == MORPHOLOGICAL_SUBSTITUTIONS.end()) { + return {}; + } + const auto &substitutions = substitutionsIt->second; std::vector results; for (const auto &form : forms) { @@ -145,7 +150,11 @@ std::vector WordNetLemmatizer::CollectSubstitutions(const std::vect } std::vector WordNetLemmatizer::CollectSubstitutions(const std::string &form, const std::string &pos) { - const auto &substitutions = MORPHOLOGICAL_SUBSTITUTIONS.at(pos); + const auto substitutionsIt = MORPHOLOGICAL_SUBSTITUTIONS.find(pos); + if (substitutionsIt == MORPHOLOGICAL_SUBSTITUTIONS.end()) { + return {}; + } + const auto &substitutions = substitutionsIt->second; std::vector results; for (const auto &[old_suffix, new_suffix] : substitutions) { @@ -174,7 +183,11 @@ std::vector WordNetLemmatizer::FilterForms(const std::vector WordNetLemmatizer::Morphy(const std::string &form, const std::string &pos, bool check_exceptions) { - const auto &pos_exceptions = exceptions_.at(pos); + const auto exceptionsIt = exceptions_.find(pos); + if (exceptionsIt == exceptions_.end()) { + return {}; + } + const auto &pos_exceptions = exceptionsIt->second; // Check exceptions first if (check_exceptions && pos_exceptions.find(form) != pos_exceptions.end()) { diff --git a/internal/binding/cpp/wordnet_lemmatizer_test.cpp b/internal/binding/cpp/wordnet_lemmatizer_test.cpp new file mode 100644 index 0000000000..cb35dcda96 --- /dev/null +++ b/internal/binding/cpp/wordnet_lemmatizer_test.cpp @@ -0,0 +1,44 @@ +#include "wordnet_lemmatizer.h" + +#include +#include +#include +#include + +int main() { + namespace fs = std::filesystem; + + const auto path = fs::temp_directory_path() / + ("ragflow-wordnet-test-" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); + fs::create_directories(path); + const auto cleanup = [&path]() { + std::error_code error; + fs::remove_all(path, error); + }; + + for (const auto *pos : {"adj", "adv", "noun", "verb"}) { + std::ofstream index(path / (std::string("index.") + pos)); + if (std::string(pos) == "noun") { + index << "dog\n"; + } + } + for (const auto *pos : {"adj", "adv", "verb"}) { + std::ofstream exceptions(path / (std::string(pos) + ".exc")); + } + + try { + WordNetLemmatizer lemmatizer(path.string()); + const auto lemma = lemmatizer.Lemmatize("dogs"); + cleanup(); + if (lemma != "dog") { + std::cerr << "Lemmatize(dogs) = " << lemma << ", want dog\n"; + return 1; + } + } catch (const std::exception &error) { + cleanup(); + std::cerr << "Lemmatize threw with noun.exc missing: " << error.what() << '\n'; + return 1; + } + + return 0; +} diff --git a/internal/handler/agent_component.go b/internal/handler/agent_component.go index fab237ce67..977ed36364 100644 --- a/internal/handler/agent_component.go +++ b/internal/handler/agent_component.go @@ -222,7 +222,7 @@ func (h *AgentHandler) DebugComponent(c *gin.Context) { for key, value := range inputs { switch { case strings.HasPrefix(key, "sys.") && key != "sys.tenant_id" && key != "sys.user_id": - debugState.Sys[strings.TrimPrefix(key, "sys.")] = value + debugState.SetSys(strings.TrimPrefix(key, "sys."), value) case strings.Contains(key, "@"): if idx := strings.Index(key, "@"); idx > 0 && idx < len(key)-1 { debugState.SetVar(key[:idx], key[idx+1:], value) @@ -232,8 +232,8 @@ func (h *AgentHandler) DebugComponent(c *gin.Context) { // Permission checks in Retrieval use the authenticated caller and the // canvas whose component is being debugged. Set these after copying inputs // so request parameters cannot impersonate another user or canvas. - debugState.Sys["canvas_id"] = cv.ID - debugState.Sys["user_id"] = user.ID + debugState.SetSys("canvas_id", cv.ID) + debugState.SetSys("user_id", user.ID) invokeCtx := runtime.WithState(c.Request.Context(), debugState) outputs, err := runtime.TrackElapsed(name, func() (map[string]any, error) { diff --git a/internal/harness/core/agent_handoff.go b/internal/harness/core/agent_handoff.go index 63527f8e4f..bfe421a3d0 100644 --- a/internal/harness/core/agent_handoff.go +++ b/internal/harness/core/agent_handoff.go @@ -155,11 +155,10 @@ func runFlowAgentWithIsolatedSession(ctx context.Context, fa *flowAgent, input * isolatedSession := &runSession{ Values: make(map[string]any), - valuesMx: &sync.Mutex{}, + valuesMx: &sync.RWMutex{}, } if parentSession != nil { - isolatedSession.Values = parentSession.Values - isolatedSession.valuesMx = parentSession.valuesMx + isolatedSession.shareValuesFrom(parentSession) } rootInput := input @@ -200,11 +199,10 @@ func resumeFlowAgentWithIsolatedSession(ctx context.Context, fa *flowAgent, info isolatedSession := &runSession{ Values: make(map[string]any), - valuesMx: &sync.Mutex{}, + valuesMx: &sync.RWMutex{}, } if parentSession != nil { - isolatedSession.Values = parentSession.Values - isolatedSession.valuesMx = parentSession.valuesMx + isolatedSession.shareValuesFrom(parentSession) } // Restore events from deterministic transfer state for _, ev := range state.EventList { diff --git a/internal/harness/core/callback.go b/internal/harness/core/callback.go index d9afc8e712..9d3e89c326 100644 --- a/internal/harness/core/callback.go +++ b/internal/harness/core/callback.go @@ -175,7 +175,7 @@ func SetRunLocalValue(ctx context.Context, key string, val any) error { if rc == nil || rc.Session == nil { return errNotInAgentExec } - rc.Session.Values[key] = val + rc.Session.setValue(key, val) return nil } @@ -184,7 +184,7 @@ func GetRunLocalValue(ctx context.Context, key string) (any, bool, error) { if rc == nil || rc.Session == nil { return nil, false, errNotInAgentExec } - v, ok := rc.Session.Values[key] + v, ok := rc.Session.getValue(key) return v, ok, nil } @@ -193,7 +193,7 @@ func DeleteRunLocalValue(ctx context.Context, key string) error { if rc == nil || rc.Session == nil { return errNotInAgentExec } - delete(rc.Session.Values, key) + rc.Session.deleteValue(key) return nil } diff --git a/internal/harness/core/interrupt.go b/internal/harness/core/interrupt.go index e0d9a7e81a..911d2c8e80 100644 --- a/internal/harness/core/interrupt.go +++ b/internal/harness/core/interrupt.go @@ -200,8 +200,10 @@ func extractCheckpointTenant(ctx context.Context) string { return tid } if rc := getRunCtx(ctx); rc != nil && rc.Session != nil { - if tid, ok := rc.Session.Values[DefaultCheckpointTenantKey].(string); ok { - return tid + if value, ok := rc.Session.getValue(DefaultCheckpointTenantKey); ok { + if tid, ok := value.(string); ok { + return tid + } } } return "" diff --git a/internal/harness/core/react_agent.go b/internal/harness/core/react_agent.go index e516d11367..5310596a17 100644 --- a/internal/harness/core/react_agent.go +++ b/internal/harness/core/react_agent.go @@ -96,7 +96,7 @@ func resolveTemplate(tmpl string, ctx context.Context) string { return tmpl } result := tmpl - for k, v := range s.Values { + for k, v := range s.valuesSnapshot() { repl := fmt.Sprintf("{%s}", k) if sv, ok := v.(string); ok { result = strings.ReplaceAll(result, repl, sv) @@ -432,7 +432,7 @@ func setOutputToSession[M MessageType](ctx context.Context, msg M, key string) { if !isNilMessage(msg) { s := getSession(ctx) if s != nil { - s.Values[key] = extractTextContent(msg) + s.setValue(key, extractTextContent(msg)) } } } @@ -573,8 +573,10 @@ func getChatModelExecCtx(ctx context.Context) *reActExecCtx { return nil } // The exec ctx is stored on the run session or passed via context value - if ec, ok := rc.Session.Values["__exec_ctx"].(*reActExecCtx); ok { - return ec + if value, ok := rc.Session.getValue("__exec_ctx"); ok { + if ec, ok := value.(*reActExecCtx); ok { + return ec + } } return nil } diff --git a/internal/harness/core/runner.go b/internal/harness/core/runner.go index 93276faa00..7a90c76aa7 100644 --- a/internal/harness/core/runner.go +++ b/internal/harness/core/runner.go @@ -116,12 +116,10 @@ func resumeInternal[M MessageType](a TypedAgent[M], store CheckPointStore, ctx c o := getCommonOptions(nil, opts...) if o.sharedParentSession { if ps := getSession(ctx); ps != nil { - rc.Session.Values = ps.Values + rc.Session.shareValuesFrom(ps) } } - if rc.Session.Values == nil { - rc.Session.Values = make(map[string]any) - } + rc.Session.ensureValues() ctx = setRunCtx(ctx, rc) AddSessionValues(ctx, o.sessionValues) diff --git a/internal/harness/core/session.go b/internal/harness/core/session.go index 88147bc3eb..e973d4ef1f 100644 --- a/internal/harness/core/session.go +++ b/internal/harness/core/session.go @@ -111,14 +111,96 @@ type branchEvents struct { type runSession struct { mu sync.Mutex Values map[string]any - valuesMx *sync.Mutex + valuesMx *sync.RWMutex events []*eventWrapEntry BranchEvents *branchEvents TypedEvents any // *[]*typedAgentEventWrapper[M] for AgenticMessage path (gob-encodable) } func newRunSession() *runSession { - return &runSession{Values: make(map[string]any), valuesMx: &sync.Mutex{}} + return &runSession{Values: make(map[string]any), valuesMx: &sync.RWMutex{}} +} + +func (s *runSession) shareValuesFrom(parent *runSession) { + if s == nil || parent == nil || s == parent { + return + } + parent.valuesMx.RLock() + values := parent.Values + valuesMx := parent.valuesMx + parent.valuesMx.RUnlock() + + s.Values = values + s.valuesMx = valuesMx +} + +func (s *runSession) ensureValues() { + if s == nil { + return + } + s.valuesMx.Lock() + defer s.valuesMx.Unlock() + if s.Values == nil { + s.Values = make(map[string]any) + } +} + +func (s *runSession) setValue(key string, value any) { + if s == nil { + return + } + s.valuesMx.Lock() + defer s.valuesMx.Unlock() + if s.Values == nil { + s.Values = make(map[string]any) + } + s.Values[key] = value +} + +func (s *runSession) setValues(values map[string]any) { + if s == nil || len(values) == 0 { + return + } + s.valuesMx.Lock() + defer s.valuesMx.Unlock() + if s.Values == nil { + s.Values = make(map[string]any, len(values)) + } + for key, value := range values { + s.Values[key] = value + } +} + +func (s *runSession) getValue(key string) (any, bool) { + if s == nil { + return nil, false + } + s.valuesMx.RLock() + defer s.valuesMx.RUnlock() + value, ok := s.Values[key] + return value, ok +} + +func (s *runSession) valuesSnapshot() map[string]any { + if s == nil { + return nil + } + s.valuesMx.RLock() + defer s.valuesMx.RUnlock() + values := make(map[string]any, len(s.Values)) + for key, value := range s.Values { + values[key] = value + } + return values +} + +func (s *runSession) deleteValue(key string) { + if s == nil { + return + } + s.valuesMx.Lock() + defer s.valuesMx.Unlock() + delete(s.Values, key) } func (s *runSession) addEvent(event any) { @@ -265,11 +347,9 @@ func forkRunCtx(ctx context.Context) context.Context { copy(eventsCopy, parent.Session.events) parent.Session.mu.Unlock() - childSession := &runSession{ - events: eventsCopy, - Values: parent.Session.Values, // Share values map - valuesMx: parent.Session.valuesMx, - } + childSession := newRunSession() + childSession.events = eventsCopy + childSession.shareValuesFrom(parent.Session) childSession.BranchEvents = &branchEvents{ Parent: parent.Session.BranchEvents, @@ -369,9 +449,5 @@ func AddSessionValues(ctx context.Context, values map[string]any) { if rc == nil || rc.Session == nil || values == nil { return } - rc.Session.valuesMx.Lock() - defer rc.Session.valuesMx.Unlock() - for k, v := range values { - rc.Session.Values[k] = v - } + rc.Session.setValues(values) } diff --git a/internal/harness/core/session_values_concurrency_test.go b/internal/harness/core/session_values_concurrency_test.go new file mode 100644 index 0000000000..02b511360a --- /dev/null +++ b/internal/harness/core/session_values_concurrency_test.go @@ -0,0 +1,54 @@ +package core + +import ( + "context" + "fmt" + "sync" + "testing" + + "ragflow/internal/harness/core/schema" +) + +func TestForkRunCtxSharedValuesConcurrentAccess(t *testing.T) { + parent := setRunCtx(t.Context(), &runContext{Session: newRunSession()}) + const workers = 8 + const writes = 500 + contexts := make([]context.Context, workers) + for i := range contexts { + contexts[i] = forkRunCtx(parent) + } + + start := make(chan struct{}) + var wg sync.WaitGroup + for worker, ctx := range contexts { + wg.Add(1) + go func(worker int, ctx context.Context) { + defer wg.Done() + <-start + for i := 0; i < writes; i++ { + if err := SetRunLocalValue(ctx, "shared", fmt.Sprintf("%d-%d", worker, i)); err != nil { + t.Errorf("SetRunLocalValue: %v", err) + return + } + if _, _, err := GetRunLocalValue(ctx, "shared"); err != nil { + t.Errorf("GetRunLocalValue: %v", err) + return + } + _ = resolveTemplate("{shared}", ctx) + setOutputToSession(ctx, &schema.Message{Content: fmt.Sprintf("%d-%d", worker, i)}, "output") + if i%2 == 0 { + if err := DeleteRunLocalValue(ctx, "shared"); err != nil { + t.Errorf("DeleteRunLocalValue: %v", err) + return + } + } + } + }(worker, ctx) + } + close(start) + wg.Wait() + + if _, ok, err := GetRunLocalValue(parent, "output"); err != nil || !ok { + t.Fatalf("shared output value missing: ok=%v err=%v", ok, err) + } +} diff --git a/internal/service/agent.go b/internal/service/agent.go index 3eb52daac2..44ac8c4f9d 100644 --- a/internal/service/agent.go +++ b/internal/service/agent.go @@ -2308,11 +2308,11 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv if c.Globals != nil { for k, v := range c.Globals { if strings.HasPrefix(k, "sys.") { - state.Sys[strings.TrimPrefix(k, "sys.")] = v + state.SetSys(strings.TrimPrefix(k, "sys."), v) } else if strings.HasPrefix(k, "env.") { - state.Env[strings.TrimPrefix(k, "env.")] = v + state.SetEnv(strings.TrimPrefix(k, "env."), v) } else { - state.Globals[k] = v + state.SetGlobal(k, v) } } } @@ -2323,7 +2323,7 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv state.SetMemory(c.Memory) state.EnsureSysDate() query := agentRunQuery(userInput) - state.Sys["query"] = query + state.SetSys("query", query) state.AppendCurrentUser(query) state.AppendSysHistory("user: " + renderUserHistoryValue(query)) // Trusted caller identity is seeded through the same entry the debug @@ -2332,10 +2332,10 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv seedTenantID, _ := root["tenant_id"].(string) state.SeedIdentity(seedUserID, seedTenantID) if canvasID != "" { - state.Sys["canvas_id"] = canvasID + state.SetSys("canvas_id", canvasID) } if sessionID != "" { - state.Sys["session_id"] = sessionID + state.SetSys("session_id", sessionID) } if rawFiles, ok := root["files"].([]map[string]interface{}); ok && len(rawFiles) > 0 { // Only used for ParseAgentUploads (read-only); nil DocRemover means @@ -2347,7 +2347,7 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv s.markRunFailed(ctx2, runID, "parse files: "+ferr.Error()) return nil, fmt.Errorf("parse agent files: %w", ferr) } - state.Sys["files"] = files + state.SetSys("files", files) } state.IncrementConversationTurns() ctx2 = runtime.WithState(ctx2, state) @@ -2370,7 +2370,7 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv resumeData := root["__resume_data__"] delete(root, "__resume_interrupt_id__") delete(root, "__resume_data__") - state.Sys["query"] = "" + state.SetSys("query", "") ctx2 = compose.ResumeWithData(ctx2, resumeID, resumeData) } diff --git a/internal/service/agentic_pipeline.go b/internal/service/agentic_pipeline.go index 3e87d604ad..aa094ebb0e 100644 --- a/internal/service/agentic_pipeline.go +++ b/internal/service/agentic_pipeline.go @@ -142,8 +142,8 @@ func (s *ChatPipelineService) agenticRag( // its compile step; the agentic path has no canvas, so attach it here. runState := runtime.NewCanvasState("", chat.ID) if userID != "" { - runState.Sys["user_id"] = userID - runState.Sys["tenant_id"] = chat.TenantID + runState.SetSys("user_id", userID) + runState.SetSys("tenant_id", chat.TenantID) } runCtx = runtime.WithState(runCtx, runState) // A result is abandoned only when the CALLER is gone. runCtx is not the diff --git a/internal/service/memory_saver_adapter.go b/internal/service/memory_saver_adapter.go index 4597ab989f..73aabfbae5 100644 --- a/internal/service/memory_saver_adapter.go +++ b/internal/service/memory_saver_adapter.go @@ -52,8 +52,8 @@ func (a *memorySaverAdapter) Save(ctx context.Context, req component.MemorySaveR if stateErr != nil || state == nil { return fmt.Errorf("%w: canvas runtime state is required to use memory from an agent", permission.ErrInvalidPermission) } - canvasID, _ := state.Sys["canvas_id"].(string) - userID, _ := state.Sys["user_id"].(string) + canvasID, _ := state.GetSysString("canvas_id") + userID, _ := state.GetSysString("user_id") canvasID = strings.TrimSpace(canvasID) userID = strings.TrimSpace(userID) authorizationUserID = userID