Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6ef6663965 | ||
|
|
5ed16f8d9b | ||
|
|
5342de37cc | ||
|
|
7be93d1d16 | ||
|
|
5ccc3711e6 | ||
|
|
9020ccf839 |
@@ -0,0 +1,346 @@
|
|||||||
|
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)
|
||||||
|
}
|
||||||
@@ -0,0 +1,183 @@
|
|||||||
|
package activity
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
var testCtx = context.Background()
|
||||||
|
|
||||||
|
func TestExtractEntities_WikiLinks(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil) // No client needed for extraction
|
||||||
|
entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-1", "Deploy [[Kubernetes]] with [[ArgoCD]]")
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
names := entityNames(entities)
|
||||||
|
assert.Contains(t, names, "Kubernetes")
|
||||||
|
assert.Contains(t, names, "ArgoCD")
|
||||||
|
|
||||||
|
// Wiki links get high confidence
|
||||||
|
for _, e := range entities {
|
||||||
|
if e.Name == "Kubernetes" || e.Name == "ArgoCD" {
|
||||||
|
assert.Equal(t, 0.95, e.Confidence)
|
||||||
|
assert.Equal(t, "reference", e.EntityType)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractEntities_ProperNouns(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil)
|
||||||
|
entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-2", "Redis runs on Ubuntu Server")
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
names := entityNames(entities)
|
||||||
|
assert.Contains(t, names, "Redis")
|
||||||
|
assert.Contains(t, names, "Ubuntu Server")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractEntities_TechnicalTerms(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil)
|
||||||
|
entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-3", "Set MAX_RETRIES and use camelCase variables")
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
names := entityNames(entities)
|
||||||
|
assert.Contains(t, names, "MAX_RETRIES")
|
||||||
|
assert.Contains(t, names, "camelCase")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractEntities_Deduplication(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil)
|
||||||
|
entities, err := sa.ExtractEntitiesActivity(testCtx, "chunk-4", "[[Redis]] uses Redis for caching")
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
count := 0
|
||||||
|
for _, e := range entities {
|
||||||
|
if e.Name == "Redis" {
|
||||||
|
count++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert.Equal(t, 1, count, "Redis should appear only once")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractFacts_VerbPatterns(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil)
|
||||||
|
entities := []ExtractedEntity{
|
||||||
|
{Name: "Kubernetes", EntityType: "tool"},
|
||||||
|
{Name: "Docker", EntityType: "tool"},
|
||||||
|
}
|
||||||
|
facts, err := sa.ExtractFactsActivity(testCtx, "chunk-5",
|
||||||
|
"Kubernetes uses Docker for container runtime. Redis depends on TCP",
|
||||||
|
entities)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
assert.Greater(t, len(facts), 0)
|
||||||
|
|
||||||
|
// Find the "uses" fact
|
||||||
|
found := false
|
||||||
|
for _, f := range facts {
|
||||||
|
if f.Predicate == "uses" && f.Subject == "Kubernetes" {
|
||||||
|
found = true
|
||||||
|
assert.Greater(t, f.Confidence, 0.7) // Boosted by known entities
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert.True(t, found, "should find Kubernetes uses Docker fact")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractFacts_EmptyText(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil)
|
||||||
|
facts, err := sa.ExtractFactsActivity(testCtx, "chunk-6", "", nil)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
assert.Empty(t, facts)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestContradicts(t *testing.T) {
|
||||||
|
assert.True(t, contradicts("Kubernetes uses port 8080", ExtractedFact{
|
||||||
|
Subject: "Kubernetes", Predicate: "uses_port", Object: "9090",
|
||||||
|
}))
|
||||||
|
|
||||||
|
assert.False(t, contradicts("Kubernetes uses port 8080", ExtractedFact{
|
||||||
|
Subject: "Kubernetes", Predicate: "uses_port", Object: "8080",
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestContainsSubject(t *testing.T) {
|
||||||
|
assert.True(t, containsSubject("Kubernetes runs on Linux", "kubernetes"))
|
||||||
|
assert.False(t, containsSubject("Docker runs on Linux", "kubernetes"))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestIsCommonWord(t *testing.T) {
|
||||||
|
assert.True(t, isCommonWord("The"))
|
||||||
|
assert.True(t, isCommonWord("This"))
|
||||||
|
assert.False(t, isCommonWord("Kubernetes"))
|
||||||
|
assert.False(t, isCommonWord("Redis"))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestClassifyEntity(t *testing.T) {
|
||||||
|
assert.Equal(t, "tool", classifyEntity("Kubernetes"))
|
||||||
|
assert.Equal(t, "tool", classifyEntity("Docker"))
|
||||||
|
assert.Equal(t, "tool", classifyEntity("Redis"))
|
||||||
|
assert.Equal(t, "concept", classifyEntity("SomeRandomThing"))
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- Tests for ExtractFactsActivity Validation ---
|
||||||
|
|
||||||
|
func TestExtractFacts_WithValidation(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil)
|
||||||
|
// Test with very long object that should be truncated
|
||||||
|
longText := "Kubernetes uses " + strings.Repeat("very long object name that should be truncated ", 20)
|
||||||
|
entities := []ExtractedEntity{}
|
||||||
|
|
||||||
|
facts, err := sa.ExtractFactsActivity(testCtx, "chunk-1", longText, entities)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
// Verify no fact has object > 500 chars
|
||||||
|
for _, f := range facts {
|
||||||
|
assert.LessOrEqual(t, len(f.Object), 500, "object should be truncated to 500 chars")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractFacts_SkipsEmpty(t *testing.T) {
|
||||||
|
sa := NewSynthesisActivities(nil)
|
||||||
|
// Text with empty patterns that would extract nothing
|
||||||
|
text := "Something uses and other things"
|
||||||
|
entities := []ExtractedEntity{}
|
||||||
|
|
||||||
|
facts, err := sa.ExtractFactsActivity(testCtx, "chunk-1", text, entities)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
// Verify no empty facts
|
||||||
|
for _, f := range facts {
|
||||||
|
assert.NotEmpty(t, f.Subject, "subject should not be empty")
|
||||||
|
assert.NotEmpty(t, f.Object, "object should not be empty")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPersistSynthesis_EmptyInput(t *testing.T) {
|
||||||
|
// Test that empty input is handled (no entities or facts to persist)
|
||||||
|
// Note: This test requires a mock memory service; for now we just test structure
|
||||||
|
input := PersistInput{
|
||||||
|
ChunkID: "chunk-123",
|
||||||
|
Project: "test",
|
||||||
|
Source: "test://1",
|
||||||
|
Kind: "L1",
|
||||||
|
Entities: []ExtractedEntity{}, // Empty
|
||||||
|
Facts: []ExtractedFact{}, // Empty
|
||||||
|
Contradictions: []ContradictionResult{},
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify input structure is valid
|
||||||
|
assert.Equal(t, "chunk-123", input.ChunkID)
|
||||||
|
assert.Equal(t, 0, len(input.Entities))
|
||||||
|
assert.Equal(t, 0, len(input.Facts))
|
||||||
|
}
|
||||||
|
|
||||||
|
// helper
|
||||||
|
func entityNames(entities []ExtractedEntity) []string {
|
||||||
|
names := make([]string, len(entities))
|
||||||
|
for i, e := range entities {
|
||||||
|
names[i] = e.Name
|
||||||
|
}
|
||||||
|
return names
|
||||||
|
}
|
||||||
@@ -0,0 +1,59 @@
|
|||||||
|
package types
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
|
// SynthesisInput contains the input for the synthesis workflow.
|
||||||
|
type SynthesisInput struct {
|
||||||
|
Project string `json:"project"`
|
||||||
|
Source string `json:"source"`
|
||||||
|
Text string `json:"text"`
|
||||||
|
Kind string `json:"kind"` // L1, L2, reference
|
||||||
|
Tags []string `json:"tags,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// SynthesisResult contains the output of the synthesis workflow.
|
||||||
|
type SynthesisResult struct {
|
||||||
|
ChunkID string `json:"chunk_id"`
|
||||||
|
EntitiesExtracted int `json:"entities_extracted"`
|
||||||
|
FactsExtracted int `json:"facts_extracted"`
|
||||||
|
Contradictions int `json:"contradictions"`
|
||||||
|
ReviewQueued int `json:"review_queued"`
|
||||||
|
Entities []ExtractedEntity `json:"entities"`
|
||||||
|
Facts []ExtractedFact `json:"facts"`
|
||||||
|
Duration time.Duration `json:"duration"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExtractedEntity represents an entity found during synthesis.
|
||||||
|
type ExtractedEntity struct {
|
||||||
|
Name string `json:"name"`
|
||||||
|
EntityType string `json:"entity_type"`
|
||||||
|
Confidence float64 `json:"confidence"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExtractedFact represents a fact extracted during synthesis.
|
||||||
|
type ExtractedFact struct {
|
||||||
|
Subject string `json:"subject"`
|
||||||
|
Predicate string `json:"predicate"`
|
||||||
|
Object string `json:"object"`
|
||||||
|
Confidence float64 `json:"confidence"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ContradictionResult represents a contradiction detection result.
|
||||||
|
type ContradictionResult struct {
|
||||||
|
FactA ExtractedFact `json:"fact_a"`
|
||||||
|
FactB ExtractedFact `json:"fact_b"`
|
||||||
|
Severity string `json:"severity"` // low, medium, high
|
||||||
|
AutoResolved bool `json:"auto_resolved"`
|
||||||
|
QueuedReview bool `json:"queued_review"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// PersistInput groups all synthesis results for persistence.
|
||||||
|
type PersistInput struct {
|
||||||
|
ChunkID string `json:"chunk_id"`
|
||||||
|
Project string `json:"project"`
|
||||||
|
Source string `json:"source"`
|
||||||
|
Kind string `json:"kind"`
|
||||||
|
Entities []ExtractedEntity `json:"entities"`
|
||||||
|
Facts []ExtractedFact `json:"facts"`
|
||||||
|
Contradictions []ContradictionResult `json:"contradictions"`
|
||||||
|
}
|
||||||
@@ -0,0 +1,125 @@
|
|||||||
|
package workflow
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"go.temporal.io/sdk/temporal"
|
||||||
|
"go.temporal.io/sdk/workflow"
|
||||||
|
"github.com/rockliang/poimen/workflows/pkg/types"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Re-export shared types from pkg/types for backward compatibility
|
||||||
|
type SynthesisInput = types.SynthesisInput
|
||||||
|
type SynthesisResult = types.SynthesisResult
|
||||||
|
type ExtractedEntity = types.ExtractedEntity
|
||||||
|
type ExtractedFact = types.ExtractedFact
|
||||||
|
type ContradictionResult = types.ContradictionResult
|
||||||
|
type PersistInput = types.PersistInput
|
||||||
|
|
||||||
|
var synthesisActivityOptions = workflow.ActivityOptions{
|
||||||
|
StartToCloseTimeout: 60 * time.Second,
|
||||||
|
RetryPolicy: &temporal.RetryPolicy{
|
||||||
|
InitialInterval: time.Second,
|
||||||
|
BackoffCoefficient: 2.0,
|
||||||
|
MaximumInterval: 30 * time.Second,
|
||||||
|
MaximumAttempts: 3,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
// SynthesisWorkflow orchestrates the 4-stage memory synthesis pipeline.
|
||||||
|
//
|
||||||
|
// Stage 1: Chunk + embed text
|
||||||
|
// Stage 2: Extract entities (LLM + reflection)
|
||||||
|
// Stage 3: Extract facts (pattern + LLM)
|
||||||
|
// Stage 4: Detect contradictions (pre-filter + LLM)
|
||||||
|
//
|
||||||
|
// Each stage is an activity with independent retry policy.
|
||||||
|
func SynthesisWorkflow(ctx workflow.Context, input SynthesisInput) (*SynthesisResult, error) {
|
||||||
|
logger := workflow.GetLogger(ctx)
|
||||||
|
startTime := workflow.Now(ctx)
|
||||||
|
|
||||||
|
logger.Info("synthesis started",
|
||||||
|
"project", input.Project,
|
||||||
|
"source", input.Source,
|
||||||
|
"kind", input.Kind,
|
||||||
|
)
|
||||||
|
|
||||||
|
actCtx := workflow.WithActivityOptions(ctx, synthesisActivityOptions)
|
||||||
|
|
||||||
|
// Stage 1: Chunk + Embed
|
||||||
|
var chunkID string
|
||||||
|
err := workflow.ExecuteActivity(actCtx, "ChunkAndEmbedActivity", input).Get(ctx, &chunkID)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("stage 1 chunk+embed: %w", err)
|
||||||
|
}
|
||||||
|
logger.Info("stage 1 complete", "chunk_id", chunkID)
|
||||||
|
|
||||||
|
// Stage 2: Entity Extraction
|
||||||
|
var entities []ExtractedEntity
|
||||||
|
err = workflow.ExecuteActivity(actCtx, "ExtractEntitiesActivity", chunkID, input.Text).Get(ctx, &entities)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("stage 2 entity extraction: %w", err)
|
||||||
|
}
|
||||||
|
logger.Info("stage 2 complete", "entities", len(entities))
|
||||||
|
|
||||||
|
// Stage 3: Fact Extraction
|
||||||
|
var facts []ExtractedFact
|
||||||
|
err = workflow.ExecuteActivity(actCtx, "ExtractFactsActivity", chunkID, input.Text, entities).Get(ctx, &facts)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("stage 3 fact extraction: %w", err)
|
||||||
|
}
|
||||||
|
logger.Info("stage 3 complete", "facts", len(facts))
|
||||||
|
|
||||||
|
// Stage 4: Contradiction Detection
|
||||||
|
var contradictions []ContradictionResult
|
||||||
|
err = workflow.ExecuteActivity(actCtx, "DetectContradictionsActivity", input.Project, facts).Get(ctx, &contradictions)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("stage 4 contradiction detection: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
reviewQueued := 0
|
||||||
|
for _, c := range contradictions {
|
||||||
|
if c.QueuedReview {
|
||||||
|
reviewQueued++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
logger.Info("stage 4 complete", "contradictions", len(contradictions), "review_queued", reviewQueued)
|
||||||
|
|
||||||
|
// Stage 5: Persist results
|
||||||
|
persistInput := PersistInput{
|
||||||
|
ChunkID: chunkID,
|
||||||
|
Project: input.Project,
|
||||||
|
Source: input.Source,
|
||||||
|
Kind: input.Kind,
|
||||||
|
Entities: entities,
|
||||||
|
Facts: facts,
|
||||||
|
Contradictions: contradictions,
|
||||||
|
}
|
||||||
|
err = workflow.ExecuteActivity(actCtx, "PersistSynthesisActivity", persistInput).Get(ctx, nil)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("stage 5 persist: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
duration := workflow.Now(ctx).Sub(startTime)
|
||||||
|
result := &SynthesisResult{
|
||||||
|
ChunkID: chunkID,
|
||||||
|
EntitiesExtracted: len(entities),
|
||||||
|
FactsExtracted: len(facts),
|
||||||
|
Contradictions: len(contradictions),
|
||||||
|
ReviewQueued: reviewQueued,
|
||||||
|
Entities: entities,
|
||||||
|
Facts: facts,
|
||||||
|
Duration: duration,
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.Info("synthesis complete",
|
||||||
|
"chunk_id", chunkID,
|
||||||
|
"entities", len(entities),
|
||||||
|
"facts", len(facts),
|
||||||
|
"contradictions", len(contradictions),
|
||||||
|
"duration", duration,
|
||||||
|
)
|
||||||
|
|
||||||
|
return result, nil
|
||||||
|
}
|
||||||
@@ -0,0 +1,179 @@
|
|||||||
|
package workflow
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/mock"
|
||||||
|
"go.temporal.io/sdk/testsuite"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Stub activity functions for test registration
|
||||||
|
func ChunkAndEmbedActivity(_ context.Context, _ SynthesisInput) (string, error) { return "", nil }
|
||||||
|
func ExtractEntitiesActivity(_ context.Context, _ string, _ string) ([]ExtractedEntity, error) { return nil, nil }
|
||||||
|
func ExtractFactsActivity(_ context.Context, _ string, _ string, _ []ExtractedEntity) ([]ExtractedFact, error) { return nil, nil }
|
||||||
|
func DetectContradictionsActivity(_ context.Context, _ string, _ []ExtractedFact) ([]ContradictionResult, error) { return nil, nil }
|
||||||
|
func PersistSynthesisActivity(_ context.Context, _ PersistInput) error { return nil }
|
||||||
|
|
||||||
|
func registerSynthesisActivities(env *testsuite.TestWorkflowEnvironment) {
|
||||||
|
env.RegisterActivity(ChunkAndEmbedActivity)
|
||||||
|
env.RegisterActivity(ExtractEntitiesActivity)
|
||||||
|
env.RegisterActivity(ExtractFactsActivity)
|
||||||
|
env.RegisterActivity(DetectContradictionsActivity)
|
||||||
|
env.RegisterActivity(PersistSynthesisActivity)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSynthesisWorkflow_Success(t *testing.T) {
|
||||||
|
ts := &testsuite.WorkflowTestSuite{}
|
||||||
|
env := ts.NewTestWorkflowEnvironment()
|
||||||
|
|
||||||
|
input := SynthesisInput{
|
||||||
|
Project: "poimen",
|
||||||
|
Source: "transcript://test-123",
|
||||||
|
Text: "Kubernetes uses port 8080 for the API server",
|
||||||
|
Kind: "L1",
|
||||||
|
}
|
||||||
|
|
||||||
|
registerSynthesisActivities(env)
|
||||||
|
|
||||||
|
// Stage 1: Chunk + Embed
|
||||||
|
env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("chunk-abc123", nil)
|
||||||
|
|
||||||
|
// Stage 2: Entity Extraction
|
||||||
|
entities := []ExtractedEntity{
|
||||||
|
{Name: "Kubernetes", EntityType: "tool", Confidence: 0.95},
|
||||||
|
{Name: "API server", EntityType: "component", Confidence: 0.90},
|
||||||
|
}
|
||||||
|
env.OnActivity(ExtractEntitiesActivity, mock.Anything, "chunk-abc123", input.Text).Return(entities, nil)
|
||||||
|
|
||||||
|
// Stage 3: Fact Extraction
|
||||||
|
facts := []ExtractedFact{
|
||||||
|
{Subject: "Kubernetes", Predicate: "uses_port", Object: "8080", Confidence: 0.85},
|
||||||
|
}
|
||||||
|
env.OnActivity(ExtractFactsActivity, mock.Anything, "chunk-abc123", input.Text, entities).Return(facts, nil)
|
||||||
|
|
||||||
|
// Stage 4: Contradiction Detection
|
||||||
|
contradictions := []ContradictionResult{}
|
||||||
|
env.OnActivity(DetectContradictionsActivity, mock.Anything, "poimen", facts).Return(contradictions, nil)
|
||||||
|
|
||||||
|
// Stage 5: Persist
|
||||||
|
env.OnActivity(PersistSynthesisActivity, mock.Anything, mock.Anything).Return(nil)
|
||||||
|
|
||||||
|
env.ExecuteWorkflow(SynthesisWorkflow, input)
|
||||||
|
|
||||||
|
assert.True(t, env.IsWorkflowCompleted())
|
||||||
|
assert.NoError(t, env.GetWorkflowError())
|
||||||
|
|
||||||
|
var result SynthesisResult
|
||||||
|
assert.NoError(t, env.GetWorkflowResult(&result))
|
||||||
|
assert.Equal(t, "chunk-abc123", result.ChunkID)
|
||||||
|
assert.Equal(t, 2, result.EntitiesExtracted)
|
||||||
|
assert.Equal(t, 1, result.FactsExtracted)
|
||||||
|
assert.Equal(t, 0, result.Contradictions)
|
||||||
|
assert.Equal(t, 0, result.ReviewQueued)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSynthesisWorkflow_WithContradictions(t *testing.T) {
|
||||||
|
ts := &testsuite.WorkflowTestSuite{}
|
||||||
|
env := ts.NewTestWorkflowEnvironment()
|
||||||
|
registerSynthesisActivities(env)
|
||||||
|
|
||||||
|
input := SynthesisInput{
|
||||||
|
Project: "poimen",
|
||||||
|
Source: "transcript://test-456",
|
||||||
|
Text: "Port 8080 is used by nginx",
|
||||||
|
Kind: "L1",
|
||||||
|
}
|
||||||
|
|
||||||
|
env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("chunk-def456", nil)
|
||||||
|
|
||||||
|
entities := []ExtractedEntity{
|
||||||
|
{Name: "nginx", EntityType: "tool", Confidence: 0.92},
|
||||||
|
}
|
||||||
|
env.OnActivity(ExtractEntitiesActivity, mock.Anything, "chunk-def456", input.Text).Return(entities, nil)
|
||||||
|
|
||||||
|
facts := []ExtractedFact{
|
||||||
|
{Subject: "nginx", Predicate: "uses_port", Object: "8080", Confidence: 0.88},
|
||||||
|
}
|
||||||
|
env.OnActivity(ExtractFactsActivity, mock.Anything, "chunk-def456", input.Text, entities).Return(facts, nil)
|
||||||
|
|
||||||
|
contradictions := []ContradictionResult{
|
||||||
|
{
|
||||||
|
FactA: ExtractedFact{Subject: "Kubernetes", Predicate: "uses_port", Object: "8080"},
|
||||||
|
FactB: ExtractedFact{Subject: "nginx", Predicate: "uses_port", Object: "8080"},
|
||||||
|
Severity: "medium",
|
||||||
|
AutoResolved: false,
|
||||||
|
QueuedReview: true,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
env.OnActivity(DetectContradictionsActivity, mock.Anything, "poimen", facts).Return(contradictions, nil)
|
||||||
|
env.OnActivity(PersistSynthesisActivity, mock.Anything, mock.Anything).Return(nil)
|
||||||
|
|
||||||
|
env.ExecuteWorkflow(SynthesisWorkflow, input)
|
||||||
|
|
||||||
|
assert.True(t, env.IsWorkflowCompleted())
|
||||||
|
assert.NoError(t, env.GetWorkflowError())
|
||||||
|
|
||||||
|
var result SynthesisResult
|
||||||
|
assert.NoError(t, env.GetWorkflowResult(&result))
|
||||||
|
assert.Equal(t, 1, result.Contradictions)
|
||||||
|
assert.Equal(t, 1, result.ReviewQueued)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSynthesisWorkflow_EntityExtractionFails(t *testing.T) {
|
||||||
|
ts := &testsuite.WorkflowTestSuite{}
|
||||||
|
env := ts.NewTestWorkflowEnvironment()
|
||||||
|
registerSynthesisActivities(env)
|
||||||
|
|
||||||
|
input := SynthesisInput{
|
||||||
|
Project: "poimen",
|
||||||
|
Source: "transcript://test-789",
|
||||||
|
Text: "Some text",
|
||||||
|
Kind: "L1",
|
||||||
|
}
|
||||||
|
|
||||||
|
env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("chunk-xyz", nil)
|
||||||
|
env.OnActivity(ExtractEntitiesActivity, mock.Anything, "chunk-xyz", input.Text).
|
||||||
|
Return(nil, assert.AnError)
|
||||||
|
|
||||||
|
env.ExecuteWorkflow(SynthesisWorkflow, input)
|
||||||
|
|
||||||
|
assert.True(t, env.IsWorkflowCompleted())
|
||||||
|
assert.Error(t, env.GetWorkflowError())
|
||||||
|
assert.Contains(t, env.GetWorkflowError().Error(), "stage 2 entity extraction")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSynthesisWorkflow_ChunkFails(t *testing.T) {
|
||||||
|
ts := &testsuite.WorkflowTestSuite{}
|
||||||
|
env := ts.NewTestWorkflowEnvironment()
|
||||||
|
registerSynthesisActivities(env)
|
||||||
|
|
||||||
|
input := SynthesisInput{
|
||||||
|
Project: "poimen",
|
||||||
|
Source: "transcript://test-fail",
|
||||||
|
Text: "Bad text",
|
||||||
|
Kind: "L1",
|
||||||
|
}
|
||||||
|
|
||||||
|
env.OnActivity(ChunkAndEmbedActivity, mock.Anything, input).Return("", assert.AnError)
|
||||||
|
|
||||||
|
env.ExecuteWorkflow(SynthesisWorkflow, input)
|
||||||
|
|
||||||
|
assert.True(t, env.IsWorkflowCompleted())
|
||||||
|
assert.Error(t, env.GetWorkflowError())
|
||||||
|
assert.Contains(t, env.GetWorkflowError().Error(), "stage 1 chunk+embed")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSynthesisInput_Fields(t *testing.T) {
|
||||||
|
input := SynthesisInput{
|
||||||
|
Project: "test",
|
||||||
|
Source: "source://1",
|
||||||
|
Text: "hello",
|
||||||
|
Kind: "L2",
|
||||||
|
Tags: []string{"tag1", "tag2"},
|
||||||
|
}
|
||||||
|
assert.Equal(t, "test", input.Project)
|
||||||
|
assert.Equal(t, "L2", input.Kind)
|
||||||
|
assert.Len(t, input.Tags, 2)
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user