worker

package
v1.9.0 Latest Latest
Warning

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

 Go to latest Published: Mar 27, 2026 License: MIT Imports: 69 Imported by: 0

Documentation

Overview

Package worker provides the main worker service for engram.

Package worker provides the main worker service for engram. This file contains shared handler utilities and health/status endpoints. Domain-specific handlers are split into:

  • handlers_sessions.go: Session lifecycle (init, start, observation, summarize)
  • handlers_context.go: Context/search (search by prompt, file context, inject)
  • handlers_data.go: Data retrieval (observations, summaries, prompts, stats)
  • handlers_update.go: Updates and self-check (update check/apply, self-check)
  • handlers_import_export.go: Import/export/archive operations

Package worker provides analytics REST handlers for the dashboard.

Package worker provides authentication HTTP handlers for the dashboard.

Package worker provides context and search-related HTTP handlers.

Package worker provides data retrieval HTTP handlers.

Package worker provides import, export, and archive HTTP handlers.

Package worker provides the main worker service for engram.

Package worker provides learning-related HTTP handlers.

Package worker provides maintenance REST handlers for the dashboard.

Package worker provides the main worker service for engram.

Package worker provides the main worker service for engram.

Package worker provides the main worker service for engram.

Package worker provides session-related HTTP handlers.

Package worker provides indexed session REST handlers for the dashboard.

Package worker provides tag management REST handlers for the dashboard.

Package worker provides update and restart HTTP handlers.

Package worker provides vault credential REST handlers for the dashboard.

Package worker provides the main worker service for engram.

Package worker provides the main worker service for engram.

Package worker provides the main worker service for engram.

Package worker provides the buffered token stats flusher.

Index

Constants

View Source 
const (
	// DefaultObservationsLimit is the default number of observations to return.
	DefaultObservationsLimit = 100

	// DefaultSummariesLimit is the default number of summaries to return.
	DefaultSummariesLimit = 50

	// DefaultPromptsLimit is the default number of prompts to return.
	DefaultPromptsLimit = 100

	// DefaultSearchLimit is the default number of search results to return.
	DefaultSearchLimit = 50

	// DefaultContextLimit is the default number of context observations to return.
	DefaultContextLimit = 50
)

Handler configuration constants

View Source 
const (
	// DefaultHTTPTimeout is the default timeout for HTTP requests.
	DefaultHTTPTimeout = 30 * time.Second

	// ReadyPollInterval is how often WaitReady checks initialization status.
	ReadyPollInterval = 50 * time.Millisecond

	// StaleQueueSize is the buffer size for background stale verification.
	StaleQueueSize = 100

	// QueueProcessInterval is how often the background queue processor runs.
	QueueProcessInterval = 2 * time.Second

	// VectorSyncMaxRetries is the maximum number of retries for vector sync operations.
	VectorSyncMaxRetries = 3

	// VectorSyncInitialBackoff is the initial backoff duration for retry.
	VectorSyncInitialBackoff = 100 * time.Millisecond
)

Service configuration constants

View Source 
const DefaultPatternsLimit = 500

DefaultPatternsLimit is the default number of patterns to return.

View Source 
const DefaultRelationsLimit = 50

DefaultRelationsLimit is the default number of relations to return.

View Source 
const DuplicatePromptWindowSeconds = 10

DuplicatePromptWindowSeconds is the time window for detecting duplicate prompt submissions. If the same prompt text is seen within this window, it's considered a duplicate hook invocation.

Variables

View Source 
var ConceptTypes = []string{

	"how-it-works",
	"why-it-exists",
	"what-changed",
	"problem-solution",
	"gotcha",
	"pattern",
	"trade-off",

	"best-practice",
	"anti-pattern",
	"architecture",
	"security",
	"performance",
	"testing",
	"debugging",
	"workflow",
	"tooling",

	"refactoring",
	"api",
	"database",
	"configuration",
	"error-handling",
	"caching",
	"logging",
	"auth",
	"validation",
}

ConceptTypes is the canonical list of valid concept types. Used by both Go backend and served to frontend.

