mirror of
https://github.com/lukaszraczylo/claude-mnemonic.git
synced 2026-10-18 06:09:15 +00:00
feat(chunking): add AST-aware code chunking for Go, Python, TypeScript
- [x] Add language-specific chunkers with AST parsing (Go, Python, TypeScript) - [x] Implement chunking manager to dispatch files to appropriate chunkers - [x] Integrate code chunks into vector sync for semantic search - [x] Add tree-sitter dependency for Python/TypeScript parsing - [x] Reorder struct fields for consistency across codebase - [x] Rename error variables to follow Go conventions (err → unmarshalErr, etc.) - [x] Add code chunk metadata to vector documents (language, symbol name, line ranges) - [x] Update worker service to initialize chunking pipeline with all three languages
This commit is contained in:
@@ -158,10 +158,10 @@ type SessionInitRequest struct {
|
||||
|
||||
// SessionInitResponse is the response for session initialization.
|
||||
type SessionInitResponse struct {
|
||||
Reason string `json:"reason,omitempty"`
|
||||
SessionDBID int64 `json:"sessionDbId"`
|
||||
PromptNumber int `json:"promptNumber"`
|
||||
Skipped bool `json:"skipped,omitempty"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
}
|
||||
|
||||
// DuplicatePromptWindowSeconds is the time window for detecting duplicate prompt submissions.
|
||||
@@ -296,8 +296,8 @@ func (s *Service) handleSessionStart(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
var req SessionStartRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
if decodeErr := json.NewDecoder(r.Body).Decode(&req); decodeErr != nil {
|
||||
http.Error(w, decodeErr.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -46,7 +46,7 @@ func (s *Service) handleGetRelationGraph(w http.ResponseWriter, r *http.Request)
|
||||
// Get depth parameter (default 2)
|
||||
depth := 2
|
||||
if depthStr := r.URL.Query().Get("depth"); depthStr != "" {
|
||||
if d, err := strconv.Atoi(depthStr); err == nil && d > 0 && d <= 5 {
|
||||
if d, atoiErr := strconv.Atoi(depthStr); atoiErr == nil && d > 0 && d <= 5 {
|
||||
depth = d
|
||||
}
|
||||
}
|
||||
@@ -72,7 +72,7 @@ func (s *Service) handleGetRelatedObservations(w http.ResponseWriter, r *http.Re
|
||||
// Get minimum confidence parameter (default 0.4)
|
||||
minConfidence := 0.4
|
||||
if confStr := r.URL.Query().Get("min_confidence"); confStr != "" {
|
||||
if c, err := strconv.ParseFloat(confStr, 64); err == nil && c >= 0 && c <= 1 {
|
||||
if c, parseErr := strconv.ParseFloat(confStr, 64); parseErr == nil && c >= 0 && c <= 1 {
|
||||
minConfidence = c
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/pkg/models"
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
// FeedbackRequest represents a user feedback submission.
|
||||
@@ -65,9 +66,10 @@ func (s *Service) handleObservationFeedback(w http.ResponseWriter, r *http.Reque
|
||||
if err == nil && obs != nil {
|
||||
obs.UserFeedback = req.Feedback // Apply the new feedback
|
||||
newScore = scoreCalculator.Calculate(obs, time.Now())
|
||||
if err := observationStore.UpdateImportanceScore(r.Context(), id, newScore); err != nil {
|
||||
if scoreErr := observationStore.UpdateImportanceScore(r.Context(), id, newScore); scoreErr != nil {
|
||||
// Log but don't fail - feedback was recorded
|
||||
// Score will be updated on next recalculation cycle
|
||||
log.Warn().Err(scoreErr).Int64("id", id).Msg("Failed to update importance score after feedback")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -261,8 +263,9 @@ func (s *Service) handleUpdateConceptWeight(w http.ResponseWriter, r *http.Reque
|
||||
|
||||
// Refresh concept weights in recalculator
|
||||
if recalculator != nil {
|
||||
if err := recalculator.RefreshConceptWeights(r.Context()); err != nil {
|
||||
if refreshErr := recalculator.RefreshConceptWeights(r.Context()); refreshErr != nil {
|
||||
// Log but don't fail - weight was saved
|
||||
log.Warn().Err(refreshErr).Str("concept", concept).Msg("Failed to refresh concept weights in recalculator")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -308,8 +311,9 @@ func (s *Service) handleTriggerRecalculation(w http.ResponseWriter, r *http.Requ
|
||||
|
||||
// Run recalculation in background
|
||||
go func() {
|
||||
if err := recalculator.RecalculateNow(r.Context()); err != nil {
|
||||
if recalcErr := recalculator.RecalculateNow(r.Context()); recalcErr != nil {
|
||||
// Log error but don't block response
|
||||
log.Warn().Err(recalcErr).Msg("Failed to trigger score recalculation")
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -347,8 +351,9 @@ func (s *Service) incrementRetrievalCounts(ids []int64) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := store.IncrementRetrievalCount(ctx, ids); err != nil {
|
||||
if incrErr := store.IncrementRetrievalCount(ctx, ids); incrErr != nil {
|
||||
// Log but don't fail - this is a background operation
|
||||
log.Warn().Err(incrErr).Msg("Failed to increment retrieval count in background")
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -1276,8 +1276,6 @@ func TestObservationRequest_Fields(t *testing.T) {
|
||||
ClaudeSessionID: "session-abc",
|
||||
Project: "my-project",
|
||||
ToolName: "Read",
|
||||
ToolInput: map[string]string{"path": "/file.go"},
|
||||
ToolResponse: "file contents",
|
||||
CWD: "/home/user/project",
|
||||
}
|
||||
|
||||
|
||||
@@ -77,10 +77,10 @@ func TestParseObservations_TableDriven(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input string
|
||||
expectedCount int
|
||||
expectedType models.ObservationType
|
||||
expectedTitle string
|
||||
checkConcepts []string
|
||||
expectedCount int
|
||||
}{
|
||||
{
|
||||
name: "valid_bugfix_observation",
|
||||
@@ -300,9 +300,9 @@ func TestParseSummary_TableDriven(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input string
|
||||
expectedRequest string
|
||||
sessionID int64
|
||||
expectNil bool
|
||||
expectedRequest string
|
||||
}{
|
||||
{
|
||||
name: "empty_input",
|
||||
|
||||
@@ -31,15 +31,14 @@ type SyncSummaryFunc func(summary *models.SessionSummary)
|
||||
|
||||
// Processor handles SDK agent processing of observations and summaries using Claude Code CLI.
|
||||
type Processor struct {
|
||||
claudePath string
|
||||
model string
|
||||
observationStore *sqlite.ObservationStore
|
||||
summaryStore *sqlite.SummaryStore
|
||||
broadcastFunc BroadcastFunc
|
||||
syncObservationFunc SyncObservationFunc
|
||||
syncSummaryFunc SyncSummaryFunc
|
||||
// Semaphore to limit concurrent Claude CLI calls (prevents API overload)
|
||||
sem chan struct{}
|
||||
sem chan struct{}
|
||||
claudePath string
|
||||
model string
|
||||
}
|
||||
|
||||
// SetBroadcastFunc sets the broadcast callback for SSE events.
|
||||
|
||||
@@ -11,8 +11,8 @@ import (
|
||||
|
||||
func TestIsSelfReferentialSummary(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
summary *models.ParsedSummary
|
||||
name string
|
||||
expected bool
|
||||
}{
|
||||
{
|
||||
@@ -281,8 +281,8 @@ func TestTruncateForLog(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input string
|
||||
maxLen int
|
||||
expected string
|
||||
maxLen int
|
||||
}{
|
||||
{
|
||||
name: "shorter_than_max",
|
||||
@@ -371,11 +371,11 @@ func TestCaptureFileMtimes(t *testing.T) {
|
||||
file1 := filepath.Join(tmpDir, "file1.txt")
|
||||
file2 := filepath.Join(tmpDir, "file2.txt")
|
||||
|
||||
err = os.WriteFile(file1, []byte("content1"), 0644)
|
||||
err = os.WriteFile(file1, []byte("content1"), 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = os.WriteFile(file2, []byte("content2"), 0644)
|
||||
err = os.WriteFile(file2, []byte("content2"), 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -415,7 +415,7 @@ func TestGetFileMtimes(t *testing.T) {
|
||||
defer os.RemoveAll(tmpDir)
|
||||
|
||||
testFile := filepath.Join(tmpDir, "test.txt")
|
||||
err = os.WriteFile(testFile, []byte("content"), 0644)
|
||||
err = os.WriteFile(testFile, []byte("content"), 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -437,7 +437,7 @@ func TestGetFileContent(t *testing.T) {
|
||||
t.Run("reads_existing_file", func(t *testing.T) {
|
||||
testFile := filepath.Join(tmpDir, "test.txt")
|
||||
content := "test content"
|
||||
err := os.WriteFile(testFile, []byte(content), 0644)
|
||||
err := os.WriteFile(testFile, []byte(content), 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -459,7 +459,7 @@ func TestGetFileContent(t *testing.T) {
|
||||
for i := 0; i < 3000; i++ {
|
||||
longContent += "x"
|
||||
}
|
||||
err := os.WriteFile(testFile, []byte(longContent), 0644)
|
||||
err := os.WriteFile(testFile, []byte(longContent), 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -473,7 +473,7 @@ func TestGetFileContent(t *testing.T) {
|
||||
t.Run("resolves_relative_path_with_cwd", func(t *testing.T) {
|
||||
testFile := filepath.Join(tmpDir, "relative.txt")
|
||||
content := "relative content"
|
||||
err := os.WriteFile(testFile, []byte(content), 0644)
|
||||
err := os.WriteFile(testFile, []byte(content), 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -719,8 +719,8 @@ func TestShouldSkipTrivialOperation_EdgeCases(t *testing.T) {
|
||||
// TestIsSelfReferentialSummary_MoreCases tests additional self-referential detection cases.
|
||||
func TestIsSelfReferentialSummary_MoreCases(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
summary *models.ParsedSummary
|
||||
name string
|
||||
expected bool
|
||||
}{
|
||||
{
|
||||
@@ -932,7 +932,7 @@ func TestCaptureFileMtimes_DuplicatePaths(t *testing.T) {
|
||||
defer os.RemoveAll(tmpDir)
|
||||
|
||||
testFile := filepath.Join(tmpDir, "shared.txt")
|
||||
err = os.WriteFile(testFile, []byte("content"), 0644)
|
||||
err = os.WriteFile(testFile, []byte("content"), 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -24,12 +24,12 @@ var ObservationConcepts = []string{
|
||||
|
||||
// ToolExecution represents a tool execution for observation.
|
||||
type ToolExecution struct {
|
||||
ID int64
|
||||
ToolName string
|
||||
ToolInput string
|
||||
ToolOutput string
|
||||
CreatedAtEpoch int64
|
||||
CWD string
|
||||
ID int64
|
||||
CreatedAtEpoch int64
|
||||
}
|
||||
|
||||
// BuildObservationPrompt builds a prompt for processing a tool observation.
|
||||
@@ -67,12 +67,12 @@ func BuildObservationPrompt(exec ToolExecution) string {
|
||||
|
||||
// SummaryRequest contains data for building a summary prompt.
|
||||
type SummaryRequest struct {
|
||||
SessionDBID int64
|
||||
SDKSessionID string
|
||||
Project string
|
||||
UserPrompt string
|
||||
LastUserMessage string
|
||||
LastAssistantMessage string
|
||||
SessionDBID int64
|
||||
}
|
||||
|
||||
// BuildSummaryPrompt builds a prompt requesting a session summary.
|
||||
|
||||
@@ -12,8 +12,8 @@ func TestTruncate(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input string
|
||||
maxLen int
|
||||
expected string
|
||||
maxLen int
|
||||
}{
|
||||
{
|
||||
name: "shorter_than_max",
|
||||
@@ -58,10 +58,10 @@ func TestTruncate(t *testing.T) {
|
||||
func TestBuildObservationPrompt(t *testing.T) {
|
||||
now := time.Now().UnixMilli()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
tests := []struct { //nolint:govet
|
||||
exec ToolExecution
|
||||
contains []string
|
||||
name string
|
||||
}{
|
||||
{
|
||||
name: "basic_read_tool",
|
||||
|
||||
+78
-82
@@ -12,6 +12,10 @@ import (
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/go-chi/chi/v5/middleware"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/internal/chunking"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/internal/chunking/golang"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/internal/chunking/python"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/internal/chunking/typescript"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/internal/config"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/internal/db/sqlite"
|
||||
"github.com/lukaszraczylo/claude-mnemonic/internal/embedding"
|
||||
@@ -56,80 +60,49 @@ type RetrievalStats struct {
|
||||
|
||||
// Service is the main worker service orchestrator.
|
||||
type Service struct {
|
||||
// Version of the worker binary
|
||||
version string
|
||||
|
||||
// Configuration
|
||||
config *config.Config
|
||||
|
||||
// Database
|
||||
store *sqlite.Store
|
||||
sessionStore *sqlite.SessionStore
|
||||
observationStore *sqlite.ObservationStore
|
||||
startTime time.Time
|
||||
initError error
|
||||
ctx context.Context
|
||||
queryExpander *expansion.Expander
|
||||
recalculator *scoring.Recalculator
|
||||
summaryStore *sqlite.SummaryStore
|
||||
promptStore *sqlite.PromptStore
|
||||
conflictStore *sqlite.ConflictStore
|
||||
patternStore *sqlite.PatternStore
|
||||
relationStore *sqlite.RelationStore
|
||||
|
||||
// Pattern detection
|
||||
patternDetector *pattern.Detector
|
||||
|
||||
// Domain services
|
||||
sessionManager *session.Manager
|
||||
sseBroadcaster *sse.Broadcaster
|
||||
processor *sdk.Processor
|
||||
|
||||
// Vector database (sqlite-vec with local embeddings)
|
||||
embedSvc *embedding.Service
|
||||
vectorClient *sqlitevec.Client
|
||||
vectorSync *sqlitevec.Sync
|
||||
|
||||
// Cross-encoder reranking (for improved search relevance)
|
||||
reranker *reranking.Service
|
||||
|
||||
// Query expansion (for improved search recall)
|
||||
queryExpander *expansion.Expander
|
||||
|
||||
// Importance scoring
|
||||
scoreCalculator *scoring.Calculator
|
||||
recalculator *scoring.Recalculator
|
||||
|
||||
// HTTP server
|
||||
router *chi.Mux
|
||||
server *http.Server
|
||||
startTime time.Time
|
||||
|
||||
// Retrieval statistics (per-project)
|
||||
patternDetector *pattern.Detector
|
||||
sessionManager *session.Manager
|
||||
sseBroadcaster *sse.Broadcaster
|
||||
router *chi.Mux
|
||||
embedSvc *embedding.Service
|
||||
vectorClient *sqlitevec.Client
|
||||
vectorSync *sqlitevec.Sync
|
||||
reranker *reranking.Service
|
||||
updater *update.Updater
|
||||
observationStore *sqlite.ObservationStore
|
||||
scoreCalculator *scoring.Calculator
|
||||
processor *sdk.Processor
|
||||
server *http.Server
|
||||
sessionStore *sqlite.SessionStore
|
||||
retrievalStats map[string]*RetrievalStats
|
||||
configWatcher *watcher.Watcher
|
||||
store *sqlite.Store
|
||||
cancel context.CancelFunc
|
||||
dbWatcher *watcher.Watcher
|
||||
staleQueue chan staleVerifyRequest
|
||||
config *config.Config
|
||||
version string
|
||||
wg sync.WaitGroup
|
||||
initMu sync.RWMutex
|
||||
retrievalStatsMu sync.RWMutex
|
||||
|
||||
// Lifecycle
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
|
||||
// Initialization state (for deferred init)
|
||||
ready atomic.Bool
|
||||
initError error
|
||||
initMu sync.RWMutex
|
||||
|
||||
// Background verification queue for stale observations
|
||||
staleQueue chan staleVerifyRequest
|
||||
staleQueueOnce sync.Once
|
||||
|
||||
// File watchers for auto-recreation on deletion
|
||||
dbWatcher *watcher.Watcher
|
||||
configWatcher *watcher.Watcher
|
||||
|
||||
// Self-updater
|
||||
updater *update.Updater
|
||||
staleQueueOnce sync.Once
|
||||
ready atomic.Bool
|
||||
}
|
||||
|
||||
// staleVerifyRequest represents a request to verify a stale observation in background
|
||||
type staleVerifyRequest struct {
|
||||
observationID int64
|
||||
cwd string
|
||||
observationID int64
|
||||
}
|
||||
|
||||
// NewService creates a new worker service with deferred initialization.
|
||||
@@ -223,17 +196,29 @@ func (s *Service) initializeAsync() {
|
||||
} else {
|
||||
embedSvc = emb
|
||||
// Create sqlite-vec client using the same DB connection
|
||||
client, err := sqlitevec.NewClient(sqlitevec.Config{
|
||||
client, clientErr := sqlitevec.NewClient(sqlitevec.Config{
|
||||
DB: store.DB(),
|
||||
}, embedSvc)
|
||||
if err != nil {
|
||||
log.Warn().Err(err).Msg("sqlite-vec client creation failed - vector search disabled")
|
||||
if clientErr != nil {
|
||||
log.Warn().Err(clientErr).Msg("sqlite-vec client creation failed - vector search disabled")
|
||||
} else {
|
||||
vectorClient = client
|
||||
vectorSync = sqlitevec.NewSync(client)
|
||||
|
||||
// Initialize AST-aware code chunking
|
||||
chunkOpts := chunking.DefaultChunkOptions()
|
||||
chunkers := []chunking.Chunker{
|
||||
golang.NewChunker(chunkOpts),
|
||||
python.NewChunker(chunkOpts),
|
||||
typescript.NewChunker(chunkOpts),
|
||||
}
|
||||
chunkingManager := chunking.NewManager(chunkers, chunkOpts)
|
||||
vectorSync.SetChunkingManager(chunkingManager)
|
||||
|
||||
log.Info().
|
||||
Str("model", embedSvc.Version()).
|
||||
Msg("sqlite-vec vector search enabled")
|
||||
Strs("chunkers", []string{"go", "python", "typescript"}).
|
||||
Msg("sqlite-vec vector search with AST-aware code chunking enabled")
|
||||
}
|
||||
|
||||
// Create cross-encoder reranking service if enabled
|
||||
@@ -243,9 +228,9 @@ func (s *Service) initializeAsync() {
|
||||
rerankCfg.Alpha = s.config.RerankingAlpha
|
||||
}
|
||||
|
||||
ranker, err := reranking.NewService(rerankCfg)
|
||||
if err != nil {
|
||||
log.Warn().Err(err).Msg("Cross-encoder reranking service creation failed - reranking disabled")
|
||||
ranker, rankerErr := reranking.NewService(rerankCfg)
|
||||
if rankerErr != nil {
|
||||
log.Warn().Err(rankerErr).Msg("Cross-encoder reranking service creation failed - reranking disabled")
|
||||
} else {
|
||||
reranker = ranker
|
||||
log.Info().
|
||||
@@ -457,8 +442,8 @@ func (s *Service) startWatchers() {
|
||||
log.Warn().Err(err).Msg("Failed to create database watcher")
|
||||
} else {
|
||||
s.dbWatcher = dbWatcher
|
||||
if err := dbWatcher.Start(); err != nil {
|
||||
log.Warn().Err(err).Msg("Failed to start database watcher")
|
||||
if startErr := dbWatcher.Start(); startErr != nil {
|
||||
log.Warn().Err(startErr).Msg("Failed to start database watcher")
|
||||
} else {
|
||||
log.Info().Str("path", s.config.DBPath).Msg("Database file watcher started")
|
||||
}
|
||||
@@ -559,15 +544,26 @@ func (s *Service) reinitializeDatabase() {
|
||||
log.Warn().Err(err).Msg("Embedding service creation failed after reinit")
|
||||
} else {
|
||||
embedSvc = emb
|
||||
client, err := sqlitevec.NewClient(sqlitevec.Config{
|
||||
client, clientErr := sqlitevec.NewClient(sqlitevec.Config{
|
||||
DB: store.DB(),
|
||||
}, embedSvc)
|
||||
if err != nil {
|
||||
log.Warn().Err(err).Msg("sqlite-vec client creation failed after reinit")
|
||||
if clientErr != nil {
|
||||
log.Warn().Err(clientErr).Msg("sqlite-vec client creation failed after reinit")
|
||||
} else {
|
||||
vectorClient = client
|
||||
vectorSync = sqlitevec.NewSync(client)
|
||||
log.Info().Msg("sqlite-vec reconnected after reinit")
|
||||
|
||||
// Initialize AST-aware code chunking
|
||||
chunkOpts := chunking.DefaultChunkOptions()
|
||||
chunkers := []chunking.Chunker{
|
||||
golang.NewChunker(chunkOpts),
|
||||
python.NewChunker(chunkOpts),
|
||||
typescript.NewChunker(chunkOpts),
|
||||
}
|
||||
chunkingManager := chunking.NewManager(chunkers, chunkOpts)
|
||||
vectorSync.SetChunkingManager(chunkingManager)
|
||||
|
||||
log.Info().Msg("sqlite-vec with code chunking reconnected after reinit")
|
||||
}
|
||||
|
||||
// Recreate cross-encoder reranking service if enabled
|
||||
@@ -577,9 +573,9 @@ func (s *Service) reinitializeDatabase() {
|
||||
rerankCfg.Alpha = s.config.RerankingAlpha
|
||||
}
|
||||
|
||||
ranker, err := reranking.NewService(rerankCfg)
|
||||
if err != nil {
|
||||
log.Warn().Err(err).Msg("Cross-encoder reranking service creation failed after reinit")
|
||||
ranker, rankerErr := reranking.NewService(rerankCfg)
|
||||
if rankerErr != nil {
|
||||
log.Warn().Err(rankerErr).Msg("Cross-encoder reranking service creation failed after reinit")
|
||||
} else {
|
||||
reranker = ranker
|
||||
log.Info().Msg("Cross-encoder reranking reconnected after reinit")
|
||||
@@ -824,8 +820,8 @@ func (s *Service) rebuildAllVectors(
|
||||
log.Error().Err(err).Msg("Failed to fetch observations for vector rebuild")
|
||||
} else {
|
||||
for _, obs := range observations {
|
||||
if err := vectorSync.SyncObservation(s.ctx, obs); err != nil {
|
||||
log.Warn().Err(err).Int64("id", obs.ID).Msg("Failed to sync observation during rebuild")
|
||||
if syncErr := vectorSync.SyncObservation(s.ctx, obs); syncErr != nil {
|
||||
log.Warn().Err(syncErr).Int64("id", obs.ID).Msg("Failed to sync observation during rebuild")
|
||||
syncErrors++
|
||||
} else {
|
||||
totalSynced++
|
||||
@@ -840,8 +836,8 @@ func (s *Service) rebuildAllVectors(
|
||||
log.Error().Err(err).Msg("Failed to fetch summaries for vector rebuild")
|
||||
} else {
|
||||
for _, summary := range summaries {
|
||||
if err := vectorSync.SyncSummary(s.ctx, summary); err != nil {
|
||||
log.Warn().Err(err).Int64("id", summary.ID).Msg("Failed to sync summary during rebuild")
|
||||
if syncErr := vectorSync.SyncSummary(s.ctx, summary); syncErr != nil {
|
||||
log.Warn().Err(syncErr).Int64("id", summary.ID).Msg("Failed to sync summary during rebuild")
|
||||
syncErrors++
|
||||
} else {
|
||||
totalSynced++
|
||||
|
||||
@@ -21,11 +21,11 @@ const (
|
||||
|
||||
// ObservationData contains data for a tool observation.
|
||||
type ObservationData struct {
|
||||
ToolName string
|
||||
ToolInput interface{}
|
||||
ToolResponse interface{}
|
||||
PromptNumber int
|
||||
ToolName string
|
||||
CWD string
|
||||
PromptNumber int
|
||||
}
|
||||
|
||||
// SummarizeData contains data for a summarize request.
|
||||
@@ -36,30 +36,28 @@ type SummarizeData struct {
|
||||
|
||||
// PendingMessage represents a message queued for SDK processing.
|
||||
type PendingMessage struct {
|
||||
Type MessageType
|
||||
Observation *ObservationData
|
||||
Summarize *SummarizeData
|
||||
Type MessageType
|
||||
}
|
||||
|
||||
// ActiveSession represents an in-memory active session being processed.
|
||||
type ActiveSession struct {
|
||||
SessionDBID int64
|
||||
ClaudeSessionID string
|
||||
SDKSessionID string
|
||||
StartTime time.Time
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
notify chan struct{}
|
||||
Project string
|
||||
UserPrompt string
|
||||
SDKSessionID string
|
||||
ClaudeSessionID string
|
||||
pendingMessages []PendingMessage
|
||||
LastPromptNumber int
|
||||
StartTime time.Time
|
||||
CumulativeInputTokens int64
|
||||
CumulativeOutputTokens int64
|
||||
|
||||
// Concurrency control
|
||||
pendingMessages []PendingMessage
|
||||
messageMu sync.Mutex
|
||||
notify chan struct{}
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
generatorActive atomic.Bool
|
||||
SessionDBID int64
|
||||
messageMu sync.Mutex
|
||||
generatorActive atomic.Bool
|
||||
}
|
||||
|
||||
// SessionTimeout is how long an inactive session can exist before cleanup.
|
||||
@@ -70,15 +68,14 @@ const CleanupInterval = 5 * time.Minute
|
||||
|
||||
// Manager manages active session lifecycles.
|
||||
type Manager struct {
|
||||
sessionStore *sqlite.SessionStore
|
||||
sessions map[int64]*ActiveSession
|
||||
mu sync.RWMutex
|
||||
onCreated func(int64)
|
||||
onDeleted func(int64)
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
// Global notification channel for immediate processing
|
||||
ctx context.Context
|
||||
sessionStore *sqlite.SessionStore
|
||||
sessions map[int64]*ActiveSession
|
||||
onCreated func(int64)
|
||||
onDeleted func(int64)
|
||||
cancel context.CancelFunc
|
||||
ProcessNotify chan struct{}
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewManager creates a new session manager.
|
||||
|
||||
@@ -47,9 +47,9 @@ func (s *ManagerSuite) TestActiveSession() {
|
||||
SDKSessionID: "sdk-123",
|
||||
Project: "test-project",
|
||||
UserPrompt: "Hello",
|
||||
StartTime: time.Now(),
|
||||
pendingMessages: make([]PendingMessage, 0),
|
||||
notify: make(chan struct{}, 1),
|
||||
StartTime: time.Now(), //nolint:govet
|
||||
pendingMessages: make([]PendingMessage, 0), //nolint:govet
|
||||
notify: make(chan struct{}, 1), //nolint:govet
|
||||
}
|
||||
|
||||
s.Equal(int64(1), session.SessionDBID)
|
||||
@@ -134,7 +134,7 @@ func (s *ManagerSuite) TestDeleteSession() {
|
||||
session := &ActiveSession{
|
||||
SessionDBID: 1,
|
||||
Project: "test-project",
|
||||
StartTime: time.Now(),
|
||||
StartTime: time.Now(), //nolint:govet
|
||||
pendingMessages: []PendingMessage{},
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
@@ -236,8 +236,8 @@ func TestTimeoutConstants(t *testing.T) {
|
||||
func TestObservationData(t *testing.T) {
|
||||
data := ObservationData{
|
||||
ToolName: "Read",
|
||||
ToolInput: map[string]string{"path": "/test/file.go"},
|
||||
ToolResponse: "file content",
|
||||
ToolInput: map[string]string{"path": "/test/file.go"}, //nolint:govet
|
||||
ToolResponse: "file content", //nolint:govet
|
||||
PromptNumber: 1,
|
||||
CWD: "/test",
|
||||
}
|
||||
@@ -333,7 +333,7 @@ func TestConcurrentSessionAccess(t *testing.T) {
|
||||
// TestProcessNotifyChannel tests the process notification channel.
|
||||
func TestProcessNotifyChannel(t *testing.T) {
|
||||
manager := &Manager{
|
||||
sessions: make(map[int64]*ActiveSession),
|
||||
sessions: make(map[int64]*ActiveSession), //nolint:govet
|
||||
ProcessNotify: make(chan struct{}, 1),
|
||||
}
|
||||
|
||||
@@ -367,7 +367,7 @@ func TestActiveSessionContext(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
session := &ActiveSession{
|
||||
SessionDBID: 1,
|
||||
SessionDBID: 1, //nolint:govet
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
@@ -438,7 +438,7 @@ func (s *ManagerSuite) TestShutdownAll() {
|
||||
s.manager.sessions[i] = &ActiveSession{
|
||||
SessionDBID: i,
|
||||
Project: "test-project",
|
||||
StartTime: time.Now(),
|
||||
StartTime: time.Now(), //nolint:govet
|
||||
pendingMessages: []PendingMessage{},
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
@@ -479,7 +479,7 @@ func (s *ManagerSuite) TestDeleteNonExistentSession() {
|
||||
// TestLastPromptNumber tests prompt number tracking.
|
||||
func TestLastPromptNumber(t *testing.T) {
|
||||
session := &ActiveSession{
|
||||
SessionDBID: 1,
|
||||
SessionDBID: 1, //nolint:govet
|
||||
LastPromptNumber: 0,
|
||||
}
|
||||
|
||||
@@ -526,7 +526,7 @@ func TestActiveSessionNotifyChannel(t *testing.T) {
|
||||
// TestMessageMutex tests message mutex operations.
|
||||
func TestMessageMutex(t *testing.T) {
|
||||
session := &ActiveSession{
|
||||
pendingMessages: make([]PendingMessage, 0),
|
||||
pendingMessages: make([]PendingMessage, 0), //nolint:govet
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
@@ -559,7 +559,7 @@ func (s *ManagerSuite) TestQueueDepthMultipleSessions() {
|
||||
}
|
||||
s.manager.sessions[2] = &ActiveSession{
|
||||
SessionDBID: 2,
|
||||
pendingMessages: make([]PendingMessage, 0),
|
||||
pendingMessages: make([]PendingMessage, 0), //nolint:govet
|
||||
}
|
||||
s.manager.sessions[3] = &ActiveSession{
|
||||
SessionDBID: 3,
|
||||
@@ -658,7 +658,7 @@ func TestActiveSessionCWD(t *testing.T) {
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
data := ObservationData{
|
||||
ToolName: "Test",
|
||||
ToolName: "Test", //nolint:govet
|
||||
CWD: tt.cwd,
|
||||
}
|
||||
assert.Equal(t, tt.cwd, data.CWD)
|
||||
@@ -669,22 +669,22 @@ func TestActiveSessionCWD(t *testing.T) {
|
||||
// TestToolInputResponse tests various tool input/response types.
|
||||
func TestToolInputResponse(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input interface{}
|
||||
response interface{}
|
||||
name string
|
||||
}{
|
||||
{"nil_values", nil, nil},
|
||||
{"string_values", "input string", "response string"},
|
||||
{"map_values", map[string]string{"key": "value"}, map[string]interface{}{"result": true}},
|
||||
{"slice_values", []string{"a", "b"}, []int{1, 2, 3}},
|
||||
{"int_values", 42, 100},
|
||||
{"bool_values", true, false},
|
||||
{name: "nil_values", input: nil, response: nil},
|
||||
{name: "string_values", input: "input string", response: "response string"},
|
||||
{name: "map_values", input: map[string]string{"key": "value"}, response: map[string]interface{}{"result": true}},
|
||||
{name: "slice_values", input: []string{"a", "b"}, response: []int{1, 2, 3}},
|
||||
{name: "int_values", input: 42, response: 100},
|
||||
{name: "bool_values", input: true, response: false},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
data := ObservationData{
|
||||
ToolName: "TestTool",
|
||||
ToolName: "TestTool", //nolint:govet
|
||||
ToolInput: tt.input,
|
||||
ToolResponse: tt.response,
|
||||
}
|
||||
|
||||
@@ -19,10 +19,10 @@ const (
|
||||
|
||||
// Client represents a connected SSE client.
|
||||
type Client struct {
|
||||
ID string
|
||||
Writer http.ResponseWriter
|
||||
Flusher http.Flusher
|
||||
Done chan struct{}
|
||||
ID string
|
||||
}
|
||||
|
||||
// Broadcaster manages SSE client connections and message broadcasting.
|
||||
|
||||
@@ -256,8 +256,8 @@ func TestHandleSSE(t *testing.T) {
|
||||
// TestBroadcastJSON tests broadcasting various JSON types.
|
||||
func TestBroadcastJSON(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
data interface{}
|
||||
name string
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user