fix: reorganize struct fields and config parameters for consistency

- [x] Reorder Config struct fields alphabetically and by related functionality
- [x] Reorganize Observation model fields with archival fields grouped together
- [x] Reorder ObservationStore fields to group related members
- [x] Reorder Store struct fields with health check caching grouped
- [x] Reorganize HealthInfo and PoolMetrics struct field order
- [x] Reorder maintenance Service struct fields logically
- [x] Reorganize MCP server handler parameter structs alphabetically
- [x] Reorder pattern detector candidate tracking fields
- [x] Reorganize search Manager struct fields by functionality
- [x] Reorder vector Client struct fields with mutex protections grouped
- [x] Reorganize handler request/response struct fields
- [x] Update handlers_test.go to expect wrapped response format
- [x] Reorder middleware TokenAuth and rate limiter fields
- [x] Reorganize Service struct fields with grouped functionality
- [x] Fix RateLimiter field ordering for clarity
- [x] Reorder CircuitBreaker metrics fields
This commit is contained in:
2026-01-11 01:22:30 +00:00
parent c585ffd773
commit a270290023
18 changed files with 312 additions and 375 deletions
+10 -12
View File
@@ -46,34 +46,32 @@ type Config struct {
VectorStorageStrategy string `json:"vector_storage_strategy"` VectorStorageStrategy string `json:"vector_storage_strategy"`
ContextObsConcepts []string `json:"context_obs_concepts"` ContextObsConcepts []string `json:"context_obs_concepts"`
ContextObsTypes []string `json:"context_obs_types"` ContextObsTypes []string `json:"context_obs_types"`
ContextMaxPromptResults int `json:"context_max_prompt_results"` ContextFullCount int `json:"context_full_count"`
RerankingResults int `json:"reranking_results"` GraphBranchFactor int `json:"graph_branch_factor"`
GraphEdgeWeight float64 `json:"graph_edge_weight"` GraphEdgeWeight float64 `json:"graph_edge_weight"`
ContextRelevanceThreshold float64 `json:"context_relevance_threshold"` ContextRelevanceThreshold float64 `json:"context_relevance_threshold"`
RerankingCandidates int `json:"reranking_candidates"` RerankingCandidates int `json:"reranking_candidates"`
WorkerPort int `json:"worker_port"` WorkerPort int `json:"worker_port"`
RerankingMinImprovement float64 `json:"reranking_min_improvement"` RerankingMinImprovement float64 `json:"reranking_min_improvement"`
ContextObservations int `json:"context_observations"` ContextObservations int `json:"context_observations"`
ContextFullCount int `json:"context_full_count"` ContextMaxPromptResults int `json:"context_max_prompt_results"`
ContextSessionCount int `json:"context_session_count"` ContextSessionCount int `json:"context_session_count"`
MaxConns int `json:"max_conns"` MaxConns int `json:"max_conns"`
RerankingAlpha float64 `json:"reranking_alpha"` RerankingAlpha float64 `json:"reranking_alpha"`
GraphMaxHops int `json:"graph_max_hops"` GraphMaxHops int `json:"graph_max_hops"`
GraphBranchFactor int `json:"graph_branch_factor"` RerankingResults int `json:"reranking_results"`
GraphRebuildIntervalMin int `json:"graph_rebuild_interval_min"` GraphRebuildIntervalMin int `json:"graph_rebuild_interval_min"`
HubThreshold int `json:"hub_threshold"` HubThreshold int `json:"hub_threshold"`
ContextShowLastSummary bool `json:"context_show_last_summary"` ObservationRetentionDays int `json:"observation_retention_days"`
RerankingEnabled bool `json:"reranking_enabled"` MaintenanceIntervalHours int `json:"maintenance_interval_hours"`
ContextShowWorkTokens bool `json:"context_show_work_tokens"` ContextShowWorkTokens bool `json:"context_show_work_tokens"`
ContextShowReadTokens bool `json:"context_show_read_tokens"` ContextShowReadTokens bool `json:"context_show_read_tokens"`
RerankingPureMode bool `json:"reranking_pure_mode"` RerankingPureMode bool `json:"reranking_pure_mode"`
GraphEnabled bool `json:"graph_enabled"` GraphEnabled bool `json:"graph_enabled"`
MaintenanceEnabled bool `json:"maintenance_enabled"`
// Maintenance settings RerankingEnabled bool `json:"reranking_enabled"`
MaintenanceEnabled bool `json:"maintenance_enabled"` // Enable scheduled maintenance ContextShowLastSummary bool `json:"context_show_last_summary"`
MaintenanceIntervalHours int `json:"maintenance_interval_hours"` // How often to run maintenance (default 6 hours) CleanupStaleObservations bool `json:"cleanup_stale_observations"`
ObservationRetentionDays int `json:"observation_retention_days"` // Delete observations older than N days (0 = no age-based deletion)
CleanupStaleObservations bool `json:"cleanup_stale_observations"` // Auto-cleanup stale observations during maintenance
} }
var ( var (
+6 -8
View File
@@ -53,16 +53,18 @@ type Observation struct {
Scope models.ObservationScope `gorm:"type:text;default:'project';check:scope IN ('project', 'global');index:idx_observations_scope;index:idx_observations_project_scope,priority:2"` Scope models.ObservationScope `gorm:"type:text;default:'project';check:scope IN ('project', 'global');index:idx_observations_scope;index:idx_observations_project_scope,priority:2"`
Type models.ObservationType `gorm:"type:text;check:type IN ('decision', 'bugfix', 'feature', 'refactor', 'discovery', 'change');index;not null"` Type models.ObservationType `gorm:"type:text;check:type IN ('decision', 'bugfix', 'feature', 'refactor', 'discovery', 'change');index;not null"`
CreatedAt string `gorm:"not null"` CreatedAt string `gorm:"not null"`
Title sql.NullString `gorm:"type:text"` Facts models.JSONStringArray `gorm:"type:text"`
Narrative sql.NullString `gorm:"type:text"` Narrative sql.NullString `gorm:"type:text"`
Concepts models.JSONStringArray `gorm:"type:text"` Concepts models.JSONStringArray `gorm:"type:text"`
FilesRead models.JSONStringArray `gorm:"type:text"` FilesRead models.JSONStringArray `gorm:"type:text"`
FilesModified models.JSONStringArray `gorm:"type:text"` FilesModified models.JSONStringArray `gorm:"type:text"`
Subtitle sql.NullString `gorm:"type:text"` Subtitle sql.NullString `gorm:"type:text"`
Facts models.JSONStringArray `gorm:"type:text"` Title sql.NullString `gorm:"type:text"`
LastRetrievedAt sql.NullInt64 `gorm:"column:last_retrieved_at_epoch"` ArchivedReason sql.NullString
PromptNumber sql.NullInt64
ScoreUpdatedAt sql.NullInt64 `gorm:"column:score_updated_at_epoch;index:idx_observations_score_updated"` ScoreUpdatedAt sql.NullInt64 `gorm:"column:score_updated_at_epoch;index:idx_observations_score_updated"`
PromptNumber sql.NullInt64
ArchivedAt sql.NullInt64 `gorm:"column:archived_at_epoch"`
LastRetrievedAt sql.NullInt64 `gorm:"column:last_retrieved_at_epoch"`
ID int64 `gorm:"primaryKey;autoIncrement"` ID int64 `gorm:"primaryKey;autoIncrement"`
ImportanceScore float64 `gorm:"type:real;default:1.0;index:idx_observations_importance,priority:1,sort:desc"` ImportanceScore float64 `gorm:"type:real;default:1.0;index:idx_observations_importance,priority:1,sort:desc"`
UserFeedback int `gorm:"default:0"` UserFeedback int `gorm:"default:0"`
@@ -70,11 +72,7 @@ type Observation struct {
CreatedAtEpoch int64 `gorm:"index:idx_observations_created,sort:desc;index:idx_observations_project_created,priority:2,sort:desc;not null"` CreatedAtEpoch int64 `gorm:"index:idx_observations_created,sort:desc;index:idx_observations_project_created,priority:2,sort:desc;not null"`
DiscoveryTokens int64 `gorm:"default:0"` DiscoveryTokens int64 `gorm:"default:0"`
IsSuperseded int `gorm:"default:0;index:idx_observations_superseded;index:idx_observations_active,priority:2"` IsSuperseded int `gorm:"default:0;index:idx_observations_superseded;index:idx_observations_active,priority:2"`
// Archival fields
IsArchived int `gorm:"default:0;index:idx_observations_archived;index:idx_observations_active,priority:1"` IsArchived int `gorm:"default:0;index:idx_observations_archived;index:idx_observations_active,priority:1"`
ArchivedAt sql.NullInt64 `gorm:"column:archived_at_epoch"`
ArchivedReason sql.NullString
} }
func (Observation) TableName() string { return "observations" } func (Observation) TableName() string { return "observations" }
+6 -8
View File
@@ -41,18 +41,16 @@ type CleanupFunc func(ctx context.Context, deletedIDs []int64)
// ObservationStore provides observation-related database operations using GORM. // ObservationStore provides observation-related database operations using GORM.
type ObservationStore struct { type ObservationStore struct {
conflictStore any
relationStore any
db *gorm.DB db *gorm.DB
rawDB *sql.DB rawDB *sql.DB
cleanupFunc CleanupFunc cleanupFunc CleanupFunc
conflictStore any // Placeholder for ConflictStore (Phase 4)
relationStore any // Placeholder for RelationStore (Phase 4)
// Cleanup queue for async observation cleanup with proper error handling
cleanupQueue chan string cleanupQueue chan string
stopCleanup chan struct{}
cleanupWg sync.WaitGroup cleanupWg sync.WaitGroup
cleanupOnce sync.Once cleanupOnce sync.Once
cleanupStarted atomic.Bool // tracks if cleanup worker was started cleanupStarted atomic.Bool
stopCleanup chan struct{}
} }
// NewObservationStore creates a new observation store. // NewObservationStore creates a new observation store.
@@ -854,10 +852,10 @@ func (s *ObservationStore) GetArchivalStats(ctx context.Context, project string)
// Use a single query with conditional aggregation to get all stats at once. // Use a single query with conditional aggregation to get all stats at once.
// This is much faster than 4 separate queries (saves 3 round trips). // This is much faster than 4 separate queries (saves 3 round trips).
type statsResult struct { type statsResult struct {
OldestEpoch *int64
NewestEpoch *int64
TotalCount int64 TotalCount int64
ArchivedCount int64 ArchivedCount int64
OldestEpoch *int64 // Pointer to handle NULL
NewestEpoch *int64
} }
var result statsResult var result statsResult
+20 -22
View File
@@ -19,15 +19,13 @@ import (
// Store represents the GORM database connection with sqlite-vec support. // Store represents the GORM database connection with sqlite-vec support.
type Store struct { type Store struct {
DB *gorm.DB
sqlDB *sql.DB // For FTS5 and sqlite-vec operations that require raw SQL
metrics *PoolMetrics
// Health check caching to reduce database load from frequent monitoring
healthCacheMu sync.RWMutex
cachedHealth *HealthInfo
healthCacheTime time.Time healthCacheTime time.Time
healthCacheTTL time.Duration // Default: 5 seconds DB *gorm.DB
sqlDB *sql.DB
metrics *PoolMetrics
cachedHealth *HealthInfo
healthCacheTTL time.Duration
healthCacheMu sync.RWMutex
} }
// Config holds database configuration. // Config holds database configuration.
@@ -313,13 +311,13 @@ func (s *Store) performHealthCheck(ctx context.Context) *HealthInfo {
// HealthInfo contains database health check results. // HealthInfo contains database health check results.
type HealthInfo struct { type HealthInfo struct {
Status string `json:"status"` // healthy, degraded, unhealthy
Timestamp time.Time `json:"timestamp"` Timestamp time.Time `json:"timestamp"`
QueryLatency time.Duration `json:"query_latency_ns"` Status string `json:"status"`
PoolStats PoolStats `json:"pool_stats"`
HistoricalMetrics MetricsSummary `json:"historical_metrics,omitempty"`
Error string `json:"error,omitempty"` Error string `json:"error,omitempty"`
Warning string `json:"warning,omitempty"` Warning string `json:"warning,omitempty"`
HistoricalMetrics MetricsSummary `json:"historical_metrics,omitempty"`
PoolStats PoolStats `json:"pool_stats"`
QueryLatency time.Duration `json:"query_latency_ns"`
} }
// PoolStats contains connection pool statistics. // PoolStats contains connection pool statistics.
@@ -345,16 +343,16 @@ const (
// PoolMetrics tracks historical connection pool metrics with a sliding window. // PoolMetrics tracks historical connection pool metrics with a sliding window.
type PoolMetrics struct { type PoolMetrics struct {
lastSampleTime time.Time
latencySamples []time.Duration
latencyIdx int
latencyCount int
totalQueries int64
totalWaitTime time.Duration
peakInUse int
peakWaitCount int64
windowSize int
mu sync.RWMutex mu sync.RWMutex
latencySamples []time.Duration // Circular buffer of latency samples
latencyIdx int // Current index in circular buffer
latencyCount int // Number of samples collected
totalQueries int64 // Total queries executed
totalWaitTime time.Duration // Cumulative wait time for connections
peakInUse int // Peak concurrent connections in use
peakWaitCount int64 // Peak wait count observed
lastSampleTime time.Time // Last time a sample was recorded
windowSize int // Size of sliding window
} }
// NewPoolMetrics creates a new pool metrics collector with the given window size. // NewPoolMetrics creates a new pool metrics collector with the given window size.
@@ -450,6 +448,7 @@ func (m *PoolMetrics) GetMetricsSummary() MetricsSummary {
// MetricsSummary contains aggregated pool metrics. // MetricsSummary contains aggregated pool metrics.
type MetricsSummary struct { type MetricsSummary struct {
LastSampleTime time.Time `json:"last_sample_time"`
TotalQueries int64 `json:"total_queries"` TotalQueries int64 `json:"total_queries"`
SampleCount int `json:"sample_count"` SampleCount int `json:"sample_count"`
AvgLatency time.Duration `json:"avg_latency_ns"` AvgLatency time.Duration `json:"avg_latency_ns"`
@@ -459,7 +458,6 @@ type MetricsSummary struct {
PeakInUse int `json:"peak_in_use"` PeakInUse int `json:"peak_in_use"`
PeakWaitCount int64 `json:"peak_wait_count"` PeakWaitCount int64 `json:"peak_wait_count"`
TotalWaitTime time.Duration `json:"total_wait_time_ns"` TotalWaitTime time.Duration `json:"total_wait_time_ns"`
LastSampleTime time.Time `json:"last_sample_time"`
} }
// GetMetrics returns the current metrics without performing a health check. // GetMetrics returns the current metrics without performing a health check.
+7 -9
View File
@@ -13,23 +13,21 @@ import (
// Service handles scheduled maintenance tasks. // Service handles scheduled maintenance tasks.
type Service struct { type Service struct {
store *gorm.Store log zerolog.Logger
observationStore *gorm.ObservationStore lastRunTime time.Time
summaryStore *gorm.SummaryStore
promptStore *gorm.PromptStore promptStore *gorm.PromptStore
store *gorm.Store
vectorCleanupFn func(ctx context.Context, deletedIDs []int64) vectorCleanupFn func(ctx context.Context, deletedIDs []int64)
config *config.Config config *config.Config
log zerolog.Logger summaryStore *gorm.SummaryStore
stopCh chan struct{} stopCh chan struct{}
doneCh chan struct{} doneCh chan struct{}
mu sync.Mutex observationStore *gorm.ObservationStore
running bool
// Metrics
lastRunTime time.Time
lastRunDuration time.Duration lastRunDuration time.Duration
totalCleanedObs int64 totalCleanedObs int64
totalOptimizeRun int64 totalOptimizeRun int64
mu sync.Mutex
running bool
} }
// NewService creates a new maintenance service. // NewService creates a new maintenance service.
+26 -26
View File
@@ -1585,15 +1585,15 @@ func (s *Server) handleGetObservation(ctx context.Context, args json.RawMessage)
// handleEditObservation updates an existing observation with provided fields. // handleEditObservation updates an existing observation with provided fields.
func (s *Server) handleEditObservation(ctx context.Context, args json.RawMessage) (string, error) { func (s *Server) handleEditObservation(ctx context.Context, args json.RawMessage) (string, error) {
var params struct { var params struct {
ID int64 `json:"id"`
Title *string `json:"title,omitempty"` Title *string `json:"title,omitempty"`
Subtitle *string `json:"subtitle,omitempty"` Subtitle *string `json:"subtitle,omitempty"`
Narrative *string `json:"narrative,omitempty"` Narrative *string `json:"narrative,omitempty"`
Scope *string `json:"scope,omitempty"`
Facts []string `json:"facts,omitempty"` Facts []string `json:"facts,omitempty"`
Concepts []string `json:"concepts,omitempty"` Concepts []string `json:"concepts,omitempty"`
FilesRead []string `json:"files_read,omitempty"` FilesRead []string `json:"files_read,omitempty"`
FilesModified []string `json:"files_modified,omitempty"` FilesModified []string `json:"files_modified,omitempty"`
Scope *string `json:"scope,omitempty"` ID int64 `json:"id"`
} }
if err := json.Unmarshal(args, &params); err != nil { if err := json.Unmarshal(args, &params); err != nil {
return "", fmt.Errorf("invalid arguments: %w", err) return "", fmt.Errorf("invalid arguments: %w", err)
@@ -1808,9 +1808,9 @@ func (s *Server) handleSuggestConsolidations(ctx context.Context, args json.RawM
// Find similar pairs using vector search if available // Find similar pairs using vector search if available
type consolidationGroup struct { type consolidationGroup struct {
Primary *models.Observation `json:"primary"` Primary *models.Observation `json:"primary"`
Reason string `json:"reason"`
Similar []*models.Observation `json:"similar"` Similar []*models.Observation `json:"similar"`
Similarity float64 `json:"avg_similarity"` Similarity float64 `json:"avg_similarity"`
Reason string `json:"reason"`
} }
groups := []consolidationGroup{} groups := []consolidationGroup{}
@@ -1920,9 +1920,9 @@ func (s *Server) handleSuggestConsolidations(ctx context.Context, args json.RawM
// handleTagObservation adds, removes, or sets tags on an observation. // handleTagObservation adds, removes, or sets tags on an observation.
func (s *Server) handleTagObservation(ctx context.Context, args json.RawMessage) (string, error) { func (s *Server) handleTagObservation(ctx context.Context, args json.RawMessage) (string, error) {
var params struct { var params struct {
ID int64 `json:"id"`
Tags []string `json:"tags"`
Mode string `json:"mode"` Mode string `json:"mode"`
Tags []string `json:"tags"`
ID int64 `json:"id"`
} }
params.Mode = "add" // default params.Mode = "add" // default
@@ -2073,8 +2073,8 @@ func (s *Server) handleGetObservationsByTag(ctx context.Context, args json.RawMe
func (s *Server) handleGetTemporalTrends(ctx context.Context, args json.RawMessage) (string, error) { func (s *Server) handleGetTemporalTrends(ctx context.Context, args json.RawMessage) (string, error) {
var params struct { var params struct {
Project string `json:"project"` Project string `json:"project"`
Days int `json:"days"`
GroupBy string `json:"group_by"` GroupBy string `json:"group_by"`
Days int `json:"days"`
} }
params.Days = 30 params.Days = 30
params.GroupBy = "day" params.GroupBy = "day"
@@ -2347,10 +2347,10 @@ func (s *Server) handleGetDataQualityReport(ctx context.Context, args json.RawMe
func (s *Server) handleBatchTagByPattern(ctx context.Context, args json.RawMessage) (string, error) { func (s *Server) handleBatchTagByPattern(ctx context.Context, args json.RawMessage) (string, error) {
var params struct { var params struct {
Pattern string `json:"pattern"` Pattern string `json:"pattern"`
Tags []string `json:"tags"`
Project string `json:"project"` Project string `json:"project"`
DryRun bool `json:"dry_run"` Tags []string `json:"tags"`
MaxMatches int `json:"max_matches"` MaxMatches int `json:"max_matches"`
DryRun bool `json:"dry_run"`
} }
params.DryRun = true params.DryRun = true
params.MaxMatches = 100 params.MaxMatches = 100
@@ -2485,14 +2485,14 @@ func (s *Server) handleExplainSearchRanking(ctx context.Context, args json.RawMe
// Build detailed explanations for each result // Build detailed explanations for each result
type RankExplanation struct { type RankExplanation struct {
Rank int `json:"rank"` ScoreBreakdown map[string]float64 `json:"score_breakdown"`
ID int64 `json:"id"` Metadata map[string]any `json:"metadata,omitempty"`
Title string `json:"title"` Title string `json:"title"`
Type string `json:"type"` Type string `json:"type"`
Score float64 `json:"score"`
ScoreBreakdown map[string]float64 `json:"score_breakdown"`
MatchedFields []string `json:"matched_fields"` MatchedFields []string `json:"matched_fields"`
Metadata map[string]any `json:"metadata,omitempty"` Rank int `json:"rank"`
ID int64 `json:"id"`
Score float64 `json:"score"`
} }
explanations := make([]RankExplanation, 0, len(result.Results)) explanations := make([]RankExplanation, 0, len(result.Results))
@@ -2562,10 +2562,10 @@ func (s *Server) handleExportObservations(ctx context.Context, args json.RawMess
var params struct { var params struct {
Format string `json:"format"` Format string `json:"format"`
Project string `json:"project"` Project string `json:"project"`
ObsType string `json:"obs_type"`
Limit int `json:"limit"` Limit int `json:"limit"`
DateStart int64 `json:"date_start"` DateStart int64 `json:"date_start"`
DateEnd int64 `json:"date_end"` DateEnd int64 `json:"date_end"`
ObsType string `json:"obs_type"`
} }
params.Format = "json" params.Format = "json"
params.Limit = 100 params.Limit = 100
@@ -2705,11 +2705,11 @@ func (s *Server) handleCheckSystemHealth(ctx context.Context) (string, error) {
} }
type HealthReport struct { type HealthReport struct {
OverallStatus string `json:"overall_status"`
HealthScore int `json:"health_score"`
Timestamp time.Time `json:"timestamp"` Timestamp time.Time `json:"timestamp"`
Subsystems map[string]*SubsystemHealth `json:"subsystems"` Subsystems map[string]*SubsystemHealth `json:"subsystems"`
OverallStatus string `json:"overall_status"`
Actions []string `json:"recommended_actions,omitempty"` Actions []string `json:"recommended_actions,omitempty"`
HealthScore int `json:"health_score"`
} }
report := &HealthReport{ report := &HealthReport{
@@ -2902,19 +2902,19 @@ func (s *Server) handleAnalyzeSearchPatterns(ctx context.Context, args json.RawM
type QueryPattern struct { type QueryPattern struct {
Query string `json:"query"` Query string `json:"query"`
LastUsed string `json:"last_used"`
Count int `json:"count"` Count int `json:"count"`
AvgResults float64 `json:"avg_results"` AvgResults float64 `json:"avg_results"`
ZeroResults int `json:"zero_result_count"` ZeroResults int `json:"zero_result_count"`
LastUsed string `json:"last_used"`
} }
type PatternAnalysis struct { type PatternAnalysis struct {
Period string `json:"period"` Period string `json:"period"`
TotalSearches int `json:"total_searches"`
UniqueQueries int `json:"unique_queries"`
TopQueries []QueryPattern `json:"top_queries"` TopQueries []QueryPattern `json:"top_queries"`
ZeroResultQueries []string `json:"zero_result_queries,omitempty"` ZeroResultQueries []string `json:"zero_result_queries,omitempty"`
Insights []string `json:"insights,omitempty"` Insights []string `json:"insights,omitempty"`
TotalSearches int `json:"total_searches"`
UniqueQueries int `json:"unique_queries"`
} }
analysis := &PatternAnalysis{ analysis := &PatternAnalysis{
@@ -3033,23 +3033,23 @@ func (s *Server) handleGetObservationRelationships(ctx context.Context, args jso
// Build response with additional context // Build response with additional context
type RelationInfo struct { type RelationInfo struct {
ID int64 `json:"id"`
SourceID int64 `json:"source_id"`
TargetID int64 `json:"target_id"`
Type string `json:"type"` Type string `json:"type"`
Confidence float64 `json:"confidence"`
SourceTitle string `json:"source_title,omitempty"` SourceTitle string `json:"source_title,omitempty"`
TargetTitle string `json:"target_title,omitempty"` TargetTitle string `json:"target_title,omitempty"`
SourceType string `json:"source_type,omitempty"` SourceType string `json:"source_type,omitempty"`
TargetType string `json:"target_type,omitempty"` TargetType string `json:"target_type,omitempty"`
ID int64 `json:"id"`
SourceID int64 `json:"source_id"`
TargetID int64 `json:"target_id"`
Confidence float64 `json:"confidence"`
} }
type GraphResponse struct { type GraphResponse struct {
Relations []RelationInfo `json:"relations"`
UniqueNodes []int64 `json:"unique_nodes"`
CenterID int64 `json:"center_id"` CenterID int64 `json:"center_id"`
MaxDepth int `json:"max_depth"` MaxDepth int `json:"max_depth"`
TotalRelations int `json:"total_relations"` TotalRelations int `json:"total_relations"`
Relations []RelationInfo `json:"relations"`
UniqueNodes []int64 `json:"unique_nodes"`
} }
// Collect unique node IDs // Collect unique node IDs
@@ -3162,10 +3162,10 @@ func (s *Server) handleGetObservationScoringBreakdown(ctx context.Context, args
// handleAnalyzeObservationImportance returns importance analysis for a project's observations. // handleAnalyzeObservationImportance returns importance analysis for a project's observations.
func (s *Server) handleAnalyzeObservationImportance(ctx context.Context, args json.RawMessage) (string, error) { func (s *Server) handleAnalyzeObservationImportance(ctx context.Context, args json.RawMessage) (string, error) {
var params struct { var params struct {
Project string `json:"project"`
IncludeTopScored *bool `json:"include_top_scored"` IncludeTopScored *bool `json:"include_top_scored"`
IncludeMostRetrieved *bool `json:"include_most_retrieved"` IncludeMostRetrieved *bool `json:"include_most_retrieved"`
IncludeConceptWeights *bool `json:"include_concept_weights"` IncludeConceptWeights *bool `json:"include_concept_weights"`
Project string `json:"project"`
Limit int `json:"limit"` Limit int `json:"limit"`
} }
if err := json.Unmarshal(args, &params); err != nil { if err := json.Unmarshal(args, &params); err != nil {
+25 -35
View File
@@ -63,18 +63,16 @@ const (
// SearchMetrics tracks search performance statistics. // SearchMetrics tracks search performance statistics.
type SearchMetrics struct { type SearchMetrics struct {
TotalSearches int64 // Total number of searches performed latencyHistogram []int64
VectorSearches int64 // Searches using vector search TotalSearches int64
FilterSearches int64 // Searches using filter/FTS search VectorSearches int64
TotalLatencyNs int64 // Cumulative latency in nanoseconds FilterSearches int64
VectorLatencyNs int64 // Cumulative vector search latency TotalLatencyNs int64
FilterLatencyNs int64 // Cumulative filter search latency VectorLatencyNs int64
CacheHits int64 // Number of result cache hits FilterLatencyNs int64
CoalescedRequests int64 // Number of requests served via singleflight coalescing CacheHits int64
SearchErrors int64 // Number of search errors CoalescedRequests int64
SearchErrors int64
// Percentile tracking (approximate using reservoir sampling)
latencyHistogram []int64 // Recent latency samples
histogramMu sync.Mutex histogramMu sync.Mutex
} }
@@ -117,36 +115,28 @@ func (m *SearchMetrics) GetStats() map[string]any {
// Manager provides unified search across SQLite and sqlite-vec. // Manager provides unified search across SQLite and sqlite-vec.
type Manager struct { type Manager struct {
observationStore *gorm.ObservationStore ctx context.Context
summaryStore *gorm.SummaryStore searchGroup singleflight.Group
promptStore *gorm.PromptStore cancel context.CancelFunc
vectorClient *sqlitevec.Client vectorClient *sqlitevec.Client
metrics *SearchMetrics metrics *SearchMetrics
promptStore *gorm.PromptStore
// Context for graceful shutdown of background goroutines observationStore *gorm.ObservationStore
ctx context.Context summaryStore *gorm.SummaryStore
cancel context.CancelFunc
// Request coalescing for concurrent identical queries
searchGroup singleflight.Group
// Result cache for repeated queries (short TTL)
resultCache map[string]*cachedResult resultCache map[string]*cachedResult
resultCacheMu sync.RWMutex queryFrequency map[string]*queryFrequencyInfo
cacheTTL time.Duration cacheTTL time.Duration
cacheMaxSize int cacheMaxSize int
resultCacheMu sync.RWMutex
// Query frequency tracking for cache warming
queryFrequency map[string]*queryFrequencyInfo
queryFrequencyMu sync.RWMutex queryFrequencyMu sync.RWMutex
} }
// queryFrequencyInfo tracks how often a query is used. // queryFrequencyInfo tracks how often a query is used.
type queryFrequencyInfo struct { type queryFrequencyInfo struct {
params SearchParams
count int64
lastUsed time.Time lastUsed time.Time
lastCached time.Time lastCached time.Time
params SearchParams
count int64
} }
// cachedResult stores a cached search result with expiry. // cachedResult stores a cached search result with expiry.
@@ -270,8 +260,8 @@ func (m *Manager) warmFrequentQueries() {
m.queryFrequencyMu.RLock() m.queryFrequencyMu.RLock()
// Find top N most frequent queries that aren't recently cached // Find top N most frequent queries that aren't recently cached
type queryScore struct { type queryScore struct {
key string
info *queryFrequencyInfo info *queryFrequencyInfo
key string
score float64 score float64
} }
candidates := make([]queryScore, 0, len(m.queryFrequency)) candidates := make([]queryScore, 0, len(m.queryFrequency))
@@ -359,8 +349,8 @@ func (m *Manager) trackQueryFrequency(params SearchParams) {
// Collect keys and times for eviction (still under lock, but fast) // Collect keys and times for eviction (still under lock, but fast)
type entry struct { type entry struct {
key string
lastUsed time.Time lastUsed time.Time
key string
} }
entries := make([]entry, 0, mapLen) entries := make([]entry, 0, mapLen)
for k, v := range m.queryFrequency { for k, v := range m.queryFrequency {
@@ -390,11 +380,11 @@ func (m *Manager) trackQueryFrequency(params SearchParams) {
// RecentQuery represents a recently executed search query. // RecentQuery represents a recently executed search query.
type RecentQuery struct { type RecentQuery struct {
LastUsed time.Time `json:"last_used"`
Query string `json:"query"` Query string `json:"query"`
Project string `json:"project,omitempty"` Project string `json:"project,omitempty"`
Type string `json:"type,omitempty"` // observations, sessions, prompts Type string `json:"type,omitempty"`
Count int64 `json:"count"` // Number of times executed Count int64 `json:"count"`
LastUsed time.Time `json:"last_used"`
} }
// GetRecentQueries returns the most recent search queries, sorted by last used time. // GetRecentQueries returns the most recent search queries, sorted by last used time.
+21 -33
View File
@@ -25,41 +25,29 @@ type embeddingCacheEntry struct {
// resultCacheEntry stores cached query results. // resultCacheEntry stores cached query results.
type resultCacheEntry struct { type resultCacheEntry struct {
queryHash string
results []QueryResult results []QueryResult
timestamp int64 // Unix nano timestamp int64
queryHash string // Hash of query + filters for validation
} }
// Client provides vector operations via sqlite-vec. // Client provides vector operations via sqlite-vec.
type Client struct { type Client struct {
embeddingGroup singleflight.Group
resultCache map[string]resultCacheEntry
db *sql.DB db *sql.DB
embedSvc *embedding.Service embedSvc *embedding.Service
// Separate mutexes for read and write operations to reduce contention
writeMu sync.Mutex // Protects write operations (AddDocuments, DeleteDocuments)
readMu sync.RWMutex // Protects read operations (Query, Count)
// Embedding cache for query deduplication
queryCache map[string]embeddingCacheEntry queryCache map[string]embeddingCacheEntry
queryCacheMu sync.RWMutex
cacheMaxSize int
cacheTTLNano int64 // Cache TTL in nanoseconds
// Result cache for repeated searches
resultCache map[string]resultCacheEntry
resultCacheMu sync.RWMutex
resultCacheMaxSize int
resultCacheTTLNano int64 // Shorter TTL for results (data changes more often)
// Cache statistics
stats CacheStats
// Background cleanup control
stopCleanup chan struct{} stopCleanup chan struct{}
stats CacheStats
cleanupWg sync.WaitGroup cleanupWg sync.WaitGroup
resultCacheTTLNano int64
// Singleflight to deduplicate concurrent embedding computations cacheTTLNano int64
embeddingGroup singleflight.Group resultCacheMaxSize int
cacheMaxSize int
resultCacheMu sync.RWMutex
queryCacheMu sync.RWMutex
readMu sync.RWMutex
writeMu sync.Mutex
} }
// CacheStats tracks cache performance metrics using atomic counters for lock-free updates. // CacheStats tracks cache performance metrics using atomic counters for lock-free updates.
@@ -442,9 +430,9 @@ func (c *Client) cleanupExpiredCaches() {
// BatchQueryResult holds results from a batch query operation. // BatchQueryResult holds results from a batch query operation.
type BatchQueryResult struct { type BatchQueryResult struct {
Query string // Original query string Error error
Results []QueryResult // Results for this query Query string
Error error // Error if query failed Results []QueryResult
} }
// QueryBatch performs multiple vector searches concurrently. // QueryBatch performs multiple vector searches concurrently.
@@ -691,15 +679,15 @@ func (c *Client) GetStaleVectors(ctx context.Context) ([]StaleVectorInfo, error)
// VectorHealthStats contains comprehensive health information about the vector store. // VectorHealthStats contains comprehensive health information about the vector store.
type VectorHealthStats struct { type VectorHealthStats struct {
TotalVectors int64 `json:"total_vectors"`
StaleVectors int64 `json:"stale_vectors"`
CurrentModel string `json:"current_model"`
NeedsRebuild bool `json:"needs_rebuild"`
RebuildReason string `json:"rebuild_reason,omitempty"`
CoverageByType map[string]int64 `json:"coverage_by_type"` CoverageByType map[string]int64 `json:"coverage_by_type"`
ModelVersions map[string]int64 `json:"model_versions"` ModelVersions map[string]int64 `json:"model_versions"`
ProjectCounts map[string]int64 `json:"project_counts"` ProjectCounts map[string]int64 `json:"project_counts"`
CurrentModel string `json:"current_model"`
RebuildReason string `json:"rebuild_reason,omitempty"`
EmbeddingCache CacheStatsSnapshot `json:"embedding_cache"` EmbeddingCache CacheStatsSnapshot `json:"embedding_cache"`
TotalVectors int64 `json:"total_vectors"`
StaleVectors int64 `json:"stale_vectors"`
NeedsRebuild bool `json:"needs_rebuild"`
} }
// GetHealthStats returns comprehensive health statistics about the vector store. // GetHealthStats returns comprehensive health statistics about the vector store.
+1 -1
View File
@@ -484,11 +484,11 @@ type UpdateObservationRequest struct {
Title *string `json:"title,omitempty"` Title *string `json:"title,omitempty"`
Subtitle *string `json:"subtitle,omitempty"` Subtitle *string `json:"subtitle,omitempty"`
Narrative *string `json:"narrative,omitempty"` Narrative *string `json:"narrative,omitempty"`
Scope *string `json:"scope,omitempty"`
Facts []string `json:"facts,omitempty"` Facts []string `json:"facts,omitempty"`
Concepts []string `json:"concepts,omitempty"` Concepts []string `json:"concepts,omitempty"`
FilesRead []string `json:"files_read,omitempty"` FilesRead []string `json:"files_read,omitempty"`
FilesModified []string `json:"files_modified,omitempty"` FilesModified []string `json:"files_modified,omitempty"`
Scope *string `json:"scope,omitempty"`
} }
// handleUpdateObservation updates an existing observation. // handleUpdateObservation updates an existing observation.
+12 -12
View File
@@ -29,23 +29,23 @@ type BulkImportRequest struct {
// BulkObservationInput represents a single observation in bulk import. // BulkObservationInput represents a single observation in bulk import.
type BulkObservationInput struct { type BulkObservationInput struct {
Type string `json:"type"` // bugfix, feature, refactor, etc. Type string `json:"type"`
Title string `json:"title"` Title string `json:"title"`
Subtitle string `json:"subtitle,omitempty"` Subtitle string `json:"subtitle,omitempty"`
Facts []string `json:"facts,omitempty"`
Narrative string `json:"narrative,omitempty"` Narrative string `json:"narrative,omitempty"`
Scope string `json:"scope,omitempty"`
Facts []string `json:"facts,omitempty"`
Concepts []string `json:"concepts,omitempty"` Concepts []string `json:"concepts,omitempty"`
FilesRead []string `json:"files_read,omitempty"` FilesRead []string `json:"files_read,omitempty"`
FilesModified []string `json:"files_modified,omitempty"` FilesModified []string `json:"files_modified,omitempty"`
Scope string `json:"scope,omitempty"` // project or global
} }
// BulkImportResponse contains the result of a bulk import operation. // BulkImportResponse contains the result of a bulk import operation.
type BulkImportResponse struct { type BulkImportResponse struct {
Errors []string `json:"errors,omitempty"`
Imported int `json:"imported"` Imported int `json:"imported"`
Failed int `json:"failed"` Failed int `json:"failed"`
SkippedDuplicates int `json:"skipped_duplicates,omitempty"` SkippedDuplicates int `json:"skipped_duplicates,omitempty"`
Errors []string `json:"errors,omitempty"`
} }
// handleBulkImport handles bulk import of observations. // handleBulkImport handles bulk import of observations.
@@ -208,10 +208,10 @@ func (s *Service) handleBulkImport(w http.ResponseWriter, r *http.Request) {
// ArchiveRequest is the request body for archiving observations. // ArchiveRequest is the request body for archiving observations.
type ArchiveRequest struct { type ArchiveRequest struct {
IDs []int64 `json:"ids,omitempty"` // Specific IDs to archive Project string `json:"project,omitempty"`
Project string `json:"project,omitempty"` // Archive all in project older than max_age_days Reason string `json:"reason,omitempty"`
MaxAgeDays int `json:"max_age_days,omitempty"` // Only used with project IDs []int64 `json:"ids,omitempty"`
Reason string `json:"reason,omitempty"` // Optional reason for archival MaxAgeDays int `json:"max_age_days,omitempty"`
} }
// handleArchiveObservations archives observations by ID or by age. // handleArchiveObservations archives observations by ID or by age.
@@ -233,8 +233,8 @@ func (s *Service) handleArchiveObservations(w http.ResponseWriter, r *http.Reque
if len(req.IDs) > 5 { if len(req.IDs) > 5 {
// Use parallel archival for batches larger than 5 // Use parallel archival for batches larger than 5
type archiveResult struct { type archiveResult struct {
id int64
err error err error
id int64
} }
results := make(chan archiveResult, len(req.IDs)) results := make(chan archiveResult, len(req.IDs))
@@ -491,10 +491,10 @@ func escapeCsvField(s string) string {
// BulkStatusRequest represents a request to update status for multiple observations. // BulkStatusRequest represents a request to update status for multiple observations.
type BulkStatusRequest struct { type BulkStatusRequest struct {
IDs []int64 `json:"ids"` Action string `json:"action"`
Action string `json:"action"` // "supersede", "archive", "set_feedback"
Reason string `json:"reason,omitempty"` Reason string `json:"reason,omitempty"`
Feedback int `json:"feedback,omitempty"` // -1, 0, 1 for set_feedback action IDs []int64 `json:"ids"`
Feedback int `json:"feedback,omitempty"`
} }
// handleBulkStatusUpdate updates status for multiple observations in one request. // handleBulkStatusUpdate updates status for multiple observations in one request.
+1 -1
View File
@@ -25,10 +25,10 @@ type SessionInitRequest struct {
// SessionInitResponse is the response for session initialization. // SessionInitResponse is the response for session initialization.
type SessionInitResponse struct { type SessionInitResponse struct {
Reason string `json:"reason,omitempty"`
SessionDBID int64 `json:"sessionDbId"` SessionDBID int64 `json:"sessionDbId"`
PromptNumber int `json:"promptNumber"` PromptNumber int `json:"promptNumber"`
Skipped bool `json:"skipped,omitempty"` Skipped bool `json:"skipped,omitempty"`
Reason string `json:"reason,omitempty"`
} }
// DuplicatePromptWindowSeconds is the time window for detecting duplicate prompt submissions. // DuplicatePromptWindowSeconds is the time window for detecting duplicate prompt submissions.
+25 -12
View File
@@ -66,6 +66,7 @@ func testService(t *testing.T) (*Service, func()) {
cancel: cancel, cancel: cancel,
startTime: time.Now(), startTime: time.Now(),
retrievalStats: make(map[string]*RetrievalStats), retrievalStats: make(map[string]*RetrievalStats),
cachedObsCounts: make(map[string]cachedCount),
} }
svc.setupRoutes() svc.setupRoutes()
@@ -345,11 +346,13 @@ func TestHandleGetObservations_Limit(t *testing.T) {
assert.Equal(t, http.StatusOK, rec.Code) assert.Equal(t, http.StatusOK, rec.Code)
// Parse as generic JSON array since the model uses custom marshaling // Parse as object with observations key (API returns wrapped response)
var observations []map[string]interface{} var response map[string]interface{}
err := json.Unmarshal(rec.Body.Bytes(), &observations) err := json.Unmarshal(rec.Body.Bytes(), &response)
require.NoError(t, err) require.NoError(t, err)
observations, ok := response["observations"].([]interface{})
require.True(t, ok, "expected observations array in response")
assert.Len(t, observations, 10) assert.Len(t, observations, 10)
} }
@@ -1135,10 +1138,13 @@ func TestHandleGetObservations_DefaultLimit(t *testing.T) {
assert.Equal(t, http.StatusOK, rec.Code) assert.Equal(t, http.StatusOK, rec.Code)
var observations []map[string]interface{} var response map[string]interface{}
err := json.Unmarshal(rec.Body.Bytes(), &observations) err := json.Unmarshal(rec.Body.Bytes(), &response)
require.NoError(t, err) require.NoError(t, err)
observations, ok := response["observations"].([]interface{})
require.True(t, ok, "expected observations array in response")
// Should return default limit (100) // Should return default limit (100)
assert.LessOrEqual(t, len(observations), DefaultObservationsLimit) assert.LessOrEqual(t, len(observations), DefaultObservationsLimit)
} }
@@ -1159,10 +1165,12 @@ func TestHandleGetObservations_FilterByProject(t *testing.T) {
assert.Equal(t, http.StatusOK, rec.Code) assert.Equal(t, http.StatusOK, rec.Code)
var observations []map[string]interface{} var response map[string]interface{}
err := json.Unmarshal(rec.Body.Bytes(), &observations) err := json.Unmarshal(rec.Body.Bytes(), &response)
require.NoError(t, err) require.NoError(t, err)
observations, ok := response["observations"].([]interface{})
require.True(t, ok, "expected observations array in response")
assert.Len(t, observations, 2) assert.Len(t, observations, 2)
} }
@@ -1412,10 +1420,12 @@ func TestHandleGetObservations(t *testing.T) {
assert.Equal(t, http.StatusOK, rec.Code) assert.Equal(t, http.StatusOK, rec.Code)
var observations []map[string]interface{} var response map[string]interface{}
err := json.Unmarshal(rec.Body.Bytes(), &observations) err := json.Unmarshal(rec.Body.Bytes(), &response)
require.NoError(t, err) require.NoError(t, err)
observations, ok := response["observations"].([]interface{})
require.True(t, ok, "expected observations array in response")
assert.GreaterOrEqual(t, len(observations), 2) assert.GreaterOrEqual(t, len(observations), 2)
} }
@@ -2697,10 +2707,13 @@ func TestHandleGetObservations_EmptyResult(t *testing.T) {
assert.Equal(t, http.StatusOK, rec.Code) assert.Equal(t, http.StatusOK, rec.Code)
// Should return empty array, not null // Should return empty array within observations key, not null
var obs []interface{} var response map[string]interface{}
err := json.Unmarshal(rec.Body.Bytes(), &obs) err := json.Unmarshal(rec.Body.Bytes(), &response)
require.NoError(t, err) require.NoError(t, err)
obs, ok := response["observations"].([]interface{})
require.True(t, ok, "expected observations array in response")
assert.NotNil(t, obs) assert.NotNil(t, obs)
} }
+3 -5
View File
@@ -91,12 +91,10 @@ func MaxBodySize(maxBytes int64) func(http.Handler) http.Handler {
// TokenAuth provides simple token-based authentication for localhost services. // TokenAuth provides simple token-based authentication for localhost services.
// The token is generated at startup and must be provided in the X-Auth-Token header. // The token is generated at startup and must be provided in the X-Auth-Token header.
type TokenAuth struct { type TokenAuth struct {
token string
enabled bool
mu sync.RWMutex
// ExemptPaths are paths that don't require authentication (e.g., health checks)
ExemptPaths map[string]bool ExemptPaths map[string]bool
token string
mu sync.RWMutex
enabled bool
} }
// NewTokenAuth creates a new TokenAuth with a randomly generated token. // NewTokenAuth creates a new TokenAuth with a randomly generated token.
+1 -1
View File
@@ -42,8 +42,8 @@ func TestSecurityHeaders_CORS(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
origin string origin string
expectCORS bool
expectedOrigin string expectedOrigin string
expectCORS bool
}{ }{
{ {
name: "localhost:37778 origin allowed", name: "localhost:37778 origin allowed",
+10 -11
View File
@@ -9,13 +9,13 @@ import (
// RateLimiter implements a token bucket rate limiter. // RateLimiter implements a token bucket rate limiter.
type RateLimiter struct { type RateLimiter struct {
rate float64 // tokens per second lastUpdate time.Time
burst int // maximum burst size rate float64
mu sync.Mutex // protects following fields burst int
tokens float64 // current tokens tokens float64
lastUpdate time.Time // last token update time requests int64
requests int64 // total requests rejected int64
rejected int64 // rejected requests mu sync.Mutex
} }
// LastUpdateTime returns the last update time. // LastUpdateTime returns the last update time.
@@ -102,14 +102,13 @@ func RateLimitMiddleware(limiter *RateLimiter) func(http.Handler) http.Handler {
// PerClientRateLimiter implements per-client rate limiting. // PerClientRateLimiter implements per-client rate limiting.
type PerClientRateLimiter struct { type PerClientRateLimiter struct {
lastCleanup time.Time
clients map[string]*RateLimiter
rate float64 rate float64
burst int burst int
clients map[string]*RateLimiter
mu sync.Mutex
// Cleanup settings
cleanupInterval time.Duration cleanupInterval time.Duration
maxIdleTime time.Duration maxIdleTime time.Duration
lastCleanup time.Time mu sync.Mutex
} }
// NewPerClientRateLimiter creates a new per-client rate limiter. // NewPerClientRateLimiter creates a new per-client rate limiter.
+33 -74
View File
@@ -101,132 +101,91 @@ const maxRecentQueries = 100
// Service is the main worker service orchestrator. // Service is the main worker service orchestrator.
type Service struct { type Service struct {
// Version of the worker binary startTime time.Time
version string ctx context.Context
initError error
// Configuration server *http.Server
config *config.Config reranker *reranking.Service
// Database
store *gorm.Store
sessionStore *gorm.SessionStore
observationStore *gorm.ObservationStore observationStore *gorm.ObservationStore
summaryStore *gorm.SummaryStore summaryStore *gorm.SummaryStore
promptStore *gorm.PromptStore promptStore *gorm.PromptStore
conflictStore *gorm.ConflictStore conflictStore *gorm.ConflictStore
patternStore *gorm.PatternStore patternStore *gorm.PatternStore
relationStore *gorm.RelationStore relationStore *gorm.RelationStore
// Pattern detection
patternDetector *pattern.Detector patternDetector *pattern.Detector
// Domain services
sessionManager *session.Manager sessionManager *session.Manager
sseBroadcaster *sse.Broadcaster sseBroadcaster *sse.Broadcaster
processor *sdk.Processor processor *sdk.Processor
// Vector database (sqlite-vec with local embeddings)
embedSvc *embedding.Service embedSvc *embedding.Service
vectorClient *sqlitevec.Client vectorClient *sqlitevec.Client
vectorSync *sqlitevec.Sync vectorSync *sqlitevec.Sync
vectorSyncSem chan struct{}
// Cross-encoder reranking (for improved search relevance)
reranker *reranking.Service
// Query expansion (for improved search recall)
queryExpander *expansion.Expander queryExpander *expansion.Expander
// Importance scoring
scoreCalculator *scoring.Calculator scoreCalculator *scoring.Calculator
recalculator *scoring.Recalculator recalculator *scoring.Recalculator
// HTTP server
router *chi.Mux router *chi.Mux
server *http.Server store *gorm.Store
startTime time.Time
// Retrieval statistics (per-project)
retrievalStats map[string]*RetrievalStats retrievalStats map[string]*RetrievalStats
retrievalStatsMu sync.RWMutex sessionStore *gorm.SessionStore
// Lifecycle
ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
wg sync.WaitGroup cachedObsCounts map[string]cachedCount
config *config.Config
// Initialization state (for deferred init) rebuildStatus *RebuildStatus
ready atomic.Bool
initError error
initMu sync.RWMutex
// Background verification queue for stale observations
staleQueue chan staleVerifyRequest staleQueue chan staleVerifyRequest
staleQueueOnce sync.Once bulkOpLimiter *BulkOperationLimiter
// File watchers for auto-recreation on deletion
dbWatcher *watcher.Watcher dbWatcher *watcher.Watcher
configWatcher *watcher.Watcher configWatcher *watcher.Watcher
// Self-updater
updater *update.Updater updater *update.Updater
// Rate limiting
rateLimiter *PerClientRateLimiter rateLimiter *PerClientRateLimiter
expensiveOpLimiter *ExpensiveOperationLimiter expensiveOpLimiter *ExpensiveOperationLimiter
bulkOpLimiter *BulkOperationLimiter version string
recentQueriesBuf [maxRecentQueries]RecentSearchQuery
// Rebuild status tracking wg sync.WaitGroup
rebuildStatus *RebuildStatus recentQueriesLen int
rebuildStatusMu sync.RWMutex recentQueriesHead int
// Recent search query tracking (circular buffer for O(1) insertion)
recentQueriesBuf [maxRecentQueries]RecentSearchQuery // fixed-size circular buffer
recentQueriesHead int // index of most recent (newest)
recentQueriesLen int // current number of items
recentQueriesMu sync.RWMutex
// Stats caching to reduce database load
cachedObsCounts map[string]cachedCount // per-project observation counts
cachedObsCountsMu sync.RWMutex
statsCacheTTL time.Duration statsCacheTTL time.Duration
initMu sync.RWMutex
// Vector sync worker pool - limits concurrent vector sync goroutines rebuildStatusMu sync.RWMutex
vectorSyncSem chan struct{} // semaphore for rate limiting retrievalStatsMu sync.RWMutex
recentQueriesMu sync.RWMutex
cachedObsCountsMu sync.RWMutex
staleQueueOnce sync.Once
ready atomic.Bool
} }
// cachedCount stores a cached count value with expiration. // cachedCount stores a cached count value with expiration.
type cachedCount struct { type cachedCount struct {
count int
timestamp time.Time timestamp time.Time
count int
} }
// RebuildStatus tracks the progress of vector rebuild operations. // RebuildStatus tracks the progress of vector rebuild operations.
type RebuildStatus struct { type RebuildStatus struct {
InProgress bool `json:"in_progress"`
StartTime time.Time `json:"start_time,omitempty"` StartTime time.Time `json:"start_time,omitempty"`
Phase string `json:"phase,omitempty"` // "observations", "summaries", "prompts", "complete" Phase string `json:"phase,omitempty"`
TotalSynced int `json:"total_synced"` TotalSynced int `json:"total_synced"`
TotalErrors int `json:"total_errors"` TotalErrors int `json:"total_errors"`
CurrentPhase int `json:"current_phase"` // 1, 2, 3 for the three phases CurrentPhase int `json:"current_phase"`
TotalPhases int `json:"total_phases"` // 3 TotalPhases int `json:"total_phases"`
ElapsedMs int64 `json:"elapsed_ms,omitempty"` ElapsedMs int64 `json:"elapsed_ms,omitempty"`
EstimatedPct float64 `json:"estimated_pct,omitempty"` // 0-100 EstimatedPct float64 `json:"estimated_pct,omitempty"`
InProgress bool `json:"in_progress"`
} }
// staleVerifyRequest represents a request to verify a stale observation in background // staleVerifyRequest represents a request to verify a stale observation in background
type staleVerifyRequest struct { type staleVerifyRequest struct {
observationID int64
cwd string cwd string
observationID int64
} }
// RecentSearchQuery tracks a search query for analytics. // RecentSearchQuery tracks a search query for analytics.
type RecentSearchQuery struct { type RecentSearchQuery struct {
Timestamp time.Time `json:"timestamp"`
Query string `json:"query"` Query string `json:"query"`
Project string `json:"project,omitempty"` Project string `json:"project,omitempty"`
Type string `json:"type,omitempty"` // observations, summaries, prompts Type string `json:"type,omitempty"`
Results int `json:"results"` Results int `json:"results"`
UsedVector bool `json:"used_vector"` UsedVector bool `json:"used_vector"`
Timestamp time.Time `json:"timestamp"`
} }
// asyncVectorSync executes a vector sync operation with rate limiting. // asyncVectorSync executes a vector sync operation with rate limiting.