View Source 
var ObservationTypes = []string{
	"bugfix",
	"feature",
	"refactor",
	"discovery",
	"decision",
	"change",
}

ObservationTypes is the canonical list of observation types. Used by both Go backend and served to frontend.

Functions

func GetRequestID 

func GetRequestID(ctx context.Context) string

GetRequestID retrieves the request ID from the context.

func IsValidConceptType 

func IsValidConceptType(t string) bool

IsValidConceptType returns true if the concept type is valid (O(1) lookup).

func IsValidObservationType 

func IsValidObservationType(t string) bool

IsValidObservationType returns true if the type is valid (O(1) lookup).

func MaxBodySize 

func MaxBodySize(maxBytes int64) func(http.Handler) http.Handler

MaxBodySize middleware limits the size of incoming request bodies. This prevents denial of service attacks via large payloads.

func PerClientRateLimitMiddleware 

func PerClientRateLimitMiddleware(limiter *PerClientRateLimiter) func(http.Handler) http.Handler

PerClientRateLimitMiddleware creates middleware that applies per-client rate limiting. Uses X-Forwarded-For or RemoteAddr to identify clients.

func RateLimitMiddleware 

func RateLimitMiddleware(limiter *RateLimiter) func(http.Handler) http.Handler

RateLimitMiddleware creates middleware that applies rate limiting. Uses a shared rate limiter for all requests.

func RequestID 

func RequestID(next http.Handler) http.Handler

RequestID middleware adds a unique request ID to each request. The ID is added to the context and response headers for tracing.

func RequireJSONContentType 

func RequireJSONContentType(next http.Handler) http.Handler

RequireJSONContentType middleware validates that POST/PUT/PATCH requests have application/json Content-Type header.

func SecurityHeaders 

func SecurityHeaders(next http.Handler) http.Handler

SecurityHeaders middleware adds essential security headers to all responses. These protect against common web vulnerabilities.

func ValidateProjectName 

func ValidateProjectName(project string) error

ValidateProjectName checks if a project name is safe to use. Returns an error if the name contains path traversal or invalid characters.

Types

type ArchiveRequest 

type ArchiveRequest struct {
	Project    string  `json:"project,omitempty"`
	Reason     string  `json:"reason,omitempty"`
	IDs        []int64 `json:"ids,omitempty"`
	MaxAgeDays int     `json:"max_age_days,omitempty"`
}

ArchiveRequest is the request body for archiving observations.

type BackfillObservation added in v0.4.0 

type BackfillObservation struct {
	Type      string   `json:"type"`
	Outcome   string   `json:"outcome"`
	Title     string   `json:"title"`
	Narrative string   `json:"narrative"`
	Concepts  []string `json:"concepts"`
	Files     []string `json:"files"`
}

BackfillObservation is a single observation from a backfill extraction.

type BackfillRequest added in v0.4.0 

type BackfillRequest struct {
	// SessionID is a unique identifier for the source session (e.g. filename hash).
	SessionID string `json:"session_id"`
	// Project is the project path from session metadata.
	Project string `json:"project"`
	// RunID groups observations from the same backfill run (for rollback).
	RunID string `json:"run_id"`
	// Observations are the extracted observations to store.
	Observations []BackfillObservation `json:"observations"`
}

BackfillRequest is the request body for POST /api/backfill.

type BackfillResponse added in v0.4.0 

type BackfillResponse struct {
	Stored  int `json:"stored"`
	Skipped int `json:"skipped"`
	Errors  int `json:"errors"`
}

BackfillResponse is the response for POST /api/backfill.

type BackfillSessionRequest added in v0.4.0 

type BackfillSessionRequest struct {
	// SessionID identifies the source session (e.g. UUID from filename).
	SessionID string `json:"session_id"`
	// Project overrides the project path from session metadata. Empty = use parsed value.
	Project string `json:"project"`
	// RunID groups observations from the same backfill run.
	RunID string `json:"run_id"`
	// Content is the raw JSONL session data.
	Content string `json:"content"`
}

BackfillSessionRequest is the request body for POST /api/backfill/session. The server parses the raw JSONL content, extracts observations via LLM, and stores them.

