Author SHA1 Message Date
Test dab2cd9e90 fix: resolve merge conflicts in CI workflow
CI / Test (pull_request) Successful in 2m11s
CI / Build & Push Image (pull_request) Skipped
Resolved conflicts between fix/registry-login-env and origin/main:
- Use single-line apt-get install
- Use simplified short SHA command
- Use 'Build and push image' step name from main

All registry login changes preserved (env vars approach).
2026-09-06 23:40:32 -07:00
Test 101ba70b57 fix: use env vars for docker registry credentials
CI / Test (pull_request) Successful in 2m24s
CI / Build & Push Image (pull_request) Skipped
Pass FORGEJO_REGISTRY_USER and FORGEJO_REGISTRY_TOKEN via environment
variables instead of direct secret interpolation. This is the standard
approach used across all repos and prevents credentials from being exposed
in logs or shell history.

Fixes registry login failures by using the proven pattern from riotpiao.com.
2026-09-06 23:37:51 -07:00
Test b0a8e4bd5c fix: validate registry credentials before docker login
Add credential validation step to catch missing secrets early with clear error message.
Use direct secret injection (not env vars) for better security.
Isolate docker config to /tmp/docker-config.
2026-09-06 23:35:02 -07:00
Test 3d1360a135 fix: standardize poimen-workflows CI to unified pattern
CI / Test (pull_request) Successful in 2m35s
CI / Build & Push Image (pull_request) Skipped
Unified pattern enforced:
- test job: runs on all branches + PRs
- build-push job: only on main push, depends on test
- Proper env vars (GOPRIVATE, REGISTRY, IMAGE)
- Install Node.js before checkout
- Install docker only in build-push
- Docker login + build + push + prune
2026-09-06 23:17:10 -07:00
7 changed files with 98 additions and 903 deletions
+20 -11
View File
@@ -5,23 +5,19 @@ on:
branches: [main]
pull_request:
branches: [main]
workflow_dispatch:
env:
GOPRIVATE: forgejo.riotpiao.com
REGISTRY: forgejo.riotpiao.com
IMAGE: forgejo.riotpiao.com/rock/poimen-workflows
DOCKER_HOST: tcp://localhost:2375
jobs:
ci:
name: CI
test:
name: Test
runs-on: golang
steps:
- name: Install Node.js and Docker
run: |
apt-get update
apt-get install -y nodejs docker.io
- name: Install Node.js for actions runtime
run: apt-get update && apt-get install -y nodejs
- name: Checkout code
uses: actions/checkout@v4
@@ -38,6 +34,18 @@ jobs:
- name: Build binary
run: CGO_ENABLED=0 GOOS=linux go build -o /tmp/poimen-worker ./cmd/worker
build-push:
name: Build & Push Image
needs: test
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
runs-on: golang
steps:
- name: Install Node.js and Docker
run: apt-get update && apt-get install -y nodejs docker.io
- name: Checkout code
uses: actions/checkout@v4
- name: Get short SHA
id: sha
run: echo "short_sha=$(git rev-parse --short HEAD)" >> $GITHUB_OUTPUT
@@ -50,17 +58,18 @@ jobs:
REGISTRY_USER: ${{ secrets.FORGEJO_REGISTRY_USER }}
REGISTRY_TOKEN: ${{ secrets.FORGEJO_REGISTRY_TOKEN }}
- name: Build Docker image
- name: Build and push image
run: |
docker build --no-cache \
-t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \
-t "${IMAGE}:latest" .
-t "${IMAGE}:latest" \
.
- name: Push Docker image
run: |
docker push "${IMAGE}:${{ steps.sha.outputs.short_sha }}"
docker push "${IMAGE}:latest"
echo "✓ Pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
echo "✓ Image pushed: ${IMAGE}:${{ steps.sha.outputs.short_sha }}"
- name: Prune unused images
run: docker image prune -a --force 2>&1 | tail -3 || true
+78
View File
@@ -0,0 +1,78 @@
# Forgejo Registry Secrets Configuration
## One-Time Setup (Org Level)
All repos in the `rock` org share the same Forgejo registry credentials.
### Configure at Organization Level
1. Navigate to: https://forgejo.riotpiao.com/rock
2. Click Settings (gear icon)
3. Go to: Actions → Secrets
4. Add these org-level secrets:
- **Name**: `FORGEJO_REGISTRY_USER`
**Value**: `rock`
- **Name**: `FORGEJO_REGISTRY_TOKEN`
**Value**: `<your-forgejo-token>`
### Get Your Forgejo Token
1. Go to: https://forgejo.riotpiao.com/user/settings/applications
2. Click "Generate New Token"
3. Set scopes: `api`, `read:registry`, `write:registry`
4. Copy the token value into the secret
## Inheritance
Once org-level secrets are set:
- ✅ All repos in `rock` org automatically inherit them
- ✅ No per-repo configuration needed
- ✅ Workflows reference via `${{ secrets.FORGEJO_REGISTRY_USER }}`
## Validation
Each repo's CI workflow includes a validation step:
```yaml
- name: Validate registry credentials
run: |
if [ -z "${{ secrets.FORGEJO_REGISTRY_USER }}" ] || [ -z "${{ secrets.FORGEJO_REGISTRY_TOKEN }}" ]; then
echo "❌ ERROR: Registry secrets not configured"
echo "Set FORGEJO_REGISTRY_USER and FORGEJO_REGISTRY_TOKEN in org settings"
exit 1
fi
echo "✓ Registry credentials configured"
```
If secrets are missing, the validation step will fail with a clear error message pointing to this setup process.
## Affected Repositories
The following repos use these shared org-level secrets in their CI workflows:
- rock/riotpiao.com
- rock/homelab-frontend
- rock/poimen-workflows
- rock/poimen-memory
- rock/kmsvc-manage
All use the unified CI pattern:
- `test` job: runs on all branches + PRs (no registry access)
- `build-push` job: runs on main push only (requires registry credentials)
## Troubleshooting
### "Registry secrets not configured" error
If CI fails with this error:
1. Check org settings: https://forgejo.riotpiao.com/rock/settings/actions/secrets
2. Verify both secrets exist and are not empty
3. Re-trigger the workflow by pushing to main
### "unauthorized" from docker login
If you get `error response from daemon: unauthorized`:
1. Check the token value is correct (copy-paste carefully)
2. Verify token has `read:registry` and `write:registry` scopes
3. Generate a new token if the old one expired
-346
View File
@@ -1,346 +0,0 @@
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)
}
-183
View File
@@ -1,183 +0,0 @@
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
}
-59
View File
@@ -1,59 +0,0 @@
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"`
}
-125
View File
@@ -1,125 +0,0 @@
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
}
-179
View File
@@ -1,179 +0,0 @@
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)
}