package activity import ( "context" "crypto/sha256" "encoding/hex" "fmt" "log/slog" "regexp" "strings" "github.com/rockliang/poimen/workflows/internal/memory" "github.com/rockliang/poimen/workflows/pkg/types" ) // SynthesisActivities holds dependencies for synthesis pipeline activities. type SynthesisActivities struct { memClient *memory.Client } // NewSynthesisActivities creates synthesis activities with a memory service client. func NewSynthesisActivities(memClient *memory.Client) *SynthesisActivities { return &SynthesisActivities{memClient: memClient} } // Re-export shared types from pkg/types type SynthesisInput = types.SynthesisInput type ExtractedEntity = types.ExtractedEntity type ExtractedFact = types.ExtractedFact type ContradictionResult = types.ContradictionResult type PersistInput = types.PersistInput // ChunkAndEmbedActivity chunks text and generates a chunk ID. // Stage 1: Creates a deterministic chunk ID from content hash, // then ingests via memory service for embedding generation. func (s *SynthesisActivities) ChunkAndEmbedActivity(ctx context.Context, input SynthesisInput) (string, error) { logger := slog.Default() // Generate deterministic chunk ID from content hash := sha256.Sum256([]byte(input.Text)) chunkID := "chunk-" + hex.EncodeToString(hash[:8]) logger.Info("chunking text", "chunk_id", chunkID, "text_len", len(input.Text)) // Ingest via memory service (generates embedding) _, err := s.memClient.Ingest(ctx, &memory.IngestRequest{ Project: input.Project, Source: input.Source, Kind: input.Kind, Text: input.Text, Metadata: map[string]interface{}{ "chunk_id": chunkID, "tags": input.Tags, }, }) if err != nil { return "", fmt.Errorf("ingest chunk: %w", err) } return chunkID, nil } // ExtractEntitiesActivity extracts entities from text using pattern matching // and wiki-link detection. LLM extraction is a future enhancement. // Stage 2: Returns entities with confidence scores. func (s *SynthesisActivities) ExtractEntitiesActivity(ctx context.Context, chunkID string, text string) ([]ExtractedEntity, error) { logger := slog.Default() logger.Info("extracting entities", "chunk_id", chunkID) entities := make([]ExtractedEntity, 0) seen := make(map[string]bool) // Pattern 1: Wiki-link extraction [[EntityName]] wikiPattern := regexp.MustCompile(`\[\[([^\]]+)\]\]`) for _, match := range wikiPattern.FindAllStringSubmatch(text, -1) { name := strings.TrimSpace(match[1]) if !seen[name] { entities = append(entities, ExtractedEntity{ Name: name, EntityType: "reference", Confidence: 0.95, }) seen[name] = true } } // Pattern 2: Capitalized proper nouns (simple NER) properNounPattern := regexp.MustCompile(`\b([A-Z][a-z]+(?:\s+[A-Z][a-z]+)*)\b`) for _, match := range properNounPattern.FindAllStringSubmatch(text, -1) { name := match[1] if !seen[name] && !isCommonWord(name) && len(name) > 2 { entities = append(entities, ExtractedEntity{ Name: name, EntityType: classifyEntity(name), Confidence: 0.70, }) seen[name] = true } } // Pattern 3: Technical terms (ALL_CAPS or camelCase) techPattern := regexp.MustCompile(`\b([A-Z][A-Z_]{2,}|[a-z]+[A-Z][a-zA-Z]+)\b`) for _, match := range techPattern.FindAllStringSubmatch(text, -1) { name := match[1] if !seen[name] { entities = append(entities, ExtractedEntity{ Name: name, EntityType: "technical", Confidence: 0.65, }) seen[name] = true } } logger.Info("entities extracted", "count", len(entities)) return entities, nil } // ExtractFactsActivity extracts subject-predicate-object facts from text. // Stage 3: Pattern-based extraction with entity context. func (s *SynthesisActivities) ExtractFactsActivity(ctx context.Context, chunkID string, text string, entities []ExtractedEntity) ([]ExtractedFact, error) { logger := slog.Default() logger.Info("extracting facts", "chunk_id", chunkID, "entity_count", len(entities)) facts := make([]ExtractedFact, 0) // Build entity name set for matching entityNames := make(map[string]bool) for _, e := range entities { entityNames[strings.ToLower(e.Name)] = true } // Pattern: "X uses/runs/has Y" verbPatterns := []struct { pattern *regexp.Regexp predicate string }{ {regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+uses?\s+(.+?)(?:\.|,|$)`), "uses"}, {regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+runs?\s+(?:on\s+)?(.+?)(?:\.|,|$)`), "runs_on"}, {regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:has|have)\s+(.+?)(?:\.|,|$)`), "has"}, {regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:is|are)\s+(.+?)(?:\.|,|$)`), "is"}, {regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:depends?\s+on|requires?)\s+(.+?)(?:\.|,|$)`), "depends_on"}, {regexp.MustCompile(`(?i)(\w+(?:\s+\w+)?)\s+(?:connects?\s+to|talks?\s+to)\s+(.+?)(?:\.|,|$)`), "connects_to"}, } for _, vp := range verbPatterns { for _, match := range vp.pattern.FindAllStringSubmatch(text, -1) { subject := strings.TrimSpace(match[1]) object := strings.TrimSpace(match[2]) // Validation: skip empty or invalid extracts if len(subject) == 0 || len(object) == 0 { continue // Skip empty subject/object } // Truncate overly long objects (avoid capturing entire sentence) if len(object) > 500 { logger.Info("truncating long object", "original_len", len(object), "subject", subject, "predicate", vp.predicate) object = object[:500] } // Truncate overly long subjects if len(subject) > 200 { logger.Info("truncating long subject", "original_len", len(subject), "predicate", vp.predicate) subject = subject[:200] } // Boost confidence if subject/object are known entities confidence := 0.60 if entityNames[strings.ToLower(subject)] { confidence += 0.15 } if entityNames[strings.ToLower(object)] { confidence += 0.15 } facts = append(facts, ExtractedFact{ Subject: subject, Predicate: vp.predicate, Object: object, Confidence: confidence, }) } } logger.Info("facts extracted", "count", len(facts)) return facts, nil } // DetectContradictionsActivity detects contradictions between new facts // and existing knowledge. Uses pre-filter to avoid unnecessary comparisons. // Stage 4: Returns contradictions with severity and review status. func (s *SynthesisActivities) DetectContradictionsActivity(ctx context.Context, project string, facts []ExtractedFact) ([]ContradictionResult, error) { logger := slog.Default() logger.Info("detecting contradictions", "project", project, "fact_count", len(facts)) contradictions := make([]ContradictionResult, 0) for _, fact := range facts { // Query existing facts about the same subject query := fmt.Sprintf("%s %s", fact.Subject, fact.Predicate) results, err := s.memClient.Query(ctx, &memory.QueryRequest{ Project: project, Query: query, LevelFilter: []string{"L1", "L2"}, Floor: 0.7, Limit: 5, }) if err != nil { logger.Warn("query for contradictions failed", "error", err, "subject", fact.Subject) continue // Non-fatal: skip this fact } for _, r := range results.Results { // Pre-filter: check if result mentions same subject + different object if containsSubject(r.Text, fact.Subject) && contradicts(r.Text, fact) { severity := "low" if r.Score > 0.9 { severity = "high" } else if r.Score > 0.8 { severity = "medium" } autoResolved := severity == "low" contradictions = append(contradictions, ContradictionResult{ FactA: ExtractedFact{ Subject: fact.Subject, Predicate: fact.Predicate, Object: r.Text, }, FactB: fact, Severity: severity, AutoResolved: autoResolved, QueuedReview: !autoResolved, }) } } } logger.Info("contradictions detected", "count", len(contradictions)) return contradictions, nil } // PersistSynthesisActivity saves all synthesis results to the memory service. // Stage 5: Persists entities, facts, and queues contradictions for review. // Returns error if any persistence fails (fail-safe semantics). func (s *SynthesisActivities) PersistSynthesisActivity(ctx context.Context, input PersistInput) error { logger := slog.Default() logger.Info("persisting synthesis results", "chunk_id", input.ChunkID, "entities", len(input.Entities), "facts", len(input.Facts), "contradictions", len(input.Contradictions), ) var errs []error // Persist entities as knowledge records for _, entity := range input.Entities { _, err := s.memClient.Ingest(ctx, &memory.IngestRequest{ Project: input.Project, Source: input.Source, Kind: "L1", Text: fmt.Sprintf("Entity: %s (type: %s, confidence: %.2f)", entity.Name, entity.EntityType, entity.Confidence), Metadata: map[string]interface{}{ "chunk_id": input.ChunkID, "entity_type": entity.EntityType, "entity_name": entity.Name, }, }) if err != nil { logger.Error("failed to persist entity", "entity", entity.Name, "error", err) errs = append(errs, fmt.Errorf("persist entity %s: %w", entity.Name, err)) } } // Persist facts for _, fact := range input.Facts { _, err := s.memClient.Ingest(ctx, &memory.IngestRequest{ Project: input.Project, Source: input.Source, Kind: "L1", Text: fmt.Sprintf("%s %s %s", fact.Subject, fact.Predicate, fact.Object), Metadata: map[string]interface{}{ "chunk_id": input.ChunkID, "subject": fact.Subject, "predicate": fact.Predicate, "object": fact.Object, }, }) if err != nil { logger.Error("failed to persist fact", "subject", fact.Subject, "predicate", fact.Predicate, "error", err) errs = append(errs, fmt.Errorf("persist fact %s %s: %w", fact.Subject, fact.Predicate, err)) } } // Return all accumulated errors (fail-safe semantics) if len(errs) > 0 { logger.Error("persistence failed with errors", "error_count", len(errs), "chunk_id", input.ChunkID) return fmt.Errorf("persist synthesis: %d errors - %v", len(errs), errs) } logger.Info("synthesis persisted successfully", "chunk_id", input.ChunkID) return nil } // --- helpers --- func isCommonWord(word string) bool { common := map[string]bool{ "The": true, "This": true, "That": true, "These": true, "There": true, "When": true, "Where": true, "What": true, "Which": true, "How": true, "But": true, "And": true, "For": true, "Not": true, "You": true, "All": true, "Can": true, "Her": true, "Was": true, "One": true, "Our": true, "Out": true, "Are": true, "Has": true, "Its": true, "May": true, "New": true, "Now": true, "Old": true, "See": true, "Way": true, "Who": true, } return common[word] } func classifyEntity(name string) string { toolPatterns := []string{"Kubernetes", "Docker", "Nginx", "Redis", "Postgres", "ArgoCD", "Terraform", "Helm"} for _, t := range toolPatterns { if strings.EqualFold(name, t) { return "tool" } } return "concept" } func containsSubject(text, subject string) bool { return strings.Contains(strings.ToLower(text), strings.ToLower(subject)) } func contradicts(existingText string, newFact ExtractedFact) bool { // Simple heuristic: if existing text mentions subject with a different value // for the same predicate pattern, it might contradict lower := strings.ToLower(existingText) subjectLower := strings.ToLower(newFact.Subject) objectLower := strings.ToLower(newFact.Object) // If text mentions subject but NOT the same object, potential contradiction return strings.Contains(lower, subjectLower) && !strings.Contains(lower, objectLower) }