type BackfillSessionResponse added in v0.4.0 

type BackfillSessionResponse struct {
	Stored                int    `json:"stored"`
	Skipped               int    `json:"skipped"`
	Errors                int    `json:"errors"`
	ObservationsExtracted int    `json:"observations_extracted"`
	MetricsReport         string `json:"metrics_report,omitempty"`
}

BackfillSessionResponse is the response for POST /api/backfill/session.

type BackfillStatus added in v0.4.0 

type BackfillStatus struct {
	TotalRuns         int                 `json:"total_runs"`
	ActiveRuns        map[string]*RunInfo `json:"active_runs"`
	TotalObservations int                 `json:"total_observations"`
}

BackfillStatus holds status information for GET /api/backfill/status.

type BulkImportRequest 

type BulkImportRequest struct {
	Project      string                 `json:"project"`
	SessionID    string                 `json:"session_id,omitempty"`
	Observations []BulkObservationInput `json:"observations"`
}

BulkImportRequest is the request body for bulk observation import.

type BulkImportResponse 

type BulkImportResponse struct {
	Errors            []string `json:"errors,omitempty"`
	Imported          int      `json:"imported"`
	Failed            int      `json:"failed"`
	SkippedDuplicates int      `json:"skipped_duplicates,omitempty"`
}

BulkImportResponse contains the result of a bulk import operation.

type BulkObservationInput 

type BulkObservationInput struct {
	Type          string   `json:"type"`
	Title         string   `json:"title"`
	Subtitle      string   `json:"subtitle,omitempty"`
	Narrative     string   `json:"narrative,omitempty"`
	Scope         string   `json:"scope,omitempty"`
	Facts         []string `json:"facts,omitempty"`
	Concepts      []string `json:"concepts,omitempty"`
	FilesRead     []string `json:"files_read,omitempty"`
	FilesModified []string `json:"files_modified,omitempty"`
}

BulkObservationInput represents a single observation in bulk import.

type BulkStatusRequest 

type BulkStatusRequest struct {
	Action   string  `json:"action"`
	Reason   string  `json:"reason,omitempty"`
	IDs      []int64 `json:"ids"`
	Feedback int     `json:"feedback,omitempty"`
}

BulkStatusRequest represents a request to update status for multiple observations.

type CheckSessionsRequest added in v0.4.0 

type CheckSessionsRequest struct {
	SessionIDs []string `json:"session_ids"`
}

CheckSessionsRequest is the request body for checking which sessions are already indexed.

type ComponentHealth 

type ComponentHealth struct {
	Name    string `json:"name"`
	Status  string `json:"status"` // "healthy", "degraded", "unhealthy"
	Message string `json:"message,omitempty"`
}

ComponentHealth represents the health status of a single component.

type ExpensiveOperationLimiter 

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

ExpensiveOperationLimiter provides stricter rate limiting for expensive operations. It wraps the base per-client rate limiter with additional per-operation limits.

func NewExpensiveOperationLimiter 

func NewExpensiveOperationLimiter() *ExpensiveOperationLimiter

NewExpensiveOperationLimiter creates a limiter for expensive operations.

func (*ExpensiveOperationLimiter) CanRebuild 

func (eol *ExpensiveOperationLimiter) CanRebuild() bool

CanRebuild checks if a vector rebuild operation is allowed. Returns false if a rebuild was triggered too recently.

type ExtractLearningsRequest 

type ExtractLearningsRequest struct {
	Messages []learning.Message `json:"messages"`
	Project  string             `json:"project"`
}

ExtractLearningsRequest is the request body for learning extraction.

type FeedbackRequest 

type FeedbackRequest struct {
	Feedback int `json:"feedback"` // -1 (thumbs down), 0 (neutral), 1 (thumbs up)
}

FeedbackRequest represents a user feedback submission.

type IngestRequest 

type IngestRequest struct {
	ToolInput     any    `json:"tool_input"`
	ToolResult    any    `json:"tool_result"`
	SessionID     string `json:"session_id"`
	Project       string `json:"project"`
	ToolName      string `json:"tool_name"`
	WorkstationID string `json:"workstation_id"`
}

