clustering

package
v1.0.0-beta.159 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 31, 2026 License: MIT Imports: 23 Imported by: 0

README

Clustering Package

Community detection and hierarchical clustering algorithms for the SemStreams entity graph.

Purpose

This package provides graph clustering capabilities using Label Propagation Algorithm (LPA) and PageRank to detect communities of related entities. Detected communities are enriched with statistical summaries and optionally enhanced with LLM-generated descriptions.

Key Types

Community

Represents a detected cluster of related entities in the graph:

type Community struct {
    ID                 string              // Unique community identifier
    Level              int                 // Hierarchy level (0=bottom, 1=mid, 2=top)
    Members            []string            // Entity IDs in this community
    ParentID           *string             // Parent community at next level (nil for top)
    StatisticalSummary string              // Fast baseline summary (always present)
    LLMSummary         string              // Enhanced description (populated async)
    Keywords           []string            // Key terms representing themes
    RepEntities        []string            // Representative entity IDs
    SummaryStatus      string              // "statistical", "llm-enhanced", or "llm-failed"
    Metadata           map[string]interface{} // Additional properties
}

Note on field access: This package uses direct field access, not getter methods. This is idiomatic Go - simply reference community.ID, community.Members, etc.

CommunityDetector

Interface for running community detection on the entity graph:

type CommunityDetector interface {
    DetectCommunities(ctx context.Context) (map[int][]*Community, error)
    UpdateCommunities(ctx context.Context, entityIDs []string) error
    GetCommunity(ctx context.Context, id string) (*Community, error)
    GetEntityCommunity(ctx context.Context, entityID string, level int) (*Community, error)
    GetCommunitiesByLevel(ctx context.Context, level int) ([]*Community, error)
}

Configuration

Configure clustering behavior using builder pattern methods:

detector := clustering.NewLPADetector(provider, storage).
    WithMaxIterations(100).     // Maximum LPA iterations
    WithLevels(3).               // Hierarchy levels (0=bottom, 1=mid, 2=top)
    WithProgressiveSummarization(summarizer, entityProvider)

Available configuration methods:

  • WithMaxIterations(int): Set max iteration count (default: 100, max: 10000)
  • WithLevels(int): Set hierarchy depth (default: 3, max: 10)
  • WithProgressiveSummarization(CommunitySummarizer, EntityProvider): Enable summarization

Usage Example

// Create graph provider for specific entity types
provider := clustering.NewPredicateGraphProvider(queryManager, "robotics.drone")

// Create storage backend
storage := clustering.NewNATSStorage(js, "COMMUNITIES")

// Initialize detector
detector := clustering.NewLPADetector(provider, storage)

// Detect communities
communities, err := detector.DetectCommunities(ctx)
if err != nil {
    return fmt.Errorf("community detection failed: %w", err)
}

// Process results by hierarchy level
for level, comms := range communities {
    log.Printf("Level %d: %d communities", level, len(comms))
    for _, comm := range comms {
        // Direct field access - no getters needed
        log.Printf("  Community %s: %d members", comm.ID, len(comm.Members))
        log.Printf("    Keywords: %v", comm.Keywords)
        log.Printf("    Summary: %s", comm.StatisticalSummary)
    }
}

Integration with QueryManager

The package defines local interfaces to avoid import cycles with querymanager:

type RelationshipQuerier interface {
    EntityQuerier
    QueryRelationships(ctx context.Context, entityID string, direction Direction) ([]*Relationship, error)
    QueryByPredicate(ctx context.Context, predicate string) ([]string, error)
}

Two graph provider implementations are available:

QueryManagerGraphProvider

Direct integration with QueryManager for neighborhood queries:

provider := clustering.NewQueryManagerGraphProvider(queryManager)

Limitation: Does not support full graph scans (GetAllEntityIDs returns error).

PredicateGraphProvider

Recommended for real-world use - clusters entities matching a specific predicate:

// Cluster only drone entities
provider := clustering.NewPredicateGraphProvider(queryManager, "robotics.drone.type")

Caches the entity set at construction for efficient repeated queries.

Algorithms

Label Propagation (LPA)

Detects communities by iteratively propagating labels through the graph until convergence:

  1. Each entity starts with unique label
  2. Iteratively adopt most common neighbor label
  3. Repeat until stable or max iterations reached

Produces bottom-level communities, which are then hierarchically aggregated.

PageRank

Identifies representative entities within each community based on graph centrality. Representatives best exemplify the community's characteristics.

Summarization

Two-tier approach for community descriptions:

  1. Statistical Summary (instant): TF-IDF keyword extraction + template-based generation
  2. LLM Enhancement (async): Optional enrichment via background workers

Check community.SummaryStatus to determine which summary type is available.

Storage

Communities are persisted in NATS KV buckets with configurable retention:

storage := clustering.NewNATSStorage(js, "COMMUNITIES")

Supports incremental updates - only recompute affected communities on graph changes.

Package Location

Previously located at pkg/graphclustering/, this package was moved to processor/graph/clustering/ per ADR-PACKAGE-RESPONSIBILITIES-CONSOLIDATION. The move eliminated import cycles and clarified that clustering is graph processing logic, not a standalone reusable library.

All graph processing capabilities now live under processor/graph/:

  • processor/graph/ - Main processor and mutations
  • processor/graph/querymanager/ - Query execution
  • processor/graph/indexmanager/ - Indexing operations
  • processor/graph/clustering/ - Community detection (this package)
  • processor/graph/embedding/ - Vector embeddings

Documentation

Overview

Package clustering provides community detection algorithms and graph clustering for discovering structural patterns in the knowledge graph.

Overview

The clustering package implements the Label Propagation Algorithm (LPA) for detecting communities of related entities. It supports hierarchical clustering with multiple granularity levels, enabling both fine-grained local communities and coarse-grained global clusters.

Detected communities are enriched with statistical summaries (TF-IDF keyword extraction) immediately, with optional LLM enhancement performed asynchronously. PageRank is used to identify representative entities within each community.

Architecture 

                                Graph Provider
                                      ↓
┌──────────────────────────────────────────────────────────────┐
│                      LPA Detector                            │
├──────────────────────────────────────────────────────────────┤
│  Level 0 (fine)  →  Level 1 (mid)  →  Level 2 (coarse)      │
└──────────────────────────────────────────────────────────────┘
                                      ↓
┌────────────────────┐    ┌────────────────────────────────────┐
│ Progressive        │───→│ COMMUNITY_INDEX KV                 │
│ Summarizer         │    │ - {level}.{community_id}           │
│ (statistical)      │    │ - entity.{level}.{entity_id}       │
└────────────────────┘    └────────────────────────────────────┘
                                      ↓
                          ┌───────────────────────┐
                          │ Enhancement Worker    │
                          │ (async LLM via KV     │
                          │  watcher)             │
                          └───────────────────────┘

Usage

Configure and run community detection: 

// Create storage backed by NATS KV
storage := clustering.NewNATSCommunityStorage(communityBucket)

// Create LPA detector with progressive summarization
detector := clustering.NewLPADetector(graphProvider, storage).
    WithLevels(3).
    WithMaxIterations(100).
    WithProgressiveSummarization(
        clustering.NewProgressiveSummarizer(),
        entityProvider,
    )

// Run detection (typically after graph updates)
communities, err := detector.DetectCommunities(ctx)
// communities[0] = fine-grained, communities[1] = mid, communities[2] = coarse

Query communities and entities: 

// Get community for an entity at specific level
community, err := detector.GetEntityCommunity(ctx, entityID, 0)

// Get all communities at a level
allLevel0, err := detector.GetCommunitiesByLevel(ctx, 0)

// Infer relationships from community co-membership
triples, err := detector.InferRelationshipsFromCommunities(ctx, 0, clustering.DefaultInferenceConfig())

Label Propagation Algorithm

LPA iteratively assigns entities to communities based on neighbor voting:

  1. Each entity starts with its own unique label
  2. Entities adopt the most frequent label among their neighbors (weighted by edge weight)
  3. Process continues until convergence or max iterations reached
  4. Shuffled processing order reduces oscillation

Hierarchical levels are computed by treating communities from level N as super-nodes for level N+1 detection.

Community Summarization

Three summarization strategies are available:

