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 }