IngestRequest is the request body for the event ingest endpoint.

type InjectedObservationResponse added in v0.3.0 

type InjectedObservationResponse struct {
	ID    int64    `json:"id"`
	Title string   `json:"title"`
	Type  string   `json:"type"`
	Facts []string `json:"facts"`
}

InjectedObservationResponse is the response shape for injected observation data.

type InstinctsImportRequest added in v0.9.0 

type InstinctsImportRequest struct {
	Path string `json:"path"`
}

InstinctsImportRequest is the request body for instincts import.

type MergePatternsRequest 

type MergePatternsRequest struct {
	SourceID int64 `json:"source_id"`
	TargetID int64 `json:"target_id"`
}

MergePatternsRequest is the request body for merging patterns.

type ObservationRequest 

type ObservationRequest struct {
	ClaudeSessionID string `json:"claudeSessionId"`
	Project         string `json:"project"`
	ToolName        string `json:"tool_name"`
	ToolInput       any    `json:"tool_input"`
	ToolResponse    any    `json:"tool_response"`
	CWD             string `json:"cwd"`
}

ObservationRequest is the request body for posting observations.

type PatternCleanupRequest added in v1.8.0 

type PatternCleanupRequest struct {
	ConfidenceThreshold float64 `json:"confidence_threshold"`
	DryRun              bool    `json:"dry_run"`
}

PatternCleanupRequest is the JSON body for POST /api/maintenance/patterns/cleanup.

type PatternCleanupResponse added in v1.8.0 

type PatternCleanupResponse struct {
	OrphansFound           int `json:"orphans_found"`
	OrphansArchived        int `json:"orphans_archived"`
	LowConfidenceFound     int `json:"low_confidence_found"`
	LowConfidenceArchived  int `json:"low_confidence_archived"`
	ConfidenceRecalculated int `json:"confidence_recalculated"`
}

PatternCleanupResponse is returned by POST /api/maintenance/patterns/cleanup.

type PatternInsightResponse added in v1.8.0 

type PatternInsightResponse struct {
	Summary            string                `json:"summary"`
	SourceObservations []*models.Observation `json:"source_observations"`
	Cached             bool                  `json:"cached"`
}

PatternInsightResponse is the envelope for POST /api/patterns/{id}/insight.

type PatternObservationsResponse added in v1.8.0 

type PatternObservationsResponse struct {
	Observations []*models.Observation `json:"observations"`
	Total        int                   `json:"total"`
}

PatternObservationsResponse is the envelope for GET /api/patterns/{id}/observations.

type PatternsListResponse added in v1.3.0 

type PatternsListResponse struct {
	Patterns []*models.Pattern `json:"patterns"`
	Total    int64             `json:"total"`
}

PatternsListResponse is the envelope returned by GET /api/patterns.

type PerClientRateLimiter 

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

PerClientRateLimiter implements per-client rate limiting.

func NewPerClientRateLimiter 

func NewPerClientRateLimiter(rate float64, burst int) *PerClientRateLimiter

NewPerClientRateLimiter creates a new per-client rate limiter.

func (*PerClientRateLimiter) Allow 

func (pcrl *PerClientRateLimiter) Allow(clientKey string) bool

Allow checks if a request from the given client should be allowed.

func (*PerClientRateLimiter) Stats 

func (pcrl *PerClientRateLimiter) Stats() map[string]any

Stats returns aggregate statistics. Uses two-phase approach to avoid nested lock acquisition.

type RateLimiter 

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

RateLimiter implements a token bucket rate limiter.

func NewRateLimiter 

func NewRateLimiter(rate float64, burst int) *RateLimiter

NewRateLimiter creates a new rate limiter. rate is the number of requests per second to allow. burst is the maximum burst of requests to allow.

func (*RateLimiter) Allow 

func (rl *RateLimiter) Allow() bool

Allow checks if a request should be allowed. Returns true if the request is allowed, false if rate limited.

func (*RateLimiter) LastUpdateTime 

func (rl *RateLimiter) LastUpdateTime() time.Time

