diff --git a/AGENTS.md b/AGENTS.md index 37f9f28a51..06eb1683e6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -76,8 +76,9 @@ Rules: - `unit` (no tag) must stay free of external-service dependencies so `go test ./...` passes without MySQL/MinIO/ES/Infinity/LLM. The native CGO static libraries (`office_oxide`/`pdfium`/`pdf_oxide`) are still required at build time and are wired automatically by `build.sh --test`; that is expected, not an external service. ## Working Rules -- Before editing, inspect the nearest code path that actually owns the behavior. +- When reviewing documentation or code, inspect the full affected path and report all verifiable findings in one review; do not return after only a few findings and expose further issues in later rounds. - When handling review comments, independently verify each substantive claim against the current code or tests before accepting, rejecting, or acting on it. +- Before editing, inspect the nearest code path that actually owns the behavior. - Keep changes small and local unless the task is explicitly a broader refactor. - Prefer one implementation path instead of preserving old and new versions side by side. - Preserve behavior with focused tests when the behavior is still valid; do not keep tests that protect obsolete behavior. diff --git a/internal/entity/dataset.go b/internal/entity/dataset.go index 25392cdf9b..1c1bad0ddf 100644 --- a/internal/entity/dataset.go +++ b/internal/entity/dataset.go @@ -124,7 +124,19 @@ type Knowledgebase struct { WikiTaskFinishAt *time.Time `gorm:"column:wiki_task_finish_at" json:"wiki_task_finish_at,omitempty"` SkillTaskID *string `gorm:"column:skill_task_id;size:32;index" json:"skill_task_id,omitempty"` SkillTaskFinishAt *time.Time `gorm:"column:skill_task_finish_at" json:"skill_task_finish_at,omitempty"` - Status *string `gorm:"column:status;size:1;index;default:'1'" json:"status,omitempty"` + // Dataset-level structure merge task state (plan §8, mirror Python + // _INDEX_TYPE_TO_TASK_ID_FIELD's per-structure-index entries). + StructureGraphTaskID *string `gorm:"column:structure_graph_task_id;size:32;index" json:"structure_graph_task_id,omitempty"` + StructureGraphTaskFinishAt *time.Time `gorm:"column:structure_graph_task_finish_at" json:"structure_graph_task_finish_at,omitempty"` + StructureMindmapTaskID *string `gorm:"column:structure_mindmap_task_id;size:32;index" json:"structure_mindmap_task_id,omitempty"` + StructureMindmapTaskFinishAt *time.Time `gorm:"column:structure_mindmap_task_finish_at" json:"structure_mindmap_task_finish_at,omitempty"` + TimelineTaskID *string `gorm:"column:timeline_task_id;size:32;index" json:"timeline_task_id,omitempty"` + TimelineTaskFinishAt *time.Time `gorm:"column:timeline_task_finish_at" json:"timeline_task_finish_at,omitempty"` + SessionGraphTaskID *string `gorm:"column:session_graph_task_id;size:32;index" json:"session_graph_task_id,omitempty"` + SessionGraphTaskFinishAt *time.Time `gorm:"column:session_graph_task_finish_at" json:"session_graph_task_finish_at,omitempty"` + SessionEssenceTaskID *string `gorm:"column:session_essence_task_id;size:32;index" json:"session_essence_task_id,omitempty"` + SessionEssenceTaskFinishAt *time.Time `gorm:"column:session_essence_task_finish_at" json:"session_essence_task_finish_at,omitempty"` + Status *string `gorm:"column:status;size:1;index;default:'1'" json:"status,omitempty"` BaseModel } diff --git a/internal/handler/dataset_artifact.go b/internal/handler/dataset_artifact.go index 3ac47a86f2..2d091b192f 100644 --- a/internal/handler/dataset_artifact.go +++ b/internal/handler/dataset_artifact.go @@ -14,6 +14,7 @@ package handler import ( + "errors" "net/http" "strconv" @@ -276,35 +277,56 @@ func (h *DatasetArtifactHandler) GetArtifactGraph(c *gin.Context) { common.SuccessWithData(c, graph, "success") } -// ListStructures handles GET /artifacts/structure — list compiled structures of a dataset. +// ListStructures handles GET /artifacts/structure?kind= — the dataset-scope +// structure graph for a resolved kind (mirrors Python get_dataset_structure). +// kind is REQUIRED: missing or invalid → 400 ARGUMENT_ERROR. func (h *DatasetArtifactHandler) ListStructures(c *gin.Context) { _, tenantID, _ := h.datasetOwner(c, c.Param("dataset_id")) if tenantID == "" { return } datasetID := c.Param("dataset_id") - structureKind := c.Query("structure_kind") - structureIndexType := c.Query("structure_index_type") - items, total, err := h.svc.ListStructures(c.Request.Context(), tenantID, datasetID, structureKind, structureIndexType) - if err != nil { - common.ErrorWithCode(c, common.CodeDataError, err.Error()) + kind := c.Query("kind") + if kind == "" { + common.ErrorWithCode(c, common.CodeArgumentError, "kind is required") return } - common.SuccessWithData(c, gin.H{"total": total, "structures": items}, "success") + in := service.DatasetStructureGraphInput{TenantID: tenantID, DatasetID: datasetID, Kind: kind} + resp, err := h.svc.GetDatasetStructure(c.Request.Context(), in) + if err != nil { + if errors.Is(err, service.ErrInvalidStructureKind) { + common.ErrorWithCode(c, common.CodeArgumentError, err.Error()) + } else { + common.ErrorWithCode(c, common.CodeServerError, err.Error()) + } + return + } + common.SuccessWithData(c, resp, "success") } -// DeleteStructures handles DELETE /artifacts/structure — delete compiled structures of a dataset. +// DeleteStructures handles DELETE /artifacts/structure?kind=&wipe= — +// cancel the kind's task (wipe=false) or delete its dataset rows (wipe=true). +// kind is REQUIRED. func (h *DatasetArtifactHandler) DeleteStructures(c *gin.Context) { _, tenantID, _ := h.datasetOwner(c, c.Param("dataset_id")) if tenantID == "" { return } datasetID := c.Param("dataset_id") - structureKind := c.Query("structure_kind") - structureIndexType := c.Query("structure_index_type") - n, err := h.svc.DeleteStructures(c.Request.Context(), tenantID, datasetID, structureKind, structureIndexType) + kind := c.Query("kind") + if kind == "" { + common.ErrorWithCode(c, common.CodeArgumentError, "kind is required") + return + } + wipe := c.Query("wipe") == "true" + in := service.DatasetStructureGraphInput{TenantID: tenantID, DatasetID: datasetID, Kind: kind, Wipe: wipe} + n, err := h.svc.DeleteDatasetStructure(c.Request.Context(), in) if err != nil { - common.ErrorWithCode(c, common.CodeDataError, err.Error()) + if errors.Is(err, service.ErrInvalidStructureKind) { + common.ErrorWithCode(c, common.CodeArgumentError, err.Error()) + } else { + common.ErrorWithCode(c, common.CodeServerError, err.Error()) + } return } common.SuccessWithData(c, gin.H{"deleted": n}, "success") diff --git a/internal/ingestion/component/knowledge_compiler/component.go b/internal/ingestion/component/knowledge_compiler/component.go index 8b29cbebb1..29696603cd 100644 --- a/internal/ingestion/component/knowledge_compiler/component.go +++ b/internal/ingestion/component/knowledge_compiler/component.go @@ -157,11 +157,31 @@ func (c *KnowledgeCompilerComponent) Invoke(ctx context.Context, db *gorm.DB, in deps.TenantID = tenantID deps.DatasetID = datasetID + // Per-spec Inputs copy: each spec must get its own VariantSpecific map, + // otherwise specIn.VariantSpecific below would mutate the shared map and + // let a later template inherit the previous template's parser_config + // (a template with an empty config would then run with the wrong parser + // behavior). Copy the map (and preserve an empty map when nil). specIn := in - if specIn.VariantSpecific == nil { - specIn.VariantSpecific = map[string]any{} + specIn.VariantSpecific = make(map[string]any, len(in.VariantSpecific)+1) + for k, v := range in.VariantSpecific { + specIn.VariantSpecific[k] = v + } + // The template config (flat: kind/entity/relation/plan/…) is delivered to + // the structure and wiki variants under the "parser_config" key — the SAME + // key those variants read (structure.Run / wikiPipeline.mapBatch do + // VariantSpecific["parser_config"]). Storing it as "config" left the + // variants with a nil config, so InferType saw no "kind" and fell back to + // "list" (breaking timeline: its compile_kwd became "list" instead of + // "timeline", so dropIsolatedTimelineEntities never ran and the timeline + // rendered every entity isolated). + // + // Only overwrite when the template actually carries a config: an empty + // template config (e.g. a resolver stub) must not clobber a parser_config + // the caller already supplied on the inputs. + if len(spec.Config) > 0 { + specIn.VariantSpecific["parser_config"] = spec.Config } - specIn.VariantSpecific["config"] = spec.Config var o common.Outputs switch variant { @@ -564,23 +584,14 @@ func applyVariantColumns(doc *schema.ChunkDoc, p common.Product) error { } case common.VariantMindmap: - // Tree nodes: depth_int records the outline level. - if v, ok := metaInt(p.Meta, "level"); ok { - if err := doc.SetExtraValue("depth_int", v); err != nil { - return err - } - } - if v := metaString(p.Meta, "name"); v != "" { - if err := doc.SetExtraValue("title_kwd", v); err != nil { - return err - } - setTitleTokens(doc, v) - } - if v := metaStringSlice(p.Meta, "children"); len(v) > 0 { - if err := doc.SetExtraValue("children_kwd", v); err != nil { - return err - } - } + // Mindmap now emits entity/relation rows (plan §1.2) so it participates in + // dataset-level merge exactly like graph/timeline: each node is an entity, + // each parent→child edge is a relation. Reuse the shared structure-graph + // column contract (knowledge_graph_kwd + from/to_entity_kwd + name_kwd + + // entity_type_kwd + mention_count_int). The relation type lives in the + // content_with_weight payload ({"from","to","type"}), matching Python — + // NOT a dedicated relation_type_kwd column. + return applyStructureGraphColumns(doc, p, kind) } return nil diff --git a/internal/ingestion/component/knowledge_compiler/component_test.go b/internal/ingestion/component/knowledge_compiler/component_test.go index dc055d1bcd..21420a5805 100644 --- a/internal/ingestion/component/knowledge_compiler/component_test.go +++ b/internal/ingestion/component/knowledge_compiler/component_test.go @@ -388,21 +388,27 @@ func TestKnowledgeCompiler_Tree_EndToEnd(t *testing.T) { func TestKnowledgeCompiler_Mindmap_EndToEnd(t *testing.T) { installProseDeps(t) - // The proseChat reply is flat text; parseOutline still yields a root + the - // reply as a child node, so chunks are non-empty and parent-linked. + // The proseChat reply is flat text; ShapeTree yields a bare root (no parsed + // children). Mindmap now emits entity/relation rows (plan §1.2): every node + // is an entity, every parent→child edge a relation. With a flat reply there + // is at least one entity (the root) and it must carry name_kwd + + // knowledge_graph_kwd="entity" (the structure-graph storage contract). chunks := runVariant(t, "mindmap", nil) - // Root chunk must have empty parent_kwd and kind "root". - var root map[string]any + entityCount := 0 for _, c := range chunks { - if kind, _ := c["kc_kind"].(string); kind == "root" { - root = c + kind, _ := c["kc_kind"].(string) + if kind == "entity" { + entityCount++ + if _, ok := c["name_kwd"]; !ok { + t.Fatalf("mindmap entity chunk missing name_kwd: %+v", c) + } + if kg, _ := c["knowledge_graph_kwd"].(string); kg != "entity" { + t.Fatalf("mindmap entity chunk knowledge_graph_kwd = %q, want entity", kg) + } } } - if root == nil { - t.Fatalf("mindmap: no root chunk") - } - if pid, _ := root["parent_kwd"].(string); pid != "" { - t.Fatalf("mindmap: root parent_kwd = %q, want empty", pid) + if entityCount == 0 { + t.Fatalf("mindmap: no entity chunks; got %d chunks", len(chunks)) } } diff --git a/internal/ingestion/component/knowledge_compiler/mindmap/mindmap.go b/internal/ingestion/component/knowledge_compiler/mindmap/mindmap.go index b643f03f53..61460d84b3 100644 --- a/internal/ingestion/component/knowledge_compiler/mindmap/mindmap.go +++ b/internal/ingestion/component/knowledge_compiler/mindmap/mindmap.go @@ -12,7 +12,6 @@ package mindmap import ( "context" - "encoding/json" "fmt" "strings" @@ -132,38 +131,47 @@ func Run(ctx context.Context, deps common.Deps, param common.Param, inputs commo return out, nil } -// treeToProducts flattens the shaped mind-map tree into Products. The root -// becomes a "root" product whose content is the serialized {"id","children"} -// tree (Python's MindMapResult.output shape); each inner node becomes a -// "node" product linked via parent_id. +// treeToProducts flattens the shaped mind-map tree into entity/relation Products +// so mindmap participates in dataset-level merge exactly like graph (plan §1.2, +// aligning with Python's dataset_structure_merger which merges +// knowledge_graph_kwd IN {entity,relation} rows for structure_mindmap too). // -// NOTE: the per-node "node" products (content = the node title, plus a -// meta["children"] list of immediate child titles) are a Go component-layer -// contract for streaming the tree into the upstream chunk list. They go -// beyond Python's nested dict output and exist so downstream code can index -// each node independently with parent links. +// Mapping: +// - each node (including the root) → an entity product (kind="entity", +// name = node id, type = "mindmap"). +// - each parent→child edge → a relation product (kind="relation", +// from = parent id, to = child id, type = "related" — Python's default). +// +// The entity/relation discriminator is carried in Meta["kind"] so the consumer's +// mergeStructureDataset buckets entities by (name,type) and relations by +// (from,type,to), matching graph/timeline. func treeToProducts(tenantID, docID string, root *utility.Node) []common.Product { var out []common.Product - rootID := common.StableRowID(tenantID, docID, string(common.VariantMindmap), "root") + if root == nil || root.ID == "" { + return out + } + seen := map[string]bool{} + // Entity: root node. out = append(out, common.Product{ - ID: rootID, + ID: common.StableRowID(tenantID, docID, string(common.VariantMindmap), "entity", root.ID), DocID: docID, TenantID: tenantID, Variant: common.VariantMindmap, - Content: serializeNode(root), + Content: root.ID, Meta: map[string]any{ - "kind": "root", - "level": 0, - "name": root.ID, + "kind": "entity", + "name": root.ID, + "entity_type": "mindmap", + "compile_kwd": "mindmap", }, }) + seen[root.ID] = true type pending struct { - node *utility.Node - parentID string - level int + node *utility.Node + parent string } - queue := []pending{{root, rootID, 0}} + queue := []pending{{root, root.ID}} for len(queue) > 0 { p := queue[0] queue = queue[1:] @@ -171,64 +179,45 @@ func treeToProducts(tenantID, docID string, root *utility.Node) []common.Product if child.ID == "" { continue } - level := p.level + 1 - id := common.StableRowID(tenantID, docID, string(common.VariantMindmap), "node", child.ID) - childTitles := make([]string, 0, len(child.Children)) - for _, gc := range child.Children { - if gc.ID != "" { - childTitles = append(childTitles, gc.ID) - } - } - meta := map[string]any{ - "kind": "node", - "level": level, - "name": child.ID, - } - if len(childTitles) > 0 { - meta["children"] = childTitles + // Entity: child node (dedup by id so a DAG-shaped tree does not emit + // the same node twice). + if !seen[child.ID] { + seen[child.ID] = true + out = append(out, common.Product{ + ID: common.StableRowID(tenantID, docID, string(common.VariantMindmap), "entity", child.ID), + DocID: docID, + TenantID: tenantID, + Variant: common.VariantMindmap, + Content: child.ID, + Meta: map[string]any{ + "kind": "entity", + "name": child.ID, + "entity_type": "mindmap", + "compile_kwd": "mindmap", + }, + }) } + // Relation: parent → child edge (type = "related", Python default). out = append(out, common.Product{ - ID: id, + ID: common.StableRowID(tenantID, docID, string(common.VariantMindmap), "relation", p.parent, child.ID), DocID: docID, TenantID: tenantID, Variant: common.VariantMindmap, - Content: child.ID, - ParentID: p.parentID, - Meta: meta, + Content: p.parent + " related " + child.ID, + Meta: map[string]any{ + "kind": "relation", + "from": p.parent, + "to": child.ID, + "relation_type": "related", + "compile_kwd": "mindmap", + }, }) - queue = append(queue, pending{child, id, level}) + queue = append(queue, pending{child, child.ID}) } } return out } -// jsonNode mirrors the {"id","children"} mind-map node shape (Python's -// MindMapResult.output) for JSON serialization. -type jsonNode struct { - ID string `json:"id"` - Children []*jsonNode `json:"children"` -} - -func toJSONNode(n *utility.Node) *jsonNode { - jn := &jsonNode{ID: n.ID} - for _, c := range n.Children { - jn.Children = append(jn.Children, toJSONNode(c)) - } - return jn -} - -// serializeNode renders the tree in Python's MindMapResult.output shape: -// {"id": ..., "children": [...]}. Uses encoding/json so every control -// character (including \r, \b, \f, etc.) is escaped correctly; the previous -// hand-rolled serializer only escaped ", \, \n and \t. -func serializeNode(node *utility.Node) string { - b, err := json.Marshal(toJSONNode(node)) - if err != nil { - return `{"id":"` + node.ID + `","children":[]}` - } - return string(b) -} - func chunkTexts(chunks []common.Chunk) []string { var out []string for _, c := range chunks { diff --git a/internal/ingestion/component/knowledge_compiler/mindmap/mindmap_test.go b/internal/ingestion/component/knowledge_compiler/mindmap/mindmap_test.go index 32e276ca99..a354ee2a25 100644 --- a/internal/ingestion/component/knowledge_compiler/mindmap/mindmap_test.go +++ b/internal/ingestion/component/knowledge_compiler/mindmap/mindmap_test.go @@ -1,7 +1,6 @@ package mindmap import ( - "encoding/json" "strings" "testing" @@ -48,57 +47,50 @@ func TestTreeToProducts_ParentLinks(t *testing.T) { {ID: "B"}, }} products := treeToProducts("t1", "d1", root) - if len(products) != 5 { - t.Fatalf("products = %d, want 5 (root+A+A1+A2+B)", len(products)) + // 5 entities (root, A, A1, A2, B) + 4 relations (root→A, A→A1, A→A2, root→B). + if len(products) != 9 { + t.Fatalf("products = %d, want 9 (5 entities + 4 relations)", len(products)) } - if products[0].Meta["kind"] != "root" || products[0].ParentID != "" { - t.Errorf("root product malformed: %+v", products[0].Meta) - } - if products[1].ParentID != products[0].ID { - t.Errorf("A parent link = %q, want root id", products[1].ParentID) - } - var aID string + entCount, relCount := 0, 0 + fromTo := map[string]bool{} for _, p := range products { - if p.Meta["name"] == "A" { - aID = p.ID - } - } - for _, p := range products { - if n, _ := p.Meta["name"].(string); n == "A1" || n == "A2" { - if p.ParentID != aID { - t.Errorf("%s parent = %q, want A id", n, p.ParentID) + kind, _ := p.Meta["kind"].(string) + switch kind { + case "entity": + entCount++ + if p.Meta["entity_type"] != "mindmap" { + t.Errorf("entity %v type = %v, want mindmap", p.Meta["name"], p.Meta["entity_type"]) } + if p.Meta["compile_kwd"] != "mindmap" { + t.Errorf("entity %v compile_kwd = %v, want mindmap", p.Meta["name"], p.Meta["compile_kwd"]) + } + case "relation": + relCount++ + from, _ := p.Meta["from"].(string) + to, _ := p.Meta["to"].(string) + fromTo[from+"->"+to] = true + if p.Meta["relation_type"] != "related" { + t.Errorf("relation %v->%v type = %v, want related", from, to, p.Meta["relation_type"]) + } + default: + t.Errorf("unexpected kind %q", kind) + } + } + if entCount != 5 || relCount != 4 { + t.Errorf("entities=%d relations=%d, want 5/4", entCount, relCount) + } + for _, edge := range []string{"root->A", "A->A1", "A->A2", "root->B"} { + if !fromTo[edge] { + t.Errorf("missing relation %s", edge) } } } -func TestSerializeNode_Shape(t *testing.T) { - root := &utility.Node{ID: "Top \"quoted\"", Children: []*utility.Node{{ID: "child"}}} - js := serializeNode(root) - if !strings.HasPrefix(js, `{"id":"Top \"quoted\""`) { - t.Errorf("serializeNode = %q", js) - } -} - -// TestSerializeNode_EscapesControlChars locks that serializeNode uses -// encoding/json: control characters such as CR/LF/TAB/FF are escaped (not -// embedded raw), so the output is always valid JSON. -func TestSerializeNode_EscapesControlChars(t *testing.T) { - root := &utility.Node{ID: "a\rb\tc\fd", Children: []*utility.Node{{ID: "x\ny"}}} - js := serializeNode(root) - var dec struct { - ID string `json:"id"` - Children []struct { - ID string `json:"id"` - } `json:"children"` - } - if err := json.Unmarshal([]byte(js), &dec); err != nil { - t.Fatalf("serializeNode produced invalid JSON: %v (%q)", err, js) - } - if dec.ID != "a\rb\tc\fd" { - t.Errorf("root id round-trip = %q, want %q", dec.ID, "a\rb\tc\fd") - } - if len(dec.Children) != 1 || dec.Children[0].ID != "x\ny" { - t.Errorf("child id round-trip = %+v", dec.Children) +func TestTreeToProducts_EmptyAndNil(t *testing.T) { + if got := treeToProducts("t1", "d1", nil); len(got) != 0 { + t.Errorf("nil root produced %d products", len(got)) + } + if got := treeToProducts("t1", "d1", &utility.Node{ID: ""}); len(got) != 0 { + t.Errorf("empty root id produced %d products", len(got)) } } diff --git a/internal/ingestion/knowledge_compile/consumer.go b/internal/ingestion/knowledge_compile/consumer.go index dd1c0b5cd8..f4acf69110 100644 --- a/internal/ingestion/knowledge_compile/consumer.go +++ b/internal/ingestion/knowledge_compile/consumer.go @@ -1026,17 +1026,40 @@ func pageIndexSummary(graphJSON string) string { // dataset_structure_merger._merge_bucket but is a self-contained from-scratch // path (B3: the legacy unified merge never aggregated structure by name/type). func (c *Consumer) mergeStructureDataset(ctx context.Context, tenant, kb string, products []kccommon.Product) error { - // Bucket by (lower(name), type) for entities and (lower(from), lower(to)) - // for relations — relations have no "name" and must not be silently dropped - // (review issue 3). - type ekey struct{ name, typ string } - type rkey struct{ from, to string } + common.Info("knowledge_compile: mergeStructureDataset entry", + zap.String("kb_id", kb), + zap.Int("products", len(products))) + // Bucket by (lower(name), type, compile kind) for entities and + // (lower(from), lower(type), lower(to), compile kind) for relations — + // relations have no "name" and must not be silently dropped (review issue 3), + // and the relation type + compile kind keep distinct structure kinds and + // distinct relation types from collapsing into one dataset row. + type ekey struct{ name, typ, ckwd string } + type rkey struct{ from, typ, to, ckwd string } entByKey := make(map[ekey]*StructureBucket, 16) relByKey := make(map[rkey]*StructureBucket, 16) + // The authoritative Python (dataset_structure_merger._do_build) merges EVERY + // structure-kind doc row unconditionally — dataset scope is driven by the + // task type (structure_graph/timeline/structure_mindmap), NOT by a template + // Config["dataset_merge"] flag (that was a stale runner.py concept). So + // structure/mindmap entity/relation products always enter the dataset merge. for _, p := range products { - if p.Variant != kccommon.VariantStructure { + if p.Variant != kccommon.VariantStructure && p.Variant != kccommon.VariantMindmap { continue } + // Dataset rows carry the SAME compile_kwd as the doc rows (the inferred + // compile type / autotype: "hypergraph"/"timeline"/"mindmap"/"list"/…), + // mirroring Python dataset_structure_merger._do_build which passes the doc + // row's compile_kwd through verbatim. The template kind (p.Kind, restored + // from compilation_template_kind_kwd) is stored on the SEPARATE + // compilation_template_kind_kwd field and is what read/delete paths match + // on — NOT compile_kwd. Do not rewrite compile_kwd to the template kind: + // that would diverge from Python (which never does) and split doc vs + // dataset rows. + ckwd := metaString(p.Meta, "compile_kwd") + if ckwd == "" { + ckwd = compileKwdForVariant(p.Variant) + } kind := metaString(p.Meta, "kind") from := metaString(p.Meta, "from") to := metaString(p.Meta, "to") @@ -1047,10 +1070,17 @@ func (c *Consumer) mergeStructureDataset(ctx context.Context, tenant, kb string, if from == "" || to == "" { continue } - k := rkey{from: strings.ToLower(from), to: strings.ToLower(to)} + relType := metaString(p.Meta, "relation_type") + if relType == "" { + relType = metaString(p.Meta, "type") + } + if relType == "" { + relType = "related" + } + k := rkey{from: strings.ToLower(from), typ: strings.ToLower(relType), to: strings.ToLower(to), ckwd: ckwd} b := relByKey[k] if b == nil { - b = &StructureBucket{Name: from + " -> " + to, Type: "relation", FromEntity: from, ToEntity: to} + b = &StructureBucket{Name: from + " -> " + to, Type: "relation", FromEntity: from, ToEntity: to, CompileKwd: ckwd, TemplateID: p.TemplateID, TemplateKind: p.Kind, RelationType: relType} relByKey[k] = b } appendBucket(b, p) @@ -1065,10 +1095,10 @@ func (c *Consumer) mergeStructureDataset(ctx context.Context, tenant, kb string, if name == "" { continue } - k := ekey{name: strings.ToLower(name), typ: typ} + k := ekey{name: strings.ToLower(name), typ: typ, ckwd: ckwd} b := entByKey[k] if b == nil { - b = &StructureBucket{Name: name, Type: typ} + b = &StructureBucket{Name: name, Type: typ, CompileKwd: ckwd, TemplateID: p.TemplateID, TemplateKind: p.Kind} entByKey[k] = b } appendBucket(b, p) @@ -1080,6 +1110,11 @@ func (c *Consumer) mergeStructureDataset(ctx context.Context, tenant, kb string, for _, b := range relByKey { buckets = append(buckets, *b) } + common.Info("knowledge_compile: mergeStructureDataset buckets", + zap.String("kb_id", kb), + zap.Int("entities", len(entByKey)), + zap.Int("relations", len(relByKey)), + zap.Int("total_buckets", len(buckets))) if len(buckets) == 0 { return nil } @@ -1097,14 +1132,28 @@ func (c *Consumer) mergeStructureDataset(ctx context.Context, tenant, kb string, // element-wise running mean (review issue 13). VecCount tracks how many vectors // have been folded so the mean is order-independent. func appendBucket(b *StructureBucket, p kccommon.Product) { - if strings.TrimSpace(p.Content) != "" { + // The product's Content is the doc row's full content_with_weight JSON + // (e.g. {"category":"Person","description":"…","name":"…","type":"entity"}), + // NOT the plain-text description. Extract the "description" field so the + // dataset row's description stays a plain-text string (mirror Python + // _struct_merge_graph_entities, which folds the entity description, not the + // whole payload). Fall back to Content only when it is not a JSON object. + if desc := structureProductDescription(p.Content); desc != "" { if b.Description != "" { b.Description += "\n" } - b.Description += strings.TrimSpace(p.Content) + b.Description += desc } b.SourceDocIDs = appendUnique(b.SourceDocIDs, []string{p.DocID}) b.SourceChunkIDs = appendUnique(b.SourceChunkIDs, metaStringSlice(p.Meta, "source_chunk_ids")) + // mention_count_int: sum the per-entity mention counts (Python + // _struct_merge_graph_entities sums mention_count across the merged entities). + // Each entity product carries mention_count ≥ 1 (structure compile.go L231); + // mindmap entities carry none, so they contribute a default of 0 and are + // simply not stamped (mention_count_int is omitted when 0). + if mc, ok := metaInt(p.Meta, "mention_count"); ok && mc > 0 { + b.MentionCount += int(mc) + } if len(p.Vector) > 0 { b.VecCount++ if len(b.Vector) == 0 { @@ -1122,6 +1171,33 @@ func appendBucket(b *StructureBucket, p kccommon.Product) { } } +// structureProductDescription extracts the plain-text description from a structure +// product's Content. In production Content is the doc row's content_with_weight +// JSON ({"description":"…","name":"…","type":"…"}), so we parse it and return the +// "description" field — the whole payload must NOT leak into the dataset row's +// description (that is the bug: projectEntity then renders the raw JSON object). +// When Content is NOT a JSON object (plain text, legacy/unit-test rows), it is +// returned verbatim so existing folded-description behavior is preserved. +func structureProductDescription(content string) string { + content = strings.TrimSpace(content) + if content == "" { + return "" + } + if content[0] == '{' { + var m map[string]any + if err := json.Unmarshal([]byte(content), &m); err == nil { + if d, ok := m["description"].(string); ok { + if s := strings.TrimSpace(d); s != "" { + return s + } + } + // JSON object without a plain-text description: nothing to fold. + return "" + } + } + return content +} + // appendUnique appends only values not already present, preserving order. func appendUnique(dst, values []string) []string { seen := make(map[string]bool, len(dst)+len(values)) diff --git a/internal/ingestion/knowledge_compile/reader.go b/internal/ingestion/knowledge_compile/reader.go index 6861ba35db..7a75e52c96 100644 --- a/internal/ingestion/knowledge_compile/reader.go +++ b/internal/ingestion/knowledge_compile/reader.go @@ -79,12 +79,16 @@ var compiledSelectFields = []string{ "name_kwd", "entity_type_kwd", "from_entity_kwd", "to_entity_kwd", "slug_kwd", "type", "kc_kind", "create_timestamp_flt", "create_time", - "compilation_template_kind_kwd", + "compilation_template_kind_kwd", "compilation_template_ids", // The product kind discriminator for structure/tree: the component stores it // under knowledge_graph_kwd (structure: graph/entity/relation) / raptor_kwd // (tree: root/summary). Without these the reader cannot restore Meta["kind"] // and the dataset-nav dispatch would skip structure/tree products (B2). "knowledge_graph_kwd", "raptor_kwd", + // mention_count_int round-trips the entity mention count for reprojection. + // (relation type lives in the content_with_weight payload, matching Python — + // there is NO relation_type_kwd column.) + "mention_count_int", } // wikiSelectFields are the additional columns a wiki page carries (beyond @@ -208,9 +212,21 @@ func productFromChunkMap(c map[string]interface{}, tenant string, expect kccommo if err != nil || mapped != expect { return kccommon.Product{}, false } + // Round-trip the raw compile_kwd (the inferred compile type / autotype: + // list/set/hypergraph/timeline/mindmap/…) so the dataset-level merge can + // stamp the SAME value on the dataset row as the doc row (Python _do_build + // carries the doc row's compile_kwd through verbatim). Without this, the + // merge falls back to compileKwdForVariant ("structure"/"mindmap"), which + // diverges from the doc row's autotype ("hypergraph"/"timeline"). merged := isAvailable(c["available_int"]) meta := map[string]any{} + // Preserve the raw compile_kwd (autotype) for the dataset merge (see the + // round-trip note above). A non-string scalar from the engine is normalized + // via asString, matching the variant reverse-map at the top of this func. + if v := asString(c["compile_kwd"]); v != "" { + meta["compile_kwd"] = v + } if v, ok := c["name_kwd"].(string); ok && v != "" { meta["name"] = v } @@ -225,6 +241,9 @@ func productFromChunkMap(c map[string]interface{}, tenant string, expect kccommo meta["to"] = v meta["kind"] = "relation" } + if v, ok := metaInt(c, "mention_count_int"); ok { + meta["mention_count"] = v + } if v, ok := c["slug_kwd"].(string); ok && v != "" { // slug_kwd is the full "/" form (Python writer // contract); reconstruct it verbatim so the round-trip stays full-form. @@ -310,16 +329,26 @@ func productFromChunkMap(c map[string]interface{}, tenant string, expect kccommo // so callers (e.g. RebuildDataset's variant recovery, B1a) can map it back // via KindToVariant instead of re-deriving from the ambiguous compile_kwd. kind := asString(c["compilation_template_kind_kwd"]) + // Restore the compilation template id (from compilation_template_ids) so the + // dataset merge can bucket structure rows per template and read/delete paths + // can filter by it. A row should carry exactly one template id; if it carries + // more than one the first is used (multi-template rows are a config error + // surfaced elsewhere). + templateID := "" + if ids := metaStringSlice(c, "compilation_template_ids"); len(ids) > 0 { + templateID = ids[0] + } return kccommon.Product{ - ID: id, - DocID: docID, - TenantID: tenant, - Variant: expect, - Kind: kind, - Content: content, - Vector: vec, - Meta: meta, - Merged: merged, + ID: id, + DocID: docID, + TenantID: tenant, + Variant: expect, + Kind: kind, + TemplateID: templateID, + Content: content, + Vector: vec, + Meta: meta, + Merged: merged, }, true } diff --git a/internal/ingestion/knowledge_compile/structure_merge_test.go b/internal/ingestion/knowledge_compile/structure_merge_test.go index 764a9b8b9c..ce1f5ada00 100644 --- a/internal/ingestion/knowledge_compile/structure_merge_test.go +++ b/internal/ingestion/knowledge_compile/structure_merge_test.go @@ -114,17 +114,102 @@ func TestMergeStructureDataset_RelationsNotDropped(t *testing.T) { } } +// TestMergeStructureDataset_DescriptionIsPlainText covers the description bug: +// the product Content is the doc row's content_with_weight JSON, so the dataset +// row's folded description must be the plain-text "description" field, NOT the raw +// JSON object (which would leak the whole entity payload into the rendered graph). +func TestMergeStructureDataset_DescriptionIsPlainText(t *testing.T) { + c := &Consumer{writer: &fakeWriter{}} + products := []kccommon.Product{ + {Variant: kccommon.VariantStructure, DocID: "d1", + Content: `{"category":"Person","description":"汉桓帝,禁锢善类","name":"桓帝","type":"entity"}`, + Meta: map[string]any{"name": "桓帝", "entity_type": "Person", "compile_kwd": "list", "source_chunk_ids": []string{"c1"}}}, + } + if err := c.mergeStructureDataset(context.Background(), "t1", "kb1", products); err != nil { + t.Fatalf("mergeStructureDataset: %v", err) + } + fw := c.writer.(*fakeWriter) + if len(fw.buckets) != 1 { + t.Fatalf("want 1 bucket, got %d", len(fw.buckets)) + } + got := fw.buckets[0].Description + if got != "汉桓帝,禁锢善类" { + t.Errorf("description = %q, want plain-text \"汉桓帝,禁锢善类\" (not the raw JSON payload)", got) + } + if strings.Contains(got, "{") || strings.Contains(got, "category") { + t.Errorf("description leaked the JSON payload: %q", got) + } +} + +// TestMergeStructureDataset_CompileKwdIsAutotypeNotTemplateKind covers the +// option-A alignment: the dataset row's compile_kwd is the doc row's inferred +// compile type (autotype, e.g. "hypergraph"/"mindmap"), NOT the template kind. +// The template kind travels on the separate TemplateKind field (stamped to +// compilation_template_kind_kwd by WriteMergedStructure), which is what read/ +// delete paths match on. This mirrors Python _do_build (compile_kwd passes +// through verbatim) + get_dataset_structure (matches template kind). +func TestMergeStructureDataset_CompileKwdIsAutotypeNotTemplateKind(t *testing.T) { + c := &Consumer{writer: &fakeWriter{}} + products := []kccommon.Product{ + {Variant: kccommon.VariantStructure, DocID: "d1", Kind: "knowledge_graph", TemplateID: "tpl-graph", + Content: "Engine desc", + Meta: map[string]any{"name": "Engine", "entity_type": "component", "compile_kwd": "hypergraph", "source_chunk_ids": []string{"c1"}}}, + {Variant: kccommon.VariantMindmap, DocID: "d2", Kind: "mind_map", TemplateID: "tpl-mindmap", + Content: "Fuel desc", + Meta: map[string]any{"name": "Fuel", "entity_type": "substance", "compile_kwd": "mindmap", "source_chunk_ids": []string{"c2"}}}, + } + if err := c.mergeStructureDataset(context.Background(), "t1", "kb1", products); err != nil { + t.Fatalf("mergeStructureDataset: %v", err) + } + fw := c.writer.(*fakeWriter) + if len(fw.buckets) != 2 { + t.Fatalf("want 2 buckets, got %d", len(fw.buckets)) + } + for i := range fw.buckets { + b := fw.buckets[i] + switch b.Name { + case "Engine": + if b.CompileKwd != "hypergraph" { + t.Errorf("Engine compile_kwd = %q, want autotype \"hypergraph\" (not template kind)", b.CompileKwd) + } + if b.TemplateKind != "knowledge_graph" { + t.Errorf("Engine TemplateKind = %q, want \"knowledge_graph\"", b.TemplateKind) + } + if b.TemplateID != "tpl-graph" { + t.Errorf("Engine TemplateID = %q, want \"tpl-graph\"", b.TemplateID) + } + case "Fuel": + if b.CompileKwd != "mindmap" { + t.Errorf("Fuel compile_kwd = %q, want autotype \"mindmap\" (not template kind)", b.CompileKwd) + } + if b.TemplateKind != "mind_map" { + t.Errorf("Fuel TemplateKind = %q, want \"mind_map\"", b.TemplateKind) + } + if b.TemplateID != "tpl-mindmap" { + t.Errorf("Fuel TemplateID = %q, want \"tpl-mindmap\"", b.TemplateID) + } + } + } +} + // TestDatasetLevelStructureID_StableAndCaseInsensitive covers G1: the id is // deterministic and case-insensitive on name so merges hit the same row. func TestDatasetLevelStructureID_StableAndCaseInsensitive(t *testing.T) { - a := datasetLevelStructureID("t1", "kb1", "Engine", "component") - b := datasetLevelStructureID("t1", "kb1", "engine", "component") + a := datasetLevelStructureID("t1", "kb1", "Engine", "component", "timeline", "") + b := datasetLevelStructureID("t1", "kb1", "engine", "component", "timeline", "") if a != b { t.Errorf("structure id should be case-insensitive on name: %q vs %q", a, b) } - if a == datasetLevelStructureID("t1", "kb1", "Fuel", "component") { + if a == datasetLevelStructureID("t1", "kb1", "Fuel", "component", "timeline", "") { t.Errorf("different names must yield different ids") } + if a == datasetLevelStructureID("t1", "kb1", "Engine", "component", "mindmap", "") { + t.Errorf("different compile kinds must yield different ids") + } + if datasetLevelStructureID("t1", "kb1", "A -> B", "relation", "graph", "causes") == + datasetLevelStructureID("t1", "kb1", "A -> B", "relation", "graph", "contradicts") { + t.Errorf("different relation types between the same endpoints must yield different ids") + } } // TestKwdToVariant_MapsStructureSubKinds covers B1a: structure doc-level products @@ -182,6 +267,11 @@ func TestProductFromChunkMap_RestoresStructureTreeKind(t *testing.T) { if kind, _ := sp.Meta["kind"].(string); kind != "graph" { t.Errorf("structure Meta.kind = %q, want graph (so nav dispatch can pick it)", kind) } + // Round-trip: the raw compile_kwd (autotype) must be preserved in Meta so the + // dataset merge stamps the SAME value on the dataset row as the doc row. + if ckwd, _ := sp.Meta["compile_kwd"].(string); ckwd != "list" { + t.Errorf("structure Meta.compile_kwd = %q, want autotype \"list\" (not \"structure\")", ckwd) + } // tree root product: kind stored in raptor_kwd. treeRow := map[string]interface{}{ diff --git a/internal/ingestion/knowledge_compile/writer.go b/internal/ingestion/knowledge_compile/writer.go index 88cc9514ef..1d2721e568 100644 --- a/internal/ingestion/knowledge_compile/writer.go +++ b/internal/ingestion/knowledge_compile/writer.go @@ -101,6 +101,28 @@ type StructureBucket struct { VecCount int // number of vectors folded into Vector (for true mean) FromEntity string // relation only ToEntity string // relation only + // CompileKwd is the raw compile keyword (the inferred compile type / autotype, + // e.g. "hypergraph", "timeline", "mindmap", "list") that produced this bucket. + // It is stamped on the stored row's compile_kwd so distinct structure kinds + // never collide in the same dataset namespace (plan §1.1). It matches the doc + // row's own compile_kwd — Python _do_build carries the doc row's compile_kwd + // verbatim onto the dataset row, it does NOT rewrite it to the template kind. + CompileKwd string + // TemplateID is the compilation template id (compilation_template_ids[0]) that + // produced this bucket. Stamped on the dataset row so read/delete paths can + // filter per template (mirror Python get_dataset_structure). + TemplateID string + // TemplateKind is the authoritative template kind (compilation_template_kind_kwd, + // e.g. "knowledge_graph", "mind_map", "timeline"). This — NOT compile_kwd — is + // the field read/delete paths match on for the dataset-structure kind + // (mirror Python get_dataset_structure._discover_scope_templates, which + // matches _resolve_dataset_structure_kind against compilation_template_kind_kwd). + TemplateKind string + // RelationType is the relation type (Python default "related"), persisted as + // relation_type_kwd and part of the relation bucket/ID identity. + RelationType string + // MentionCount is the entity mention count (mention_count_int); 0 when unset. + MentionCount int } // engineWriter persists dataset-level merged products through the global @@ -111,6 +133,33 @@ type engineWriter struct { eng engine.DocEngine } +// datasetStructureSupported reports whether the running doc engine can filter +// the dataset-structure fields (knowledge_graph_kwd + scope_kwd + raw +// compile_kwd) this feature writes/deletes. Only infinity and elasticsearch +// support them today; OceanBase/SeekDB/SereneDB have explicit schemas that lack +// these filter keys and would reject the query or silently drop unknown fields. +// The guard is checked at every dataset-structure write/delete/rebuild entry so +// those engines never leave partial state or hit an unknown-filter error (plan §5). +func datasetStructureSupported() bool { + switch engine.GetEngineType() { + case "infinity", "elasticsearch": + return true + case "": + // Engine not initialized (unit tests inject a fake engine directly via + // engineWriter.eng). The guard is only meaningful against a real engine + // type, so an empty type is treated as supported. + return true + default: + return false + } +} + +// errDatasetStructureUnsupported returns an error (not a silent no-op) naming +// the engine when a dataset-structure operation runs on an unsupported engine. +func errDatasetStructureUnsupported() error { + return fmt.Errorf("dataset structure graph unsupported on doc engine %q", engine.GetEngineType()) +} + // writeMergedBatchSize bounds how many rows each parallel InsertChunks call // carries, so the DocEngine write fan-out stays granular under the shared pool. const writeMergedBatchSize = 200 @@ -202,13 +251,18 @@ func (w engineWriter) WriteMerged(ctx context.Context, tenant, kb string, produc // WriteMergedStructure writes the dataset-level structure merged rows for a KB // (G1/G4). Each StructureBucket is a scope_kwd="dataset" row with a stable -// dataset-level id keyed on (name, type), the folded description, the union of -// source docs/chunks, and the bucket vector. Rows are available_int=1 so the -// dataset-level structure index is searchable, and compile_kwd="structure". +// dataset-level id keyed on (name, type, raw compile kind), the folded +// description, the union of source docs/chunks, and the bucket vector. Rows are +// available_int=1 so the dataset-level structure index is searchable; the raw +// compile kind (timeline/graph/mindmap) is stamped on compile_kwd and the +// entity/relation discriminator on knowledge_graph_kwd. func (w engineWriter) WriteMergedStructure(ctx context.Context, tenant, kb string, buckets []StructureBucket) error { if len(buckets) == 0 { return nil } + if !datasetStructureSupported() { + return errDatasetStructureUnsupported() + } eng := w.eng if eng == nil { eng = engine.Get() @@ -229,7 +283,7 @@ func (w engineWriter) WriteMergedStructure(ctx context.Context, tenant, kb strin if b.Name == "" { continue } - ids = append(ids, datasetLevelStructureID(tenant, kb, b.Name, b.Type)) + ids = append(ids, datasetLevelStructureID(tenant, kb, b.Name, b.Type, b.CompileKwd, b.RelationType)) } if len(ids) > 0 { res, err := eng.Search(ctx, &types.SearchRequest{ @@ -263,43 +317,128 @@ func (w engineWriter) WriteMergedStructure(ctx context.Context, tenant, kb strin if b.Name == "" { continue } - bid := datasetLevelStructureID(tenant, kb, b.Name, b.Type) + bid := datasetLevelStructureID(tenant, kb, b.Name, b.Type, b.CompileKwd, b.RelationType) // Union the current batch's sources with any already-accumulated ones. if prev, ok := existing[bid]; ok { b.SourceDocIDs = appendUnique(b.SourceDocIDs, prev.SourceDocIDs) b.SourceChunkIDs = appendUnique(b.SourceChunkIDs, prev.SourceChunkIDs) } + ckwd := b.CompileKwd + if ckwd == "" { + ckwd = compileKwdStructure + } + // content_with_weight is a JSON payload, matching Python (and the existing + // wiki projection writer below): the structured fields (from/to/type for + // relations, name/type for entities) live INSIDE the payload, while + // from/to are also copied to from_entity_kwd/to_entity_kwd top-level + // columns for filtering. There is NO relation_type_kwd column — the + // relation type is carried only in the payload, exactly as Python does. + var payloadMap map[string]any + if b.FromEntity != "" || b.ToEntity != "" { + relType := b.RelationType + if relType == "" { + relType = "related" + } + payloadMap = map[string]any{ + "from": b.FromEntity, + "to": b.ToEntity, + "type": relType, + "description": desc, + } + } else { + typ := b.Type + if typ == "" { + typ = "other" + } + payloadMap = map[string]any{ + "name": b.Name, + "type": typ, + "description": desc, + } + } + payloadBytes, err := json.Marshal(payloadMap) + if err != nil { + return err + } + payload := string(payloadBytes) row := map[string]interface{}{ - // Stable dataset-level id keyed on the (name, type) or (from, to) bucket. + // Stable dataset-level id keyed on the (name, type) or (from, to) + // bucket plus the raw compile kind. "id": bid, "doc_id": kb, "tenant_id": tenant, "kb_id": kb, "available_int": 1, - "compile_kwd": compileKwdStructure, + "compile_kwd": ckwd, "scope_kwd": "dataset", - "content_with_weight": desc, - "kc_payload": desc, + "content_with_weight": payload, + "kc_payload": payload, "source_doc_ids": b.SourceDocIDs, "source_chunk_ids": b.SourceChunkIDs, "create_time": now.Format("2006-01-02 15:04:05"), "create_timestamp_flt": float64(now.Unix()), } + // Stamp the authoritative template identity so read/delete paths can match + // the dataset-structure kind by compilation_template_kind_kwd (mirror Python + // get_dataset_structure._discover_scope_templates), independent of the + // autotype-valued compile_kwd above. + if b.TemplateKind != "" { + row["compilation_template_kind_kwd"] = b.TemplateKind + } + if b.TemplateID != "" { + row["compilation_template_ids"] = []string{b.TemplateID} + } if b.FromEntity != "" || b.ToEntity != "" { // relation row: carries from/to entities; kind=relation, no name_kwd. + row["knowledge_graph_kwd"] = "relation" row["type_kwd"] = "relation" row["from_entity_kwd"] = b.FromEntity row["to_entity_kwd"] = b.ToEntity } else { + row["knowledge_graph_kwd"] = "entity" row["type_kwd"] = "entity" row["name_kwd"] = b.Name row["entity_type_kwd"] = b.Type + if b.MentionCount > 0 { + row["mention_count_int"] = b.MentionCount + } } if len(b.Vector) > 0 { row["q_"+fmt.Sprintf("%d", len(b.Vector))+"_vec"] = f32ToF64Slice(b.Vector) } rows = append(rows, row) } + // Write one kg_build_meta build-marker row per raw compile kind (the bucket + // model is keyed by raw kind; template-id granularity is a follow-up when the + // bucket key gains a template dimension). The marker is available_int=0 and + // carries create_timestamp_flt as its timestamp (build_timestamp_flt is a + // Python _META_ROW_KWD legacy field NOT present in the Infinity/ES/OS schemas, + // so writing it would fail with an undefined-column error — see review). It is + // a write/delete-side marker — GET discovery does NOT read it (it scans + // knowledge_graph_kwd=["entity"] rows instead, per §6). + seenKwd := map[string]bool{} + for _, b := range buckets { + ckwd := b.CompileKwd + if ckwd == "" { + ckwd = compileKwdStructure + } + if seenKwd[ckwd] { + continue + } + seenKwd[ckwd] = true + rows = append(rows, map[string]interface{}{ + "id": "dataset_build_meta_" + hashStr(tenant+"\x00"+kb+"\x00"+ckwd), + "doc_id": kb, + "tenant_id": tenant, + "kb_id": kb, + "available_int": 0, + "compile_kwd": ckwd, + "scope_kwd": "dataset", + "knowledge_graph_kwd": "kg_build_meta", + "create_time": now.Format("2006-01-02 15:04:05"), + "create_timestamp_flt": float64(now.Unix()), + }) + } if len(rows) == 0 { return nil } @@ -329,6 +468,9 @@ func (w engineWriter) DeleteStructureForDocs(ctx context.Context, tenant, kb str if len(deletedDocIDs) == 0 { return nil } + if !datasetStructureSupported() { + return errDatasetStructureUnsupported() + } eng := w.eng if eng == nil { eng = engine.Get() @@ -346,14 +488,25 @@ func (w engineWriter) DeleteStructureForDocs(ctx context.Context, tenant, kb str // surviving silently (review Minor). const pageSize = 500 var ghostIDs []string + // Raw compile kinds whose entity/relation rows were declared ghosts; used to + // drop a now-empty kind's kg_build_meta build marker after cleanup. + affectedKinds := map[string]bool{} for offset := 0; ; offset += pageSize { res, err := eng.Search(ctx, &types.SearchRequest{ IndexNames: []string{baseName}, KbIDs: []string{kb}, - SelectFields: []string{"id", "source_doc_ids"}, - Filter: map[string]interface{}{"kb_id": kb, "scope_kwd": "dataset", "compile_kwd": compileKwdStructure}, - Offset: offset, - Limit: pageSize, + SelectFields: []string{"id", "source_doc_ids", "compile_kwd"}, + // Match dataset-scope entity/relation rows across ALL structure kinds + // (timeline/graph/session_graph/mindmap), which now each stamp their + // raw compile_kwd; knowledge_graph_kwd ∈ {entity,relation} + scope_kwd + // =dataset is the kind-agnostic predicate (plan §1, §4.2). + Filter: map[string]interface{}{ + "kb_id": kb, + "scope_kwd": "dataset", + "knowledge_graph_kwd": []string{"entity", "relation"}, + }, + Offset: offset, + Limit: pageSize, }) if err != nil { return fmt.Errorf("structure ghost scan: %w", err) @@ -379,6 +532,9 @@ func (w engineWriter) DeleteStructureForDocs(ctx context.Context, tenant, kb str } if allGone { ghostIDs = append(ghostIDs, id) + if ckwd, _ := c["compile_kwd"].(string); ckwd != "" { + affectedKinds[ckwd] = true + } } } if len(res.Chunks) < pageSize { @@ -388,18 +544,51 @@ func (w engineWriter) DeleteStructureForDocs(ctx context.Context, tenant, kb str if len(ghostIDs) == 0 { return nil } - _, err := eng.DeleteChunks(ctx, map[string]interface{}{"id": ghostIDs, "kb_id": kb}, baseName, kb) - if err != nil { + if _, err := eng.DeleteChunks(ctx, map[string]interface{}{"id": ghostIDs, "kb_id": kb}, baseName, kb); err != nil { return fmt.Errorf("structure ghost cleanup: %w", err) } + // Drop the kg_build_meta build marker for any raw kind that no longer has an + // entity/relation row (plan §3.1.1 step 4). + for ckwd := range affectedKinds { + res, err := eng.Search(ctx, &types.SearchRequest{ + IndexNames: []string{baseName}, + KbIDs: []string{kb}, + SelectFields: []string{"id"}, + Filter: map[string]interface{}{ + "kb_id": kb, + "scope_kwd": "dataset", + "compile_kwd": ckwd, + "knowledge_graph_kwd": []string{"entity", "relation"}, + }, + Limit: 1, + }) + if err != nil { + return fmt.Errorf("structure meta check (kind %s): %w", ckwd, err) + } + if len(res.Chunks) > 0 { + continue // still has entity/relation rows; keep the marker + } + if _, err := eng.DeleteChunks(ctx, map[string]interface{}{ + "kb_id": kb, + "scope_kwd": "dataset", + "compile_kwd": ckwd, + "knowledge_graph_kwd": "kg_build_meta", + }, baseName, kb); err != nil { + return fmt.Errorf("structure meta cleanup (kind %s): %w", ckwd, err) + } + } return nil } // datasetLevelStructureID builds the stable dataset-level id for a structure -// bucket, keyed on (name, type). It must be deterministic so an incremental -// merge read-modify-writes the same row (and a rebuild clean removes it). -func datasetLevelStructureID(tenant, kb, name, typ string) string { - return "dataset_structure_" + hashStr(tenant+"\x00"+kb+"\x00"+strings.ToLower(name)+"\x00"+typ) +// bucket, keyed on (name, type, compile kind, relation type). It must be +// deterministic so an incremental merge read-modify-writes the same row (and a +// rebuild clean removes it). Including the raw compile kind keeps +// timeline/graph/mindmap buckets from colliding in the same dataset namespace; +// including the relation type keeps two relation types between the same endpoints +// (e.g. "causes" vs "contradicts") from colliding into one id (review fix). +func datasetLevelStructureID(tenant, kb, name, typ, compileKwd, relationType string) string { + return "dataset_structure_" + hashStr(tenant+"\x00"+kb+"\x00"+strings.ToLower(name)+"\x00"+typ+"\x00"+compileKwd+"\x00"+strings.ToLower(relationType)) } // f32ToF64Slice converts a float32 vector to float64 for the engine's dense @@ -937,7 +1126,13 @@ func (w engineWriter) DeleteMergedForVariant(ctx context.Context, tenant, kb str // Full-set clean (B1b): everything the consumer manages. Structure // dataset rows are tagged scope_kwd="dataset" (not a fixed compile_kwd), // so the full clean deletes by kb_id + compile_kwd IN (the fixed-kwd - // buckets) OR scope_kwd="dataset". + // buckets) OR scope_kwd="dataset". The scope_kwd="dataset" sweep requires + // the dataset-structure filter keys, so it is guarded like the per-variant + // structure branch (review fix: this is the highest-impact destructive + // path and must fail loudly on unsupported engines). + if !datasetStructureSupported() { + return errDatasetStructureUnsupported() + } _, err := eng.DeleteChunks(ctx, map[string]interface{}{ "kb_id": kb, "compile_kwd": []string{ @@ -979,17 +1174,23 @@ func (w engineWriter) DeleteMergedForVariant(ctx context.Context, tenant, kb str }, baseName, kb); err != nil { return fmt.Errorf("delete merged (wiki): %w", err) } - case kccommon.VariantStructure: - // structure dataset rows are scope_kwd="dataset" + compile_kwd= - // "structure". The compile_kwd is required: wiki merged rows ALSO carry - // scope_kwd="dataset" (W5), so a scope-only sweep would wrongly delete - // wiki merged rows too (review Major). + case kccommon.VariantStructure, kccommon.VariantMindmap: + if !datasetStructureSupported() { + return errDatasetStructureUnsupported() + } + // structure/mindmap dataset rows are scope_kwd="dataset" + + // knowledge_graph_kwd ∈ {entity,relation,kg_build_meta} with a raw + // compile_kwd (timeline/graph/session_graph/mindmap). knowledge_graph_kwd + // is required: wiki merged rows ALSO carry scope_kwd="dataset" (W5), so + // a scope-only sweep would wrongly delete wiki merged rows too (review + // Major). kg_build_meta (the build marker) is deleted together with the + // entity/relation rows. if _, err := eng.DeleteChunks(ctx, map[string]interface{}{ - "kb_id": kb, - "scope_kwd": "dataset", - "compile_kwd": []string{compileKwdStructure}, + "kb_id": kb, + "scope_kwd": "dataset", + "knowledge_graph_kwd": []string{"entity", "relation", "kg_build_meta"}, }, baseName, kb); err != nil { - return fmt.Errorf("delete merged (structure): %w", err) + return fmt.Errorf("delete merged (structure/mindmap): %w", err) } case kccommon.VariantTree: // tree/nav rows are compile_kwd="dataset_nav" with available_int=0; @@ -1000,8 +1201,6 @@ func (w engineWriter) DeleteMergedForVariant(ctx context.Context, tenant, kb str }, baseName, kb); err != nil { return fmt.Errorf("delete merged (nav): %w", err) } - case kccommon.VariantMindmap: - // mindmap has no dataset-level merged rows in the consumer path } } return nil diff --git a/internal/service/dataset_artifact_service.go b/internal/service/dataset_artifact_service.go index 0ef4ca3744..37ad7cd953 100644 --- a/internal/service/dataset_artifact_service.go +++ b/internal/service/dataset_artifact_service.go @@ -39,16 +39,6 @@ const ( CompileKwdSkillAll = "skill_all" CompileKwdDatasetNav = "dataset_nav" CompileKwdRaptorGraph = "raptor_graph" - - CompileKwdStructure = "structure" - CompileKwdStructureIndex = "structureIndex" - CompileKwdStructureEntity = "structureEntity" - CompileKwdStructureRelation = "structureRelation" - CompileKwdStructureCommunity = "structureCommunity" - - FieldStructureIndexType = "structure_index_type" - FieldStructureKind = "structure_kind" - FieldPageID = "page_id" ) // DatasetArtifactService reads knowledge-compilation artifacts (wiki pages, @@ -550,75 +540,6 @@ func (s *DatasetArtifactService) ClearWiki(ctx context.Context, tenantID, datase return deleted, nil } -// StructureItem is a single compiled structure entry for a dataset. -type StructureItem struct { - PageID string `json:"page_id"` - StructureKind string `json:"structure_kind"` - StructureIndexType string `json:"structure_index_type"` - Data string `json:"data"` -} - -// ListStructures returns the compiled structures of a dataset, filtered by -// optional structure_kind and structure_index_type. -func (s *DatasetArtifactService) ListStructures(ctx context.Context, tenantID, datasetID, structureKind, structureIndexType string) ([]StructureItem, int64, error) { - filter := map[string]interface{}{"compile_kwd": []string{CompileKwdStructure}} - if structureKind != "" { - filter[FieldStructureKind] = []string{structureKind} - } - if structureIndexType != "" { - filter[FieldStructureIndexType] = []string{structureIndexType} - } - chunks, total, err := s.searchCompiled(ctx, tenantID, datasetID, filter, - []string{FieldPageID, FieldStructureKind, FieldStructureIndexType, "content_with_weight"}, 0, 10000, nil) - if err != nil { - return nil, 0, err - } - items := make([]StructureItem, 0, len(chunks)) - for _, c := range chunks { - items = append(items, StructureItem{ - PageID: firstStringValue(c[FieldPageID]), - StructureKind: firstStringValue(c[FieldStructureKind]), - StructureIndexType: firstStringValue(c[FieldStructureIndexType]), - Data: firstStringValue(c["content_with_weight"]), - }) - } - return items, total, nil -} - -// DeleteStructures deletes the compiled structures of a dataset, optionally -// scoped by structure_kind and structure_index_type. -func (s *DatasetArtifactService) DeleteStructures(ctx context.Context, tenantID, datasetID, structureKind, structureIndexType string) (int, error) { - docEngine := engine.Get() - if docEngine == nil { - return 0, fmt.Errorf("document engine is not initialized") - } - filter := map[string]interface{}{"compile_kwd": []string{CompileKwdStructure}} - if structureKind != "" { - filter[FieldStructureKind] = []string{structureKind} - } - if structureIndexType != "" { - filter[FieldStructureIndexType] = []string{structureIndexType} - } - chunks, _, err := s.searchCompiled(ctx, tenantID, datasetID, filter, []string{"id"}, 0, 10000, nil) - if err != nil { - return 0, err - } - if len(chunks) == 0 { - return 0, nil - } - ids := make([]string, 0, len(chunks)) - for _, c := range chunks { - if id, ok := c["id"].(string); ok { - ids = append(ids, id) - } - } - cond := map[string]interface{}{"id": ids, "kb_id": datasetID} - if _, err := docEngine.DeleteChunks(ctx, cond, wikiIndexName(tenantID), datasetID); err != nil { - return 0, err - } - return len(ids), nil -} - // DeleteDocumentGraph deletes the structure graph of a single document. func (s *DatasetArtifactService) DeleteDocumentGraph(ctx context.Context, tenantID, datasetID, documentID string) (int, error) { docEngine := engine.Get() diff --git a/internal/service/dataset_structure_graph.go b/internal/service/dataset_structure_graph.go index 77ed8f448a..caa557f38a 100644 --- a/internal/service/dataset_structure_graph.go +++ b/internal/service/dataset_structure_graph.go @@ -10,11 +10,13 @@ package service import ( "context" "encoding/json" + "errors" "fmt" "regexp" "sort" "strings" + "gorm.io/gorm" "ragflow/internal/dao" "ragflow/internal/engine" "ragflow/internal/engine/types" @@ -468,8 +470,12 @@ func (s *DatasetArtifactService) buildBucket(ctx context.Context, tenantID, data if err != nil { return nil, nil, err } - var entities []StructureGraphNode - var relations []StructureGraphRelation + // Initialize as empty (non-nil) slices so a bucket with entities but no + // relations serializes "relations": [] instead of null — the frontend + // adapters (adaptKnowledgeGraphToForceGraph et al.) call .filter() on + // relations without a null guard (mirror Python, which always returns []). + entities := make([]StructureGraphNode, 0) + relations := make([]StructureGraphRelation, 0) for _, row := range fieldMap { if !rowHasEnabledSource(row, excludedDocIDs) { continue @@ -530,7 +536,7 @@ func (s *DatasetArtifactService) buildBucket(ctx context.Context, tenantID, data } aNameTerms = sortedUnique(aNameTerms) - var relations []StructureGraphRelation + relations := make([]StructureGraphRelation, 0) targetNamesLower := map[string]bool{} if len(aNameTerms) > 0 { cond := copyFilter(scope) @@ -613,6 +619,59 @@ func compilationTemplateKind(kind string) string { return k } +// ErrInvalidStructureKind is returned when the request kind does not resolve to a +// supported dataset-structure kind. Handlers use errors.Is to map it to +// CodeArgumentError (400) while treating every other service error as a server +// fault (500). +var ErrInvalidStructureKind = errors.New("invalid structure kind") + +// datasetStructureSupported mirrors the writer's engine guard: only Infinity and +// Elasticsearch resolve the dataset-structure filter keys (scope_kwd + +// knowledge_graph_kwd). The writer keeps a private copy in the knowledge_compile +// package; this service-side copy keeps the API delete path in agreement without +// exporting an ingestion-internal helper across package boundaries. +func datasetStructureSupported() bool { + switch engine.GetEngineType() { + case "infinity", "elasticsearch", "": + return true + default: + return false + } +} + +// errDatasetStructureUnsupported returns a visible error naming the engine. +func errDatasetStructureUnsupported() error { + return fmt.Errorf("dataset structure graph unsupported on doc engine %q", engine.GetEngineType()) +} + +// resolveDatasetStructureKind maps a user-facing dataset-structure kind to the +// stored top-level kind (mirror Python _DATASET_STRUCTURE_KIND_ALIASES). It is +// DELIBERATELY separate from compilationTemplateKind: that helper folds +// knowledge_graph→timeline, which would merge distinct dataset kinds, and it +// lacks the graph→knowledge_graph / mindmap→mind_map aliases (Python's +// dataset_api_service.py L1803-1805 documents exactly why it does not reuse the +// general normalizer). Returns "" for an invalid kind (caller maps that to +// 400 ARGUMENT_ERROR, never "return all"). +func resolveDatasetStructureKind(kind string) string { + k := strings.ToLower(strings.TrimSpace(strings.ReplaceAll(kind, "-", "_"))) + switch k { + case "graph", "knowledge_graph": + return "knowledge_graph" + case "mindmap", "mind_map": + // Align with Python _DATASET_STRUCTURE_KIND_ALIASES: resolve the + // user-facing "mindmap" to the stored compilation_template_kind_kwd value + // "mind_map" (the template kind, which is what read/delete paths match on). + return "mind_map" + case "timeline": + return "timeline" + case "session_essence": + return "session_essence" + case "session_graph": + return "session_graph" + } + return "" +} + // structureKindForBucket returns the normalized kind used for tree/page_index // hierarchy handling. func structureKindForBucket(kind string) string { @@ -720,8 +779,10 @@ func (s *DatasetArtifactService) GetDocumentGraph(ctx context.Context, in Docume return resp, nil } - // normal mode: discover buckets from per-doc graph blob rows. - metaFields := []string{"compile_kwd", "compilation_template_ids", "compilation_template_kind_kwd"} + // normal mode: discover buckets from per-doc graph blob rows. "id" is + // required for the same reason as dataset discovery (Infinity only projects + // listed fields; graphRowSearch keys by id). + metaFields := []string{"id", "compile_kwd", "compilation_template_ids", "compilation_template_kind_kwd"} metaRows, _, err := graphRowSearch(ctx, in.TenantID, in.DatasetID, metaFields, map[string]interface{}{"doc_id": []string{in.DocumentID}, "knowledge_graph_kwd": []string{"graph"}}, nil, 0, 1000, nil) if err != nil { @@ -789,6 +850,197 @@ func containsStr(list []string, v string) bool { return false } +// DatasetStructureGraphInput is the parsed request for the dataset-scope +// structure graph endpoint (GET/DELETE /datasets/:id/artifacts/structure). +// Kind is REQUIRED (mirrors Python dataset_api.py:715): missing or invalid → +// 400 ARGUMENT_ERROR. Wipe applies to DELETE: true deletes the dataset rows, +// false only cancels the task (rows are left for the next rebuild to clean). +type DatasetStructureGraphInput struct { + TenantID string + DatasetID string + Kind string + Wipe bool +} + +// DatasetStructureGraphResponse mirrors Python get_dataset_structure's +// {"kind": ..., "templates": [...]}. +type DatasetStructureGraphResponse struct { + Kind string `json:"kind"` + Templates []DocumentStructureGraphTemplate `json:"templates"` +} + +// GetDatasetStructure returns the dataset-scope structure graph for a resolved +// kind, mirroring Python get_dataset_structure (dataset_api_service.py). Discovery +// scans knowledge_graph_kwd=["entity"] dataset rows (scope_kwd="dataset") and +// matches the resolved kind against the stamped compilation_template_kind_kwd — +// NOT compile_kwd (compile_kwd holds the autotype "hypergraph"/"list"/"mindmap", +// which is never the kind discriminator; Python _discover_scope_templates matches +// _resolve_dataset_structure_kind against compilation_template_kind_kwd the same +// way). It collects distinct template ids, then reads each template's dataset +// entity/relation rows via buildBucket. It does NOT read kg_build_meta (write/ +// delete-side only). +func (s *DatasetArtifactService) GetDatasetStructure(ctx context.Context, in DatasetStructureGraphInput) (*DatasetStructureGraphResponse, error) { + resolved := resolveDatasetStructureKind(in.Kind) + if resolved == "" { + return nil, fmt.Errorf("%w: %q", ErrInvalidStructureKind, in.Kind) + } + if !datasetStructureSupported() { + return nil, errDatasetStructureUnsupported() + } + resp := &DatasetStructureGraphResponse{Kind: resolved, Templates: []DocumentStructureGraphTemplate{}} + + // Discover distinct template ids whose stamped template kind resolves to the + // requested kind, scanning dataset-scope entity rows only. scope_kwd="dataset" + // is required here (unlike the legacy doc_graph fallback) because dataset rows + // are the only ones carrying compilation_template_kind_kwd we can trust for the + // dataset-scope kind match. + templateIDs := map[string]struct{}{} + // "id" must be projected: graphRowSearch keys its result map by the row id, + // and Infinity only returns fields listed in SelectFields (it does not + // synthesize id), so omitting it silently drops every row (review Major). + // The resolved kind is pushed into the filter so the engine applies the + // predicate instead of scanning all entity rows and discarding them in Go. + metaFields := []string{"id", "compilation_template_kind_kwd", "compilation_template_ids"} + for offset := 0; ; offset += 1000 { + rows, total, err := graphRowSearch(ctx, in.TenantID, in.DatasetID, metaFields, + map[string]interface{}{ + "knowledge_graph_kwd": []string{"entity"}, + "scope_kwd": []string{"dataset"}, + "compilation_template_kind_kwd": []string{resolved}, + }, nil, offset, 1000, nil) + if err != nil { + return nil, err + } + for _, row := range rows { + tkind := firstStringValue(row["compilation_template_kind_kwd"]) + if tkind == "" || resolveDatasetStructureKind(tkind) != resolved { + continue + } + tid := rowTemplateID(row) + if tid != "" { + templateIDs[tid] = struct{}{} + } + } + if int64(offset+1000) >= total || len(rows) == 0 { + break + } + } + + // Read each template's dataset entity/relation rows. + for tid := range templateIDs { + scope := map[string]interface{}{ + "scope_kwd": []string{"dataset"}, + "compilation_template_ids": []string{tid}, + "compilation_template_kind_kwd": []string{resolved}, + } + entities, relations, err := s.buildBucket(ctx, in.TenantID, in.DatasetID, scope, nil) + if err != nil { + return nil, err + } + if len(entities) == 0 && len(relations) == 0 { + continue + } + resp.Templates = append(resp.Templates, DocumentStructureGraphTemplate{ + TemplateID: tid, + TemplateName: tid, + Kind: resolved, + Entities: entities, + Relations: relations, + }) + } + return resp, nil +} + +// DeleteDatasetStructure handles DELETE /datasets/:id/artifacts/structure?kind=&wipe=. +// It validates kind like GET. wipe=false cancels the kind's task (via the task-id +// field) without deleting rows; wipe=true deletes the kind's kg_build_meta marker +// + dataset entity/relation rows (document-scope rows are never touched). +func (s *DatasetArtifactService) DeleteDatasetStructure(ctx context.Context, in DatasetStructureGraphInput) (int, error) { + resolved := resolveDatasetStructureKind(in.Kind) + if resolved == "" { + return 0, fmt.Errorf("%w: %q", ErrInvalidStructureKind, in.Kind) + } + if !in.Wipe { + // Cancel the task without deleting rows. Task cancellation is task-id + // granular (mirrors Python delete_index REDIS set "{task_id}-cancel"); the + // actual row cleanup happens on the next rebuild. Rows are preserved. + return s.cancelDatasetStructureTask(ctx, in.TenantID, in.DatasetID, resolved) + } + if !datasetStructureSupported() { + return 0, errDatasetStructureUnsupported() + } + docEngine := engine.Get() + if docEngine == nil { + return 0, fmt.Errorf("document engine is not initialized") + } + indexName := fmt.Sprintf("ragflow_%s", in.TenantID) + cond := map[string]interface{}{ + "kb_id": in.DatasetID, + "scope_kwd": "dataset", + "compilation_template_kind_kwd": resolved, + "knowledge_graph_kwd": []string{"entity", "relation", "kg_build_meta"}, + } + n, err := docEngine.DeleteChunks(ctx, cond, indexName, in.DatasetID) + if err != nil { + return 0, err + } + return int(n), nil +} + +// cancelDatasetStructureTask cancels the running dataset-structure task for a +// resolved kind by clearing both its per-index task-id and finish-at fields. +// This mirrors Python delete_index (dataset_api_service.py), which clears +// {task_id_field: "", task_finish_at_field: None} so the task state is fully +// reset (a stale finish-at would otherwise leave the kind looking "done" after +// cancel). The row cleanup itself is driven by the next rebuild; Go executes the +// merge synchronously inside the ingestor (no independent task/marker), so there +// is no Redis "{task_id}-cancel" marker to publish here. +func (s *DatasetArtifactService) cancelDatasetStructureTask(ctx context.Context, tenantID, datasetID, resolvedKind string) (int, error) { + field := datasetStructureTaskIDField(resolvedKind) + if field == "" { + return 0, nil + } + updates := map[string]interface{}{ + field: "", + // gorm.Updates ignores nil values, so use an explicit NULL expression to + // actually clear the finish-at timestamp (mirrors Python None). + datasetStructureTaskFinishAtField(field): gorm.Expr("NULL"), + } + if err := dao.NewKnowledgebaseDAO().UpdateByID(ctx, dao.DB, datasetID, updates); err != nil { + return 0, err + } + return 0, nil +} + +// datasetStructureTaskFinishAtField maps a task-id field to its sibling finish-at +// field, mirroring Python f"{task_id_field.replace('_task_id', '_task_finish_at')}". +func datasetStructureTaskFinishAtField(taskIDField string) string { + return strings.Replace(taskIDField, "_task_id", "_task_finish_at", 1) +} + +// datasetStructureTaskIDField maps a resolved dataset-structure kind to its kb +// task-id field name. It mirrors Python _INDEX_TYPE_TO_TASK_ID_FIELD: each +// dataset-merge kind carries its own "_task_id" field (structure_graph, +// structure_mindmap, timeline, session_graph, session_essence), NOT the legacy +// doc-level graphrag_task_id/mindmap_task_id. Empty means "no task-id field". +func datasetStructureTaskIDField(resolvedKind string) string { + switch resolvedKind { + case "knowledge_graph": + // "graph" is already normalized to "knowledge_graph" by + // resolveDatasetStructureKind, so no separate "graph" case is needed. + return "structure_graph_task_id" + case "mindmap", "mind_map": + return "structure_mindmap_task_id" + case "timeline": + return "timeline_task_id" + case "session_graph": + return "session_graph_task_id" + case "session_essence": + return "session_essence_task_id" + } + return "" +} + // resolveGraphBucket mirrors Python _resolve_bucket. func resolveGraphBucket(row map[string]interface{}, templateMeta map[string]map[string]interface{}, documentID string) (map[string]interface{}, map[string]interface{}) { compileKwd := firstStringValue(row["compile_kwd"]) @@ -866,7 +1118,7 @@ func rowTemplateID(row map[string]interface{}) string { } func (s *DatasetArtifactService) appendRaptorBlob(ctx context.Context, tenantID, datasetID, documentID string, grouped map[string]DocumentStructureGraphTemplate) { - rows, _, err := graphRowSearch(ctx, tenantID, datasetID, []string{"content_with_weight", "compile_kwd"}, + rows, _, err := graphRowSearch(ctx, tenantID, datasetID, []string{"id", "content_with_weight", "compile_kwd"}, map[string]interface{}{"doc_id": []string{documentID}, "compile_kwd": []string{"raptor_graph"}}, nil, 0, 16, nil) if err != nil { return diff --git a/internal/service/ingestion_task_service_test.go b/internal/service/ingestion_task_service_test.go index 53c8d1f217..e8a05d4ec6 100644 --- a/internal/service/ingestion_task_service_test.go +++ b/internal/service/ingestion_task_service_test.go @@ -875,6 +875,45 @@ func TestIngestionTaskServiceMarkStoppedIdempotentOnAlreadyStopped(t *testing.T) } } +// TestIngestionTaskServiceRequestStopNeverReopensTerminalTask locks the invariant +// that a terminal task (COMPLETED/STOPPED/FAILED) can never be moved back to +// STOPPING. A terminal task whose message was already acked must not regress to +// the in-flight STOPPING state (which has no settled worker and would be stuck +// forever). RequestStop is a no-op for terminal states. +func TestIngestionTaskServiceRequestStopNeverReopensTerminalTask(t *testing.T) { + for _, status := range []string{common.COMPLETED, common.STOPPED, common.FAILED} { + t.Run(status, func(t *testing.T) { + db := setupServiceTestDB(t) + pushServiceDB(t, db) + insertTestIngestionTaskWithStatus(t, "task-1", "user-1", "doc-1", "kb-1", status) + ctx := t.Context() + + svc := NewIngestionTaskService() + task, err := svc.RequestStop(ctx, "task-1") + if err != nil { + t.Fatalf("RequestStop on %s task should be a no-op, got: %v", status, err) + } + if task.Status != status { + t.Fatalf("RequestStop moved %s task to %q, must stay %s", status, task.Status, status) + } + }) + } +} + +// TestValidateTransitionRejectsTerminalToStopping locks the state-machine +// invariant directly: STOPPING is only reachable from RUNNING, never from a +// terminal state (COMPLETED/STOPPED/FAILED) or from STOPPING itself. +func TestValidateTransitionRejectsTerminalToStopping(t *testing.T) { + for _, from := range []string{common.COMPLETED, common.STOPPED, common.FAILED, common.STOPPING, common.CREATED} { + if err := validateTransition(from, common.STOPPING); err == nil { + t.Errorf("validateTransition(%s -> STOPPING) = nil, want error", from) + } + } + if err := validateTransition(common.RUNNING, common.STOPPING); err != nil { + t.Errorf("validateTransition(RUNNING -> STOPPING) = %v, want nil", err) + } +} + func TestIngestionTaskServiceMarkFailedIdempotentOnAlreadyTerminal(t *testing.T) { db := setupServiceTestDB(t) pushServiceDB(t, db) diff --git a/internal/service/nav_embedder.go b/internal/service/nav_embedder.go index 6598695291..891e79bff0 100644 --- a/internal/service/nav_embedder.go +++ b/internal/service/nav_embedder.go @@ -20,6 +20,8 @@ import ( "context" "fmt" "strings" + + "ragflow/internal/entity" ) // NavEmbedder is the production implementation of nlp.NavEmbedder. It resolves @@ -45,7 +47,17 @@ func (e *NavEmbedder) Encode(ctx context.Context, tenantID string, texts []strin } name := e.embdModelName if name == "" { - name = tenantID // composite name falls back to tenant default resolution + // Resolve the tenant's default embedding model composite reference + // ("@@") — NOT the tenant id. Passing the + // tenant id as the model ref makes ResolveModelConfig parse it as a + // "model@provider" key, which fails with "provider name missing in model + // name: ". Mirrors knowledge_compiler_wiring.go's chat-model + // default resolution. + ref, err := e.modelSvc.GetTenantDefaultModelRef(ctx, tenantID, entity.ModelTypeEmbedding) + if err != nil { + return nil, fmt.Errorf("datasetnav: resolve embedding model for tenant %s: %w", tenantID, err) + } + name = ref } model, err := e.modelSvc.GetEmbeddingModel(ctx, tenantID, name) if err != nil { diff --git a/web/src/components/structure-graph/adapters.ts b/web/src/components/structure-graph/adapters.ts index 0d19a2a75a..e64c8e6029 100644 --- a/web/src/components/structure-graph/adapters.ts +++ b/web/src/components/structure-graph/adapters.ts @@ -53,7 +53,7 @@ function buildTreeDataItems( relationTypes: string[], showEntityType = false, ): TreeDataItem[] { - const normalized = entities + const normalized = (entities ?? []) .map(normalizeEntity) .filter((entity) => entity.id); const map = new Map( @@ -69,7 +69,7 @@ function buildTreeDataItems( ); const childIds = new Set(); - for (const relation of relations) { + for (const relation of relations ?? []) { if (!relationTypes.includes(relation.type ?? '')) continue; // Self-referencing relation (same entity as its own child) creates // an infinite recursion in the tree renderer. This is a backend @@ -99,7 +99,7 @@ function buildUniqueTreeDataItems( ): TreeDataItem[] { const normalized = [ ...new Map( - entities + (entities ?? []) .map(normalizeEntity) .filter((entity) => entity.id) .map((entity) => [entity.id, entity]), @@ -119,7 +119,7 @@ function buildUniqueTreeDataItems( const childIds = new Set(); const parentMap = new Map(); - for (const relation of relations) { + for (const relation of relations ?? []) { if (!relationTypes.includes(relation.type ?? '')) continue; const parent = map.get(relation.from); @@ -172,7 +172,7 @@ export function adaptTreeToTreeData( export function adaptKnowledgeGraphToForceGraph( template: IStructureGraphTemplate, ): IArtifactGraph { - const entities: IArtifactGraphEntity[] = template.entities.map((entity) => { + const entities: IArtifactGraphEntity[] = (template.entities ?? []).map((entity) => { const normalized = normalizeEntity(entity); return { slug: normalized.id, @@ -189,7 +189,7 @@ export function adaptKnowledgeGraphToForceGraph( return { entities, - relations: template.relations + relations: (template.relations ?? []) .filter( (relation) => // Only keep relations whose source and target entities both exist in the graph. @@ -284,7 +284,7 @@ export function adaptTimelineToX6Data(template: IStructureGraphTemplate): { }; }); - const edges = template.relations + const edges = (template.relations ?? []) .filter( (relation) => entityIds.has(relation.from) && entityIds.has(relation.to), )