StatisticalSummarizer:

  • TF-IDF-like keyword extraction from entity types and properties
  • PageRank-based representative entity selection
  • Template-based summary generation
  • Always available, no external dependencies

LLMSummarizer:

  • OpenAI-compatible LLM for natural language summaries
  • Works with seminstruct, OpenAI, Ollama, vLLM, etc.
  • Falls back to statistical on service unavailability

ProgressiveSummarizer:

  • Statistical summary immediately available
  • LLM enhancement performed asynchronously
  • Best for user-facing applications needing fast initial response

Enhancement Worker

The EnhancementWorker watches COMMUNITY_INDEX KV as a TRIGGER ONLY and writes LLM summaries to the worker-owned, content-addressed COMMUNITY_SUMMARIES store keyed by {level}.{membership_hash} (ADR-087). It never writes COMMUNITY_INDEX, so a lagging worker can neither clobber a fresher partition nor resurrect a pruned community: 

worker, err := clustering.NewEnhancementWorker(&clustering.EnhancementWorkerConfig{
    LLMSummarizer:   llmSummarizer,
    Querier:         queryManager,
    CommunityBucket: communityBucket, // COMMUNITY_INDEX — trigger only
    SummaryBucket:   summaryBucket,   // COMMUNITY_SUMMARIES — worker-owned
})

worker.Start(ctx)
defer worker.Stop()

PageRank

PageRank identifies influential entities within communities: 

config := clustering.DefaultPageRankConfig()
config.TopN = 10

result, err := clustering.ComputePageRankForCommunity(ctx, provider, memberIDs, config)
// result.Ranked = top 10 entities by PageRank score
// result.Scores = map of entity ID to normalized score

Default configuration:

  • Iterations: 20
  • DampingFactor: 0.85
  • Tolerance: 1e-6 (convergence threshold)

Configuration

LPA detector configuration: 

MaxIterations:    100       # Maximum iterations (limit: 10000)
Levels:           3         # Hierarchical levels (limit: 10)

LLM summaries survive rebuilds by content-addressing (membership hash), not by a Jaccard-overlap transfer between runs — see the Enhancement Worker section and ADR-087.

Inference configuration for relationship generation: 

MinCommunitySize:        2     # Minimum size for inference
MaxInferredPerCommunity: 50    # Limit to prevent O(n²) explosion

PageRank configuration: 

Iterations:    20        # Max iterations
DampingFactor: 0.85      # Random walk continuation probability
Tolerance:     1e-6      # Convergence threshold
TopN:          0         # Return all (or limit to top N)

Storage

Communities are stored in the COMMUNITY_INDEX KV bucket: 

{level}.{community_id}        → Community JSON
entity.{level}.{entity_id}   → Community ID (for entity lookup)

Storage can optionally create member_of triples: 

config := clustering.CommunityStorageConfig{
    CreateTriples:   true,
    TriplePredicate: "graph.community.member-of",
}
storage := clustering.NewNATSCommunityStorageWithConfig(kv, config)

Thread Safety

LPADetector, NATSCommunityStorage, and EnhancementWorker are safe for concurrent use. The enhancement worker supports pause/resume for coordinated graph updates: 

worker.Pause()    // Stop processing new communities
// ... perform graph updates ...
worker.Resume()   // Continue processing

Metrics

The clustering package exports Prometheus metrics under the semstreams_clustering namespace:

  • communities_detected_total: Communities detected by level
  • detection_duration_seconds: Detection run duration
  • llm_enhancement_latency_seconds: LLM summarization duration
  • community_summary_cache_hits_total: Triggers served from an existing summary (no LLM call)
  • community_summary_generated_total: Summaries produced by a fresh LLM call
  • community_summary_failed_total: Failed summary generations
  • community_summaries_size: Current count of stored community summaries

See Also

Related packages:

Package clustering provides graph clustering algorithms and community detection.

Package clustering provides community detection algorithms and graph providers.

Index

Constants

View Source 
const (
	// DefaultMaxIterations is the default maximum iteration count
	DefaultMaxIterations = 100

	// MaxIterationsLimit is the maximum allowed iteration count
	MaxIterationsLimit = 10000

	// DefaultLevels is the default number of hierarchical levels
	DefaultLevels = 3

	// MaxLevelsLimit is the maximum allowed hierarchical levels
	MaxLevelsLimit = 10
)
View Source 
const (
	// DefaultSemanticMaxNeighbors is the mutual-kNN k: the per-direction top-k
	// candidate set size. Bounds per-entity semantic degree so the new tier
	// competes with, rather than dominates, the structural edges.
	DefaultSemanticMaxNeighbors = 8
	// DefaultSemanticThreshold is the minimum similarity for a candidate to
	// count toward the top-k directed set.
	DefaultSemanticThreshold = 0.75
	// DefaultSemanticEdgeWeight is the synthesized semantic virtual-edge weight
	// — below explicit (1.0) but above the rebalanced structural tiers, so a
	// thematically-related but structurally-heterogeneous pair can still be
	// voted together by LPA.
	DefaultSemanticEdgeWeight = 0.9
)

Semantic-edge starting values (EMPIRICAL, ADR-086 / design.md). These are the semantic-enabled profile's STARTING point, to be tuned against `partition_colocation_mean` on the theme-spanning fixture queries — recorded here as measured-not-asserted, NOT final. They apply only when the operator enables the semantic-edge tier (`enable_semantic_edges`); an unopted deployment is untouched (gh#461 invariant).

View Source 
const (
	// SummaryStatusEnhanced marks a record that carries a usable LLM summary.
	SummaryStatusEnhanced = "llm-enhanced"
	// SummaryStatusFailed marks a record whose enhancement failed; the worker
	// retries it only after the failed-retry backoff elapses.
	SummaryStatusFailed = "llm-failed"
)

Summary record status values. A record is either an LLM-enhanced summary or a record of a failed enhancement (which the worker re-attempts only after a backoff). These are SEPARATE from the detector-owned Community.SummaryStatus enum on COMMUNITY_INDEX — the split is the whole point of ADR-087.

View Source 
const (

	// MaxCommunityLevels is the maximum hierarchy depth for scanning
	MaxCommunityLevels = 10
)

Variables

View Source 
var ErrSemanticIndexNotReady = errors.New("clustering: semantic similarity index not ready")

ErrSemanticIndexNotReady signals that the similarity source could not be queried because the embedding index is not ready — a classified transient, NOT a genuine empty result. A SemanticNeighborFinder returns it (wrapped) so refreshCache ABORTS the whole mutual-kNN build rather than committing a partial/empty adjacency: the cache is not settled, the cycle degrades to structural-only, and a later cycle (gated on a re-confirmed ready index) rebuilds it fresh. The concrete ErrorCodeIndexNotReady classification lives at the component's finder adapter (which imports graph); this leaf package stays free of that import and reacts only to the neutral sentinel (B2 §5.1).

View Source 
var ErrSemanticQueryTransient = errors.New("clustering: semantic similarity query failed transiently")

ErrSemanticQueryTransient signals that a SINGLE entity's similarity query failed with a non-index_not_ready transient (a timeout or no-responders on that one RPC, NOT the whole index being cold). Unlike ErrSemanticIndexNotReady it does NOT abort the whole refresh: refreshCache counts it toward the coverage-threshold (B2 §7.3 / #662) and leaves the entity UN-cached so it is re-queried next cycle, never latching a hollow "no semantic neighbors" for it.