LastUpdateTime returns the last update time. Thread-safe - acquires the limiter's lock.

func (*RateLimiter) Stats 

func (rl *RateLimiter) Stats() map[string]any

Stats returns rate limiter statistics.

type RebuildStatus 

type RebuildStatus struct {
	StartTime    time.Time `json:"start_time,omitempty"`
	Phase        string    `json:"phase,omitempty"`
	TotalSynced  int       `json:"total_synced"`
	TotalErrors  int       `json:"total_errors"`
	CurrentPhase int       `json:"current_phase"`
	TotalPhases  int       `json:"total_phases"`
	ElapsedMs    int64     `json:"elapsed_ms,omitempty"`
	EstimatedPct float64   `json:"estimated_pct,omitempty"`
	InProgress   bool      `json:"in_progress"`
}

RebuildStatus tracks the progress of vector rebuild operations.

type RecentSearchQuery 

type RecentSearchQuery struct {
	Timestamp  time.Time `json:"timestamp"`
	Query      string    `json:"query"`
	Project    string    `json:"project,omitempty"`
	Type       string    `json:"type,omitempty"`
	Results    int       `json:"results"`
	UsedVector bool      `json:"used_vector"`
}

RecentSearchQuery tracks a search query for analytics.

type RetrievalStats 

type RetrievalStats struct {
	TotalRequests      int64 `json:"total_requests"`      // Total retrieval requests (inject + search)
	ObservationsServed int64 `json:"observations_served"` // Observations returned to clients
	VerifiedStale      int64 `json:"verified_stale"`      // Stale observations that passed verification
	DeletedInvalid     int64 `json:"deleted_invalid"`     // Invalid observations deleted
	SearchRequests     int64 `json:"search_requests"`     // Semantic search requests
	ContextInjections  int64 `json:"context_injections"`  // Session-start context injections
	StaleExcluded      int64 `json:"stale_excluded"`      // Observations excluded due to staleness check
	FreshCount         int64 `json:"fresh_count"`         // Observations that passed staleness check
	DuplicatesRemoved  int64 `json:"duplicates_removed"`  // Observations removed by clustering
	LastUpdated        int64 `json:"last_updated"`        // Unix timestamp of last update (atomic)
}

RetrievalStats tracks observation retrieval metrics.

type RunInfo added in v0.4.0 

type RunInfo struct {
	RunID    string `json:"run_id"`
	Stored   int    `json:"stored"`
	Skipped  int    `json:"skipped"`
	Errors   int    `json:"errors"`
	Sessions int    `json:"sessions"`
}

RunInfo tracks per-run statistics.

type SelfCheckResponse 

type SelfCheckResponse struct {
	Overall    string            `json:"overall"` // "healthy", "degraded", "unhealthy"
	Version    string            `json:"version"`
	Uptime     string            `json:"uptime"`
	Components []ComponentHealth `json:"components"`
}

SelfCheckResponse contains the health status of all components.

type Service 

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

Service is the main worker service orchestrator.

func NewService 

func NewService(version string, logBuffer *logbuf.RingBuffer) (*Service, error)

NewService creates a new worker service with deferred initialization. The service starts immediately with health endpoint available, while database and SDK initialization happens in the background.

func (*Service) GetInitError 

func (s *Service) GetInitError() error

GetInitError returns any initialization error.

func (*Service) GetRetrievalStats 

func (s *Service) GetRetrievalStats(project string) RetrievalStats

GetRetrievalStats returns a copy of the retrieval stats for a project. If project is empty, returns aggregate stats across all projects.

func (*Service) Shutdown 

func (s *Service) Shutdown(ctx context.Context) error

Shutdown gracefully shuts down the service.

func (*Service) Start 

func (s *Service) Start() error

Start starts the worker service. The HTTP server starts immediately; database initialization happens async.

type SessionInitRequest 

type SessionInitRequest struct {
	ClaudeSessionID     string `json:"claudeSessionId"`
	Project             string `json:"project"`
	Prompt              string `json:"prompt"`
	MatchedObservations int    `json:"matchedObservations"`
}

SessionInitRequest is the request body for session initialization.

type SessionInitResponse 

type SessionInitResponse struct {
	Reason       string `json:"reason,omitempty"`
	SessionDBID  int64  `json:"sessionDbId"`
	PromptNumber int    `json:"promptNumber"`
	Skipped      bool   `json:"skipped,omitempty"`
}

SessionInitResponse is the response for session initialization.

type SessionStartRequest 

type SessionStartRequest struct {
	UserPrompt   string `json:"userPrompt"`
	PromptNumber int    `json:"promptNumber"`
}

SessionStartRequest is the request body for starting SDK agent.

type SubagentCompleteRequest 

type SubagentCompleteRequest struct {
	ClaudeSessionID string `json:"claudeSessionId"`
	Project         string `json:"project"`
}

SubagentCompleteRequest is the request body for subagent completion.

type SummarizeRequest 

type SummarizeRequest struct {
	LastUserMessage      string `json:"lastUserMessage"`
	LastAssistantMessage string `json:"lastAssistantMessage"`
}

SummarizeRequest is the request body for summarize requests.

type TokenAuth 

type TokenAuth struct {
	ExemptPaths map[string]bool
	// contains filtered or unexported fields
}

TokenAuth provides token-based authentication for the worker HTTP API. Supports three auth methods:

  1. Master token (ENGRAM_API_TOKEN env var) via X-Auth-Token or Authorization: Bearer header -> admin
  2. Client API tokens (engram_* prefix, bcrypt-hashed in DB) via same headers -> scoped access
  3. HMAC-signed session cookie (engram_session) -> admin (dashboard)

func NewTokenAuth 

func NewTokenAuth(token string) (*TokenAuth, error)

NewTokenAuth creates a new TokenAuth using a provided token. If token is empty and ENGRAM_AUTH_DISABLED is set, authentication is skipped. Otherwise, authentication will be enforced at startup (see Service.Start).

func (*TokenAuth) CookieKey added in v1.0.0 

func (ta *TokenAuth) CookieKey() []byte

CookieKey returns the HMAC key used for signing session cookies.

func (*TokenAuth) IsEnabled 

func (ta *TokenAuth) IsEnabled() bool

IsEnabled returns whether token authentication is enabled.

func (*TokenAuth) Middleware 

func (ta *TokenAuth) Middleware(next http.Handler) http.Handler

Middleware returns HTTP middleware that enforces token authentication. Auth priority: header token (master or client) > session cookie > 401.

func (*TokenAuth) SetTokenStore added in v1.0.0 

func (ta *TokenAuth) SetTokenStore(store *gormdb.TokenStore)

SetTokenStore sets the token store for client token lookups. Called after DB initialization completes.

func (*TokenAuth) StatsCh added in v1.0.0 

func (ta *TokenAuth) StatsCh() chan string

StatsCh returns the buffered channel for async token stats increment.

func (*TokenAuth) Token 

func (ta *TokenAuth) Token() string

Token returns the authentication token. Returns empty string if authentication is disabled.

type UpdateObservationRequest 

type UpdateObservationRequest struct {
	Title         *string  `json:"title,omitempty"`
	Subtitle      *string  `json:"subtitle,omitempty"`
	Narrative     *string  `json:"narrative,omitempty"`
	Scope         *string  `json:"scope,omitempty"`
	Facts         []string `json:"facts,omitempty"`
	Concepts      []string `json:"concepts,omitempty"`
	FilesRead     []string `json:"files_read,omitempty"`
	FilesModified []string `json:"files_modified,omitempty"`
}

UpdateObservationRequest is the request body for updating an observation.

type UtilityRequest 

type UtilityRequest struct {
	Signal string `json:"signal"` // "used", "corrected", "ignored"
}

UtilityRequest represents a utility signal for an observation.

Source Files

View all Source files

Directories

Path Synopsis
Package sdk provides SDK agent integration for engram.
 Package sdk provides SDK agent integration for engram.
Package session provides session lifecycle management for engram.
 Package session provides session lifecycle management for engram.
Package sse provides Server-Sent Events broadcasting for engram.
 Package sse provides Server-Sent Events broadcasting for engram.

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.