It exists because the readiness-aware finder adapter used to swallow every non-index_not_ready error into a bare empty result, making a per-entity timeout indistinguishable from a genuine "this entity has no close neighbors" — the exact confusion that let a burst of query timeouts latch a semantically-blind cache permanently (#662). Surfacing the transient class separately is what lets the coverage-threshold abort see it.

Functions

func ComputeRepresentativeEntities 

func ComputeRepresentativeEntities(ctx context.Context, provider Provider, communityMembers []string, topN int) ([]string, map[string]float64, error)

ComputeRepresentativeEntities computes representative entities for a community using PageRank Returns the top N entities by PageRank score

func MembershipHash 

func MembershipHash(members []string) string

MembershipHash returns the content-address of a community's membership: the hex-encoded sha256 of the lexically-sorted member IDs joined by "\n".

This is the SINGLE shared definition of the membership hash. It keys the worker-owned COMMUNITY_SUMMARIES store, is joined by the graph-query read path, and is computed by the B0 thematic eval — all three MUST derive the hash through this helper so the definition cannot drift into two subtly different hashes that never join (ADR-087). The input slice is copied before sorting, so the hash is sort-stable and independent of caller order or map-iteration order; the hash carries no level (summarization has no level input) — level lives only in the KV key prefix.

func SummaryKey 

func SummaryKey(level int, membershipHash string) string

SummaryKey builds the {level}.{membership_hash} COMMUNITY_SUMMARIES KV key. It is the ONE definition of the key format, shared by the worker's store and the graph-query read-join so the two cannot build keys that never match. The hash is hex (no dots), so a first-dot split unambiguously recovers the level.

Types

type Community 

type Community struct {
	// ID is the unique identifier for this community
	ID string `json:"id"`

	// Level indicates the hierarchy level (0=bottom, 1=mid, 2=top)
	Level int `json:"level"`

	// Members contains the entity IDs belonging to this community
	Members []string `json:"members"`

	// ParentID references the parent community at the next level up (nil for top level)
	ParentID *string `json:"parent_id,omitempty"`

	// StatisticalSummary is the fast statistical baseline summary (always present)
	// Generated using TF-IDF keyword extraction and template-based summarization
	StatisticalSummary string `json:"statistical_summary,omitempty"`

	// LLMSummary is the enhanced LLM-generated summary (populated asynchronously)
	// Empty until LLM enhancement completes successfully
	LLMSummary string `json:"llm_summary,omitempty"`

	// Keywords are extracted key terms representing this community's themes
	// e.g., ["autonomous", "navigation", "sensor-fusion"]
	Keywords []string `json:"keywords,omitempty"`

	// RepEntities contains IDs of representative entities within this community
	// These entities best exemplify the community's characteristics
	RepEntities []string `json:"rep_entities,omitempty"`

	// SummaryStatus tracks the summarization state
	// Values: "statistical" (initial), "llm-enhanced" (enhanced), "llm-failed" (enhancement failed)
	SummaryStatus string `json:"summary_status,omitempty"`

	// SummaryTruncated is true when the LLM summary hit the token budget
	// (finish_reason "length"). A SEPARATE flag, not a SummaryStatus enum value, so
	// exact-match consumers of SummaryStatus (e.g. lpa.go archival-preservation)
	// are unaffected. Read by the B2 partition-colocation recorder's dilution
	// channel to attribute a recall miss (truncated summary vs semantically diluted).
	SummaryTruncated bool `json:"summary_truncated,omitempty"`

	// Metadata stores additional community properties
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

Community represents a detected community/cluster in the graph

type CommunityDetector 

type CommunityDetector interface {
	// DetectCommunities runs community detection on the entire graph
	// Returns communities organized by hierarchical level
	DetectCommunities(ctx context.Context) (map[int][]*Community, error)

	// UpdateCommunities incrementally updates communities based on recent graph changes
	// entityIDs are entities that have been added/modified since last detection
	UpdateCommunities(ctx context.Context, entityIDs []string) error

	// GetCommunity retrieves a specific community by ID
	GetCommunity(ctx context.Context, id string) (*Community, error)

	// GetEntityCommunity returns the community containing the given entity
	// level specifies which hierarchical level to query (0=bottom, 1=mid, 2=top)
	GetEntityCommunity(ctx context.Context, entityID string, level int) (*Community, error)

	// GetCommunitiesByLevel returns all communities at a specific hierarchical level
	GetCommunitiesByLevel(ctx context.Context, level int) ([]*Community, error)

	// InferRelationshipsFromCommunities generates inferred triples from community co-membership.
	// For each community with >= minCommunitySize members, creates bidirectional
	// "inferred.clustered_with" triples between members.
	InferRelationshipsFromCommunities(ctx context.Context, level int, config InferenceConfig) ([]InferredTriple, error)
}

CommunityDetector performs community detection on a graph

type CommunityStorage 

type CommunityStorage interface {
	// SaveCommunity persists a community
	SaveCommunity(ctx context.Context, community *Community) error

	// GetCommunity retrieves a community by ID
	GetCommunity(ctx context.Context, id string) (*Community, error)

	// GetCommunitiesByLevel retrieves all communities at a level
	GetCommunitiesByLevel(ctx context.Context, level int) ([]*Community, error)

	// GetEntityCommunity retrieves the community for an entity at a level
	GetEntityCommunity(ctx context.Context, entityID string, level int) (*Community, error)

	// DeleteCommunity removes a community
	DeleteCommunity(ctx context.Context, id string) error

	// Prune removes stored state that does not belong to the supplied partition.
	//
	// This is the replacement half of an in-place index rebuild: a detector
	// overwrites keys with SaveCommunity as it goes and then calls Prune once,
	// at the end, with the complete new partition. Everything the previous
	// partition left behind is dropped; everything in keep is retained.
	//
	// Detectors MUST NOT Clear() before a rebuild. Detection takes seconds, and
	// a cleared index publishes an authoritative-looking empty answer for that
	// whole window. Write-then-prune means readers see old ∪ new instead — a
	// slightly stale partition, never an empty one (ADR-085).
	Prune(ctx context.Context, keep []*Community) error

	// Clear removes all communities.
	//
	// Only for teardown and explicit operator-driven reset. Not part of the
	// rebuild path — see Prune.
	Clear(ctx context.Context) error

	// GetAllCommunities returns all communities across all levels
	// Used for archiving enhanced communities before a rebuild
	GetAllCommunities(ctx context.Context) ([]*Community, error)
}

CommunityStorage abstracts persistence layer for communities

type CommunityStorageConfig 

type CommunityStorageConfig struct {
	// CreateTriples enables creation of member-of triples during SaveCommunity
	CreateTriples bool

	// TriplePredicate specifies the predicate to use for community membership triples
	// Default: "graph.community.member-of"
	TriplePredicate string
}

CommunityStorageConfig configures community storage behavior

type CommunitySummarizer 

type CommunitySummarizer interface {
	// SummarizeCommunity generates a summary for a community
	// Returns updated Community with Summary, Keywords, RepEntities, and Summarizer fields populated
	SummarizeCommunity(ctx context.Context, community *Community, entities []*gtypes.EntityState) (*Community, error)
}

CommunitySummarizer generates summaries for communities

type CommunitySummaryRecord 

type CommunitySummaryRecord struct {
	// MembershipHash is the content-address (clustering.MembershipHash) of the
	// membership this summary describes. It is the hash half of the KV key.
	MembershipHash string `json:"membership_hash"`

	// Level is the hierarchy level of the community. It is the level half of the
	// KV key; the hash itself is level-independent.
	Level int `json:"level"`

	// LLMSummary is the generated natural-language summary (empty on a failed record).
	LLMSummary string `json:"llm_summary,omitempty"`

	// Model is the LLM model identifier that produced the summary.
	Model string `json:"model,omitempty"`

	// Status is SummaryStatusEnhanced or SummaryStatusFailed.
	Status string `json:"status"`

	// Truncated is true when the LLM summary hit the token budget (finish_reason
	// "length").
	Truncated bool `json:"truncated,omitempty"`

	// MemberCount is the size of the membership that was summarized (observability;
	// the hash is the identity).
	MemberCount int `json:"member_count"`

	// GeneratedAt is when the record was written. The failed-retry backoff is
	// measured from this timestamp.
	GeneratedAt time.Time `json:"generated_at"`
}

CommunitySummaryRecord is the worker-owned, content-addressed LLM summary for a community membership. It is stored in COMMUNITY_SUMMARIES keyed by {level}.{membership_hash}. It carries NO full member snapshot — the membership hash IS the identity; storing the members would reintroduce a divergence surface. Keywords and the statistical summary stay detector-owned on COMMUNITY_INDEX; only the LLM prose lives here.

type Direction 

type Direction string

Direction represents the direction of relationship traversal. Local copy to avoid import cycle with querymanager. 

const (
	DirectionOutgoing Direction = "outgoing"
	DirectionIncoming Direction = "incoming"
	DirectionBoth     Direction = "both"
)

Direction constants for relationship traversal.

type EnhancementMetrics 

type EnhancementMetrics struct {
	// contains filtered or unexported fields
}

EnhancementMetrics provides Prometheus metrics for the content-addressed LLM community-summary worker (ADR-087).

There is deliberately NO queue-depth gauge. Content-addressing makes backlog benign: a steady graph re-triggers the same memberships, every trigger is a microsecond cache hit that skips the LLM, and only a NEW distinct membership costs a call. The old Inc/DecQueueDepth gauge bracketed already-dequeued work and was a phantom (#617); it is removed. The store's accumulation is observed instead by the summaries-size gauge (add-3).

func NewEnhancementMetrics 

func NewEnhancementMetrics(component string, registry *metric.MetricsRegistry) *EnhancementMetrics

NewEnhancementMetrics creates a new EnhancementMetrics instance using MetricsRegistry

func (*EnhancementMetrics) RecordCacheHit 

func (m *EnhancementMetrics) RecordCacheHit()

RecordCacheHit records a trigger served from a stored llm-enhanced record (no LLM call performed).

func (*EnhancementMetrics) RecordFailed 

func (m *EnhancementMetrics) RecordFailed(latencySeconds float64)

RecordFailed records a failed enhancement (an llm-failed record was written) with its latency.

func (*EnhancementMetrics) RecordGenerated 

func (m *EnhancementMetrics) RecordGenerated(latencySeconds float64)

RecordGenerated records a summary that was generated by an LLM call (a cache miss that did work) with its latency.

func (*EnhancementMetrics) SetSummariesSize 

func (m *EnhancementMetrics) SetSummariesSize(n int)

SetSummariesSize sets the current number of stored summary records.

type EnhancementWorker 

type EnhancementWorker struct {
	// contains filtered or unexported fields
}

EnhancementWorker consumes COMMUNITY_INDEX changes as a TRIGGER ONLY and writes LLM summaries to the worker-exclusive, content-addressed COMMUNITY_SUMMARIES store. It holds NO CommunityStorage — it structurally cannot write the partition bucket, which is the single-writer invariant that closes #607/#617 (ADR-087).

func NewEnhancementWorker 

func NewEnhancementWorker(config *EnhancementWorkerConfig) (*EnhancementWorker, error)

NewEnhancementWorker creates a new enhancement worker

func (*EnhancementWorker) IsPaused 

func (w *EnhancementWorker) IsPaused() bool

IsPaused returns whether the worker is currently paused.

func (*EnhancementWorker) Pause 

func (w *EnhancementWorker) Pause()

Pause stops processing new communities while allowing in-flight work to complete. Safe to call multiple times. Returns immediately after signaling workers to pause.

func (*EnhancementWorker) Resume 

func (w *EnhancementWorker) Resume()

Resume allows processing to continue after a Pause. Safe to call multiple times.

func (*EnhancementWorker) Start 

func (w *EnhancementWorker) Start(ctx context.Context) error

Start begins watching for communities needing LLM enhancement

func (*EnhancementWorker) Stop 

func (w *EnhancementWorker) Stop() error

Stop gracefully stops the enhancement worker

func (*EnhancementWorker) WithWorkers 

func (w *EnhancementWorker) WithWorkers(n int) *EnhancementWorker

WithWorkers sets the number of concurrent workers. Must be called before Start(). Has no effect if worker is already started.

type EnhancementWorkerConfig 

type EnhancementWorkerConfig struct {
	LLMSummarizer *LLMSummarizer
	Querier       EntityQuerier
	// CommunityBucket is the COMMUNITY_INDEX bucket, watched as a trigger ONLY.
	CommunityBucket jetstream.KeyValue
	// SummaryBucket is the COMMUNITY_SUMMARIES bucket the worker owns and writes.
	SummaryBucket jetstream.KeyValue
	Logger        *slog.Logger
	Registry      *metric.MetricsRegistry // Optional: for summary-worker metrics
	// LLMTimeout caps the per-call inner sub-context that wraps each LLM
	// summarization round-trip. Zero means use the 30s default. The HTTP
	// client's transport-level timeout (set via the OpenAIClient) is the
	// outer ceiling; this is the inner ctx.WithTimeout that fires first
	// for slow upstreams. Both must be at least the operator-intended
	// ceiling — see processor/graph-clustering for the wiring.
	LLMTimeout time.Duration
}

EnhancementWorkerConfig holds configuration for the enhancement worker

type EntityIDProvider 

type EntityIDProvider struct {
	// contains filtered or unexported fields
}

EntityIDProvider wraps a base Provider and adds virtual edges based on EntityID hierarchy. This enables LPA clustering to find communities using the 6-part EntityID structure even when explicit relationship triples don't exist.

EntityID format: org.platform.domain.system.type.instance Entities with the same 5-part TypePrefix (org.platform.domain.system.type) are considered siblings.

Virtual edges are computed on-demand and cached for performance. These edges are NOT persisted - they're ephemeral hints for the clustering algorithm.

Explicit edges (from base provider) always take precedence over virtual edges.

func NewEntityIDProvider 

func NewEntityIDProvider(
	base Provider,
	config EntityIDProviderConfig,
	logger *slog.Logger,
) *EntityIDProvider

NewEntityIDProvider creates a Provider that augments explicit edges with virtual edges based on EntityID hierarchy (6-part dotted format).

Parameters:

  • base: The underlying Provider for explicit edges (also used to list all entities)
  • config: Configuration for edge weights and limits
  • logger: Optional logger for observability (can be nil)

func (*EntityIDProvider) ClearCache 

func (p *EntityIDProvider) ClearCache()

ClearCache clears the type prefix cache and propagates to wrapped providers. Call this when entities are added/removed.

func (*EntityIDProvider) GetAllEntityIDs 

func (p *EntityIDProvider) GetAllEntityIDs(ctx context.Context) ([]string, error)

GetAllEntityIDs delegates to the base provider

func (*EntityIDProvider) GetCacheStats 

func (p *EntityIDProvider) GetCacheStats() (prefixes int, entities int)

GetCacheStats returns statistics about the type prefix cache for monitoring.

func (*EntityIDProvider) GetEdgeWeight 

func (p *EntityIDProvider) GetEdgeWeight(ctx context.Context, fromID, toID string) (float64, error)

GetEdgeWeight returns the weight of an edge between two entities.

For explicit edges: delegates to base provider For sibling edges: returns configured sibling weight (default 0.7)

Explicit edges always take precedence - if base returns weight > 0, that's used directly. Sibling edge weight is only used when no explicit edge exists.

func (*EntityIDProvider) GetNeighbors 

func (p *EntityIDProvider) GetNeighbors(ctx context.Context, entityID string, direction string) ([]string, error)

GetNeighbors returns both explicit neighbors and sibling neighbors from EntityID hierarchy. Sibling neighbors are entities that share the same 5-part type prefix.

Direction parameter is respected for explicit edges but ignored for sibling edges (sibling relationships are symmetric).

func (*EntityIDProvider) GetSiblingEdgeMetrics 

func (p *EntityIDProvider) GetSiblingEdgeMetrics() (successes, errors int64)

GetSiblingEdgeMetrics returns metrics for sibling edge operations.

func (*EntityIDProvider) ResetEdgeCache 

func (p *EntityIDProvider) ResetEdgeCache()

ResetEdgeCache propagates the per-cycle explicit-edge cache reset to the wrapped base provider (gh#666). It deliberately does NOT clear the sibling/system-peer prefix caches — those keep their own lifetime via ClearCache — so a per-cycle edge-cache reset stays scoped to the explicit topology the detector re-snapshots each cycle. A base provider without such a cache is skipped.

type EntityIDProviderConfig 

type EntityIDProviderConfig struct {
	// SiblingWeight is the edge weight for sibling relationships.
	// Higher values = stronger connection influence in LPA.
	// Recommended: 0.7 (lower than explicit edges at 1.0)
	SiblingWeight float64

	// MaxSiblings limits sibling neighbors per entity to control
	// computation cost during LPA iterations.
	// Recommended: 10
	MaxSiblings int

	// IncludeSiblings enables sibling edge discovery.
	// Set to false to disable EntityID-based edges entirely.
	IncludeSiblings bool

	// IncludeSystemPeers enables system-affinity edges between entities
	// sharing the same system (part[3] of the 6-part entity ID).
	// This biases LPA toward system-coherent communities when the graph
	// contains entities from heterogeneous data sources.
	IncludeSystemPeers bool

	// SystemPeerWeight is the edge weight for system-affinity edges.
	// Lower than SiblingWeight because system is a weaker signal than
	// exact type match, but enough to bias LPA toward system-level coherence.
	// Recommended: 0.3
	SystemPeerWeight float64

	// MaxSystemPeers limits system-affinity neighbors per entity.
	// Recommended: 15
	MaxSystemPeers int
}

EntityIDProviderConfig holds configuration for EntityIDProvider

func DefaultEntityIDProviderConfig 

func DefaultEntityIDProviderConfig() EntityIDProviderConfig

DefaultEntityIDProviderConfig returns sensible defaults for clustering

type EntityProvider 

type EntityProvider interface {
	GetEntities(ctx context.Context, ids []string) ([]*gtypes.EntityState, error)
}

EntityProvider interface for fetching full entity states for summarization

type EntityQuerier 

type EntityQuerier interface {
	GetEntities(ctx context.Context, ids []string) ([]*gtypes.EntityState, error)
}

EntityQuerier provides minimal interface for querying entities. This interface exists to avoid import cycle with querymanager package.

type InferenceConfig 

type InferenceConfig struct {
	// MinCommunitySize is the minimum community size for generating inferences
	// Singleton communities (size=1) never produce inferences
	MinCommunitySize int

	// MaxInferredPerCommunity limits inferred relationships per community
	// Prevents O(n²) explosion in large communities
	MaxInferredPerCommunity int
}

InferenceConfig holds configuration for relationship inference

func DefaultInferenceConfig 

func DefaultInferenceConfig() InferenceConfig

DefaultInferenceConfig returns sensible defaults for relationship inference

type InferredTriple 

type InferredTriple struct {
	Subject     string
	Predicate   string
	Object      string
	Source      string
	Confidence  float64
	Timestamp   time.Time
	CommunityID string // Community that produced this inference
	Level       int    // Hierarchical level
}

InferredTriple represents a relationship inferred from community detection. This is a lightweight struct for returning inference results. The caller converts these to message.Triple for persistence.

type LLMSummarizer 

type LLMSummarizer struct {
	// Client is the LLM client for making chat completion requests.
	Client llm.Client

	// FallbackSummarizer is used if LLM service is unavailable.
	FallbackSummarizer *StatisticalSummarizer

	// MaxTokens limits the response length (default: 150).
	MaxTokens int

	// ContentFetcher optionally fetches entity content (title, abstract) for richer prompts.
	// If nil, prompts use only entity IDs and triple-derived keywords.
	ContentFetcher llm.ContentFetcher
}

LLMSummarizer implements CommunitySummarizer using an OpenAI-compatible LLM API. This summarizer calls an external LLM service for higher quality natural language summaries.

It works with any OpenAI-compatible backend:

  • seminstruct (recommended for local llama.cpp inference)
  • OpenAI cloud
  • Ollama, vLLM, etc.

func NewLLMSummarizer 

func NewLLMSummarizer(cfg LLMSummarizerConfig, opts ...LLMSummarizerOption) (*LLMSummarizer, error)

NewLLMSummarizer creates an LLM-based summarizer with the given configuration. Optional functional options can be provided to configure additional features.

func (*LLMSummarizer) BuildCorpusDF 

func (s *LLMSummarizer) BuildCorpusDF(entities []*gtypes.EntityState)

BuildCorpusDF forwards to the fallback statistical summarizer so IDF scoring applies to the keyword path that LLMSummarizer reuses.

func (*LLMSummarizer) SummarizeCommunity 

func (s *LLMSummarizer) SummarizeCommunity(
	ctx context.Context,
	community *Community,
	entities []*gtypes.EntityState,
) (*Community, error)

SummarizeCommunity generates an LLM-based summary of the community. Implements CommunitySummarizer interface with 3-param signature. Content fetching happens internally using the optional ContentFetcher.

type LLMSummarizerConfig 

type LLMSummarizerConfig struct {
	// Client is the LLM client (required).
	Client llm.Client

	// MaxTokens limits the response length (default: 150).
	MaxTokens int
}

LLMSummarizerConfig configures the LLM summarizer.

type LLMSummarizerOption 

type LLMSummarizerOption func(*LLMSummarizer) error

LLMSummarizerOption configures an LLMSummarizer using the functional options pattern. Options return errors for validation (following natsclient pattern).

func WithContentFetcher 

func WithContentFetcher(fetcher llm.ContentFetcher) LLMSummarizerOption

WithContentFetcher sets the ContentFetcher for enriching prompts with entity content. If not set, prompts use only entity IDs and triple-derived keywords.

type LPADetector 

type LPADetector struct {
	// contains filtered or unexported fields
}

LPADetector implements community detection using Label Propagation Algorithm

func NewLPADetector 

func NewLPADetector(provider Provider, storage CommunityStorage) *LPADetector

NewLPADetector creates a new Label Propagation Algorithm detector

func (*LPADetector) DetectCommunities 

func (d *LPADetector) DetectCommunities(ctx context.Context) (map[int][]*Community, error)

DetectCommunities runs full community detection across all hierarchical levels

func (*LPADetector) GetCommunitiesByLevel 

func (d *LPADetector) GetCommunitiesByLevel(ctx context.Context, level int) ([]*Community, error)

GetCommunitiesByLevel returns all communities at a level

func (*LPADetector) GetCommunity 

func (d *LPADetector) GetCommunity(ctx context.Context, id string) (*Community, error)

GetCommunity retrieves a community by ID

func (*LPADetector) GetEntityCommunity 

func (d *LPADetector) GetEntityCommunity(ctx context.Context, entityID string, level int) (*Community, error)

GetEntityCommunity returns the community for an entity at a specific level

func (*LPADetector) InferRelationshipsFromCommunities 

func (d *LPADetector) InferRelationshipsFromCommunities(
	ctx context.Context,
	level int,
	config InferenceConfig,
) ([]InferredTriple, error)

InferRelationshipsFromCommunities generates inferred triples from community co-membership. For each community with >= minCommunitySize members, this creates bidirectional "inferred.clustered_with" triples between members.

Parameters:

  • level: Hierarchical level to process (0 = most granular)
  • config: Inference configuration (min size, max pairs)

Returns triples suitable for persistence via graph.mutation.triple.add. The caller is responsible for persisting these triples.

Confidence scoring:

  • Base confidence: 0.5 (inferred relationships)
  • Adjusted by community tightness: +0.0 to +0.3 based on internal similarity
  • Final range: 0.5-0.8 for inferred relationships

func (*LPADetector) SetEntityProvider 

func (d *LPADetector) SetEntityProvider(provider EntityProvider)

SetEntityProvider sets the entity provider for fetching entities during summarization. This method supports deferred initialization - call after the entity provider becomes available. Both summarizer and entityProvider must be set for summarization to occur.

func (*LPADetector) UpdateCommunities 

func (d *LPADetector) UpdateCommunities(ctx context.Context, _ []string) error

UpdateCommunities incrementally updates communities based on changed entities

func (*LPADetector) WithLevels 

func (d *LPADetector) WithLevels(levels int) *LPADetector

WithLevels sets the number of hierarchical levels with validation

func (*LPADetector) WithLogger 

func (d *LPADetector) WithLogger(logger *slog.Logger) *LPADetector

WithLogger sets the logger for the detector

func (*LPADetector) WithMaxIterations 

func (d *LPADetector) WithMaxIterations(maxN int) *LPADetector

WithMaxIterations sets the maximum iteration count with validation

func (*LPADetector) WithProgressiveSummarization 

func (d *LPADetector) WithProgressiveSummarization(
	summarizer CommunitySummarizer,
	entityProvider EntityProvider,
) *LPADetector

WithProgressiveSummarization enables progressive summarization with LLM enhancement summarizer: generates statistical summaries immediately entityProvider: fetches full entity states for summarization Note: EnhancementWorker watches COMMUNITY_INDEX KV for async LLM enhancement (no NATS events needed)

func (*LPADetector) WithSummarizer 

func (d *LPADetector) WithSummarizer(summarizer CommunitySummarizer) *LPADetector

WithSummarizer sets the summarizer without requiring an entity provider. Use SetEntityProvider() later to enable summarization once the provider is available. This supports deferred initialization patterns where the entity provider isn't available at detector creation time.

type NATSCommunityStorage 

type NATSCommunityStorage struct {
	// contains filtered or unexported fields
}

NATSCommunityStorage implements CommunityStorage using NATS KV

func NewNATSCommunityStorage 

func NewNATSCommunityStorage(kv jetstream.KeyValue) *NATSCommunityStorage

NewNATSCommunityStorage creates a new NATS-backed community storage with default configuration (no triple creation)

func NewNATSCommunityStorageWithConfig 

func NewNATSCommunityStorageWithConfig(kv jetstream.KeyValue, config CommunityStorageConfig) *NATSCommunityStorage

NewNATSCommunityStorageWithConfig creates a new NATS-backed community storage with custom configuration for triple creation

func (*NATSCommunityStorage) Clear 

func (s *NATSCommunityStorage) Clear(ctx context.Context) error

Clear removes all communities and entity mappings.

This is for teardown and explicit operator-driven reset only. Rebuilds use SaveCommunity + Prune so the index is never transiently empty (see Prune).

This is a best-effort operation - context cancellation during cleanup is ignored since partial cleanup is acceptable during shutdown.

func (*NATSCommunityStorage) DeleteCommunity 

func (s *NATSCommunityStorage) DeleteCommunity(ctx context.Context, id string) error

DeleteCommunity removes a community

func (*NATSCommunityStorage) GetAllCommunities 

func (s *NATSCommunityStorage) GetAllCommunities(ctx context.Context) ([]*Community, error)

GetAllCommunities returns all communities across all levels Used by the LPA detector to archive enhanced communities before a rebuild

func (*NATSCommunityStorage) GetCommunitiesByLevel 

func (s *NATSCommunityStorage) GetCommunitiesByLevel(ctx context.Context, level int) ([]*Community, error)

GetCommunitiesByLevel retrieves all communities at a level

func (*NATSCommunityStorage) GetCommunity 

func (s *NATSCommunityStorage) GetCommunity(ctx context.Context, id string) (*Community, error)

GetCommunity retrieves a community by ID. Since community IDs no longer embed the level, this scans all levels to find the community.

func (*NATSCommunityStorage) GetCreatedTriples 

func (s *NATSCommunityStorage) GetCreatedTriples() []message.Triple

GetCreatedTriples returns all triples created during SaveCommunity operations This method is primarily for testing and verification purposes

func (*NATSCommunityStorage) GetEntityCommunity 

func (s *NATSCommunityStorage) GetEntityCommunity(ctx context.Context, entityID string, level int) (*Community, error)

GetEntityCommunity retrieves the community for an entity at a level

func (*NATSCommunityStorage) Prune 

func (s *NATSCommunityStorage) Prune(ctx context.Context, keep []*Community) error

Prune removes every stored key that does not belong to the supplied partition.

The keep set is derived here rather than passed in as raw keys so the KV key format stays private to this file: a caller hands over the communities it just wrote, and this method reconstructs both the community keys ({level}.{id}) and the entity mapping keys (entity.{level}.{entity_id}) they imply.

Passing an empty keep set deletes everything, which is the correct end state for a graph that genuinely has no entities.

Best-effort with respect to shutdown: context cancellation mid-delete is skipped rather than reported, matching Clear.

func (*NATSCommunityStorage) SaveCommunity 

func (s *NATSCommunityStorage) SaveCommunity(ctx context.Context, community *Community) error

SaveCommunity persists a community to NATS KV

type NATSSummaryStore 

type NATSSummaryStore struct {
	// contains filtered or unexported fields
}

NATSSummaryStore implements SummaryStore over a NATS KV bucket.

func NewNATSSummaryStore 

func NewNATSSummaryStore(kv jetstream.KeyValue) *NATSSummaryStore

NewNATSSummaryStore creates a NATS-backed summary store over the given bucket.

func (*NATSSummaryStore) CountSummaries 

func (s *NATSSummaryStore) CountSummaries(ctx context.Context) (int, error)

CountSummaries returns the number of stored summary records.

func (*NATSSummaryStore) GetSummary 

func (s *NATSSummaryStore) GetSummary(ctx context.Context, level int, membershipHash string) (*CommunitySummaryRecord, error)

GetSummary returns the stored record for {level}.{hash}, or (nil, nil) on miss.

func (*NATSSummaryStore) PutFailedUnlessEnhanced 

func (s *NATSSummaryStore) PutFailedUnlessEnhanced(ctx context.Context, rec *CommunitySummaryRecord) error

PutFailedUnlessEnhanced writes rec (an llm-failed record) unless an llm-enhanced record already occupies its {level}.{hash} key. See the SummaryStore interface doc for the full race-safety argument. It uses revision CAS so a concurrent llm-enhanced success can never be downgraded to llm-failed.

func (*NATSSummaryStore) PutSummary 

func (s *NATSSummaryStore) PutSummary(ctx context.Context, rec *CommunitySummaryRecord) error

PutSummary writes the record for its {level}.{hash} key.

type PageRankConfig 

type PageRankConfig struct {
	// Iterations is the number of iterations to run (default: 20)
	Iterations int

	// DampingFactor is the probability of continuing the random walk (default: 0.85)
	DampingFactor float64

	// Tolerance is the convergence threshold (default: 1e-6)
	Tolerance float64

	// TopN is the number of top-ranked nodes to return (0 = all)
	TopN int
}

PageRankConfig holds configuration for PageRank computation

func DefaultPageRankConfig 

func DefaultPageRankConfig() PageRankConfig

DefaultPageRankConfig returns the standard PageRank configuration

type PageRankResult 

type PageRankResult struct {
	// Scores maps entity ID to PageRank score
	Scores map[string]float64

	// Ranked contains entity IDs sorted by PageRank score (descending)
	Ranked []string

	// Iterations is the actual number of iterations run
	Iterations int

	// Converged indicates whether the algorithm converged before max iterations
	Converged bool
}

PageRankResult holds the results of PageRank computation

func ComputePageRank 

func ComputePageRank(ctx context.Context, provider Provider, config PageRankConfig) (*PageRankResult, error)

ComputePageRank computes PageRank scores for all nodes in the graph

func ComputePageRankForCommunity 

func ComputePageRankForCommunity(ctx context.Context, provider Provider, communityMembers []string, config PageRankConfig) (*PageRankResult, error)

ComputePageRankForCommunity computes PageRank for entities within a community This is more efficient than full graph PageRank for large graphs

type PredicateProvider 

type PredicateProvider struct {
	// contains filtered or unexported fields
}

PredicateProvider implements Provider for entities matching a predicate This is more practical for real-world use: cluster entities of specific types

func NewPredicateProvider 

func NewPredicateProvider(qm RelationshipQuerier, predicate string) *PredicateProvider

NewPredicateProvider creates a Provider for entities matching a predicate It caches the valid entity set at construction time for performance

func (*PredicateProvider) GetAllEntityIDs 

func (p *PredicateProvider) GetAllEntityIDs(ctx context.Context) ([]string, error)

GetAllEntityIDs returns all entities matching the predicate

func (*PredicateProvider) GetEdgeWeight 

func (p *PredicateProvider) GetEdgeWeight(ctx context.Context, fromID, toID string) (float64, error)

GetEdgeWeight returns the weight of an edge (unweighted: 1.0 or 0.0)

func (*PredicateProvider) GetNeighbors 

func (p *PredicateProvider) GetNeighbors(ctx context.Context, entityID string, direction string) ([]string, error)

GetNeighbors returns entity IDs connected to the given entity

type ProgressiveSummarizer 

type ProgressiveSummarizer struct {
	// contains filtered or unexported fields
}

ProgressiveSummarizer provides progressive enhancement - statistical summary immediately, LLM enhancement asynchronously via events

func NewProgressiveSummarizer 

func NewProgressiveSummarizer() *ProgressiveSummarizer

NewProgressiveSummarizer creates a progressive summarizer with default settings

func (*ProgressiveSummarizer) BuildCorpusDF 

func (s *ProgressiveSummarizer) BuildCorpusDF(entities []*gtypes.EntityState)

BuildCorpusDF forwards to the wrapped statistical summarizer.

func (*ProgressiveSummarizer) SummarizeCommunity 

func (s *ProgressiveSummarizer) SummarizeCommunity(
	ctx context.Context,
	community *Community,
	entities []*gtypes.EntityState,
) (*Community, error)

SummarizeCommunity generates an immediate statistical summary Caller is responsible for saving and publishing community.detected event for async LLM enhancement

type Provider 

type Provider = gtypes.Provider

Provider is an alias to the shared interface in graph package. Abstracts the graph data source for community detection.

type QueryManagerProvider 

type QueryManagerProvider struct {
	// contains filtered or unexported fields
}

QueryManagerProvider implements Provider using QueryManager

func NewQueryManagerProvider 

func NewQueryManagerProvider(qm RelationshipQuerier) *QueryManagerProvider

NewQueryManagerProvider creates a Provider backed by QueryManager

func (*QueryManagerProvider) GetAllEntityIDs 

func (p *QueryManagerProvider) GetAllEntityIDs(_ context.Context) ([]string, error)

GetAllEntityIDs returns all entity IDs in the graph

func (*QueryManagerProvider) GetEdgeWeight 

func (p *QueryManagerProvider) GetEdgeWeight(ctx context.Context, fromID, toID string) (float64, error)

GetEdgeWeight returns the weight of an edge between two entities

func (*QueryManagerProvider) GetNeighbors 

func (p *QueryManagerProvider) GetNeighbors(ctx context.Context, entityID string, direction string) ([]string, error)

GetNeighbors returns entity IDs connected to the given entity

type Relationship 

type Relationship struct {
	Subject      string
	Predicate    string
	Object       interface{}
	FromEntityID string  // Source entity
	ToEntityID   string  // Target entity
	Weight       float64 // For weighted edges
}

Relationship represents an edge between entities. Local copy to avoid import cycle with querymanager.

type RelationshipQuerier 

type RelationshipQuerier interface {
	EntityQuerier
	QueryRelationships(ctx context.Context, entityID string, direction Direction) ([]*Relationship, error)
	QueryByPredicate(ctx context.Context, predicate string) ([]string, error)
}

RelationshipQuerier provides minimal interface for querying relationships. This interface exists to avoid import cycle with querymanager package.

type SemanticEdgeMetrics 

type SemanticEdgeMetrics interface {
	// ObserveBuildMs records one refresh's wall-clock duration in milliseconds.
	// Observed on EVERY refresh that reaches the query loop, including near-zero
	// reuse cycles, so the distribution shows how often an actual rebuild happens
	// versus a cheap all-reused pass.
	ObserveBuildMs(ms float64)
	// AddSimilarQueries adds the number of FindSimilar calls a refresh issued
	// (0 on a fully-reused, unchanged-corpus cycle — the query load §7.1 bounds).
	AddSimilarQueries(n int)
}

SemanticEdgeMetrics records the cost of a mutual-kNN cache refresh (B2 §7.2). It is a narrow sink so this leaf package needs no prometheus import; the graph-clustering component satisfies it with a Prometheus-backed adapter over its shared registry. A nil recorder — the default, and every direct provider unit test — skips recording.

type SemanticEdgeParams 

type SemanticEdgeParams struct {
	// K is the mutual-kNN k (per-direction top-k candidate set size).
	K int
	// Threshold is the minimum similarity for a candidate to count.
	Threshold float64
}

SemanticEdgeParams holds the mutual-kNN tuning knobs for SemanticEdgeProvider.

type SemanticEdgeProvider 

type SemanticEdgeProvider struct {
	// contains filtered or unexported fields
}

SemanticEdgeProvider decorates an EntityIDProvider with mutual-kNN semantic virtual edges, so entities that are thematically related but structurally heterogeneous (different type, different system) can still land in the same LPA community. It is the third link in the detection provider chain: 

kvProvider (explicit edges)
  -> EntityIDProvider (sibling + system-peer virtual edges)
    -> SemanticEdgeProvider (mutual-kNN semantic virtual edges)

It is inserted only when the operator enables the semantic-edge tier; when disabled the chain is exactly the two providers it is today and behavior is byte-identical (B2 §1.4).

Semantic virtual edges are ephemeral — computed on demand from the similarity finder and never persisted — exactly like the EntityID virtual edges.

func NewSemanticEdgeProvider 

func NewSemanticEdgeProvider(
	base *EntityIDProvider,
	finder SemanticNeighborFinder,
	weights WeightConfig,
	params SemanticEdgeParams,
	logger *slog.Logger,
) *SemanticEdgeProvider

NewSemanticEdgeProvider wraps an EntityIDProvider with the semantic-edge tier. Zero-valued params and weights fall back to the documented starting values.

func (*SemanticEdgeProvider) AppliedSemanticEdges 

func (p *SemanticEdgeProvider) AppliedSemanticEdges() bool

AppliedSemanticEdges reports the ACTUAL completed mode of the cycle that just ran: true iff the tier was active AND the cycle actually SERVED semantic edges in the vote — it was neither producer-wide-aborted (abortedThisCycle) nor coverage-degraded to structural-only with nothing to serve (degradedThisCycle). It is read AFTER detection and is deliberately distinct from the preflight readiness INTENT SetActive stamps BEFORE it: an intended-active cycle that aborts/degrades mid-refresh serves structural-only, and the semantic_edges_applied signal must report what happened, not what was intended (B2 §4, Codex P1#2 + §7.3 seam close).

The four cases the two latches resolve (all gated on active):

  • producer-wide not-ready/fatal abort -> FALSE (abortedThisCycle). In the refreshable model this leaves the prior epoch's adjacency populated but UN-served (refreshCache returns the sentinel, so lookups degrade to structural), so a len(mutualNeighbors)>0 check would falsely report applied — the latch is the truth.
  • coverage-threshold abort that KEEPS a prior committed cache -> TRUE: the prior epoch's semantic edges are in the vote. degradedThisCycle keys on whether a prior refresh committed (cacheInitialized), NOT on emptiness — an initialized-but-empty prior is still "ran, found nothing" (applied), so it correctly reports TRUE too (neither latch set).
  • coverage-threshold abort with NO prior cache (first cold cycle / >threshold burst) -> FALSE (degradedThisCycle): the cycle served structural-only. This is the seam the §7.3 mechanism otherwise left reporting a misleading 1.
  • a healthy build that committed ZERO mutual pairs -> TRUE: the tier RAN and its (empty) semantic edges were in the vote — the #618 "semantics ran, found nothing" distinction the semantic_edges_applied metric exists to preserve (a completed commit clears degradedThisCycle).

Both latches are lock-free atomics reset per cycle by SetActive, so AppliedSemanticEdges stays a lock-free read (a concurrent B3 enhancement worker can call it safely).

func (*SemanticEdgeProvider) BeginCycle 

func (p *SemanticEdgeProvider) BeginCycle(embeddingRevision uint64, active bool)

BeginCycle performs the ENTIRE per-cycle transition in one synchronized step (B2 §7.1, Codex P2#4): it records the coarse embedding-index watermark this cycle's refresh keys off, sets the tier active/inactive from the readiness verdict, resets the per-cycle abort/degrade latches, and LAST advances the cycle epoch so refreshCache does exactly one refresh pass per cycle (reusing unchanged directed sets and re-querying only what the watermark cannot vouch for). The component calls it from applySemanticGate before detection reads the provider.

The store ORDER is load-bearing and must not be reshuffled: the epoch bump is the LAST atomic write, and every reader loads the epoch BEFORE the active flag / abort latches (GetNeighbors gates on active then refreshCache reads epoch-then-latches). Go's sync/atomic operations are sequentially consistent, so a reader that observes the NEW epoch is guaranteed to observe the reset latches + new active/revision too — it can never see a torn (new-epoch, prior-cycle-verdict/stale-latch) state. This closes the window the previous two-call BeginCycle-then-SetActive sequence left, where B3's concurrent enhancement-worker reader could observe the new epoch under the prior cycle's active verdict or an un-reset abort latch. Race-safe via atomics: the detector loop is the sole writer; B3's worker is a read-only consumer.

func (*SemanticEdgeProvider) ClearCache 

func (p *SemanticEdgeProvider) ClearCache()

ClearCache resets the mutual-kNN adjacency and propagates the clear to the wrapped provider, mirroring EntityIDProvider.ClearCache.

func (*SemanticEdgeProvider) GetAllEntityIDs 

func (p *SemanticEdgeProvider) GetAllEntityIDs(ctx context.Context) ([]string, error)

GetAllEntityIDs delegates to the wrapped provider — the semantic tier adds edges, never entities.

func (*SemanticEdgeProvider) GetEdgeWeight 

func (p *SemanticEdgeProvider) GetEdgeWeight(ctx context.Context, fromID, toID string) (float64, error)

GetEdgeWeight resolves the edge weight across all four tiers in one place (B2 §2). Explicit edges (from the underlying base provider) are strictly dominant; otherwise the weight is the max across the sibling, system-peer, and semantic tiers the pair qualifies under — never a sum. This deliberately does NOT delegate to EntityIDProvider.GetEdgeWeight, whose first-match cascade would collapse tier identity before the max could be computed.

func (*SemanticEdgeProvider) GetNeighbors 

func (p *SemanticEdgeProvider) GetNeighbors(ctx context.Context, entityID string, direction string) ([]string, error)

GetNeighbors returns the wrapped provider's neighbors (explicit + sibling + system-peer) plus this entity's mutual-kNN semantic neighbors, deduplicated. Semantic relationships are symmetric, so direction is respected only for the wrapped (explicit) edges, exactly as EntityIDProvider treats sibling edges.

The semantic tier is strictly additive: if the mutual-kNN cache cannot be built (e.g. the embedding index is cold and the finder returns nothing), the structural neighbor set is returned unchanged rather than failing the cycle.

func (*SemanticEdgeProvider) IsActive 

func (p *SemanticEdgeProvider) IsActive() bool

IsActive reports the current per-cycle tier state (observability / tests).

func (*SemanticEdgeProvider) ResetEdgeCache 

func (p *SemanticEdgeProvider) ResetEdgeCache()

ResetEdgeCache forwards the per-cycle explicit-edge cache reset down the provider chain (gh#666). The mutual-kNN cache has its own per-cycle lifetime (BeginCycle/refreshCache), so this only forwards to the wrapped EntityIDProvider.

func (*SemanticEdgeProvider) SetActive 

func (p *SemanticEdgeProvider) SetActive(active bool)

SetActive enables or disables the semantic tier WITHOUT advancing the cycle epoch — the standalone active toggle (B2 §4). The per-cycle production transition is BeginCycle's job now (Codex P2#4), which sets active atomically WITH the epoch + latch reset; SetActive is the narrow "flip active in place" primitive the §4 activation tests drive to exercise the structural-only path in isolation, and the shared-provider race test uses to model an out-of-band toggle. When inactive, GetNeighbors and GetEdgeWeight behave exactly as the wrapped EntityIDProvider and never trigger a mutual-kNN refresh, so the cache is only ever built from a ready embedding index (the no-not-ready-latch guarantee). Concurrency-safe via the atomic flag.

func (*SemanticEdgeProvider) WithMetrics 

func (p *SemanticEdgeProvider) WithMetrics(m SemanticEdgeMetrics) *SemanticEdgeProvider

WithMetrics attaches the refresh-cost recorder (B2 §7.2) and returns the provider for chaining. Call once at construction, before the provider is shared with any goroutine; a nil recorder leaves recording off.

type SemanticNeighborFinder 

type SemanticNeighborFinder interface {
	SimilarNeighbors(ctx context.Context, entityID string, threshold float64, limit int) ([]string, error)
}

SemanticNeighborFinder returns the entity IDs semantically similar to entityID at or above threshold, capped at limit (the directed top-k set for mutual-kNN synthesis). It is deliberately a narrow, clustering-local interface so this package does not import graph/inference: the production wire satisfies it with a thin adapter over the component's existing graph.embedding.query.similar finder (no second similarity RPC — B2 §1.2).

The adapter maps the embedding service's error classes onto the two package sentinels above: ErrSemanticIndexNotReady (whole index cold — abort the refresh) and ErrSemanticQueryTransient (this one query timed out — count it toward the coverage threshold). A genuine miss ("no embedding yet") and a genuine empty result both return `(nil, nil)`.

type StatisticalSummarizer 

type StatisticalSummarizer struct {
	// MaxKeywords limits the number of keywords extracted
	MaxKeywords int

	// MaxRepEntities limits the number of representative entities
	MaxRepEntities int
	// contains filtered or unexported fields
}

StatisticalSummarizer implements CommunitySummarizer using statistical methods This is the default summarizer that doesn't require external LLM services

func NewStatisticalSummarizer 

func NewStatisticalSummarizer() *StatisticalSummarizer

NewStatisticalSummarizer creates a statistical summarizer with default settings

func (*StatisticalSummarizer) BuildCorpusDF 

func (s *StatisticalSummarizer) BuildCorpusDF(entities []*gtypes.EntityState)

BuildCorpusDF populates the corpus-wide document-frequency map from the given entity slice. DF[T] counts entities containing term T at least once (set semantics, not occurrence count). Term extraction matches extractKeywords via the shared termOccurrencesForEntity helper. Empty input clears the map. Safe to call multiple times.

func (*StatisticalSummarizer) SummarizeCommunity 

func (s *StatisticalSummarizer) SummarizeCommunity(
	ctx context.Context,
	community *Community,
	entities []*gtypes.EntityState,
) (*Community, error)

SummarizeCommunity generates a statistical summary of the community

type SummaryStore 

type SummaryStore interface {
	// GetSummary returns the summary record for a membership hash at a level, or
	// (nil, nil) when no record exists.
	GetSummary(ctx context.Context, level int, membershipHash string) (*CommunitySummaryRecord, error)

	// PutSummary writes (or overwrites) the record for its {level}.{hash} key. It is
	// the SUCCESS lane: a same-membership double-write is idempotent by construction
	// (content-addressed key), and a success unconditionally overwrites a prior
	// llm-failed record (desired recovery) or an equivalent prior success. It is NOT
	// used for llm-failed records — those go through PutFailedUnlessEnhanced so a
	// failure can never clobber a success.
	PutSummary(ctx context.Context, rec *CommunitySummaryRecord) error

	// PutFailedUnlessEnhanced writes rec (an llm-failed record) for its
	// {level}.{hash} key, but NEVER overwrites an existing llm-enhanced record — a
	// successful summary always wins over a later failure (the ADR-087 idempotency
	// guarantee). It is race-safe against a concurrent success: a plain
	// read-then-write is TOCTOU (a concurrent llm-enhanced Put can land between the
	// read and the write), so the implementation reads the record and its revision,
	// then writes conditionally (Create on a miss, revision-checked Update on an
	// existing failed record). A concurrent success occupies the key / changes the
	// revision, the conditional write conflicts, and the retry re-reads, sees the
	// llm-enhanced record, and skips. A skip is success (nil), not an error.
	PutFailedUnlessEnhanced(ctx context.Context, rec *CommunitySummaryRecord) error

	// CountSummaries returns the number of stored summary records (for the
	// bucket-size gauge / future bounded-GC decision).
	CountSummaries(ctx context.Context) (int, error)
}

SummaryStore abstracts persistence for community LLM summaries keyed {level}.{membership_hash}. It is worker-exclusive: the enhancement worker is the SOLE writer (single-writer invariant, ADR-087). Read-only consumers (the graph-query community cache) watch the bucket directly rather than through this interface.

type WeightConfig 

type WeightConfig struct {
	// SiblingWeight is the resolved weight for an EntityID sibling edge (same
	// 5-part type prefix). Starting value 0.7 (unchanged from today).
	SiblingWeight float64
	// SystemPeerWeight is the resolved weight for an EntityID system-peer edge
	// (same system segment). Starting value 0.2 in the semantic-enabled profile
	// (0.3 today), so the total structural vote mass stays competitive once the
	// semantic tier is added rather than growing.
	SystemPeerWeight float64
	// SemanticWeight is the resolved weight for a mutual-kNN semantic edge.
	// Starting value 0.9.
	SemanticWeight float64
}

WeightConfig is the single resolved home for every edge-tier weight used by SemanticEdgeProvider.GetEdgeWeight. Its resolve method (B2 §2, "one testable place") replaces EntityIDProvider's first-match cascade, which cannot be reused unmodified once a fourth (semantic) tier joins the vote: the cascade silently drops tier identity, so a pair that is both a sibling (0.7) and a mutual-kNN semantic match (0.9) would resolve to whichever tier the cascade checked first rather than to the max of